Add support for Salesforce bulk api (apache#24473)

* add support for Salesforce bulk api

airflow/providers/salesforce/operators/bulk.py

@@ -0,0 +1,112 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import TYPE_CHECKING, Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.salesforce.hooks.salesforce import SalesforceHook
+
+if TYPE_CHECKING:
+    from airflow.utils.context import Context
+
+
+class SalesforceBulkOperator(BaseOperator):
+    """
+    Execute a Salesforce Bulk API and pushes results to xcom.
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:SalesforceBulkOperator`
+
+    :param operation: Bulk operation to be performed
+        Available operations are in ['insert', 'update', 'upsert', 'delete', 'hard_delete']
+    :param object_name: The name of the Salesforce object
+    :param payload: list of dict to be passed as a batch
+    :param external_id_field: unique identifier field for upsert operations
+    :param batch_size: number of records to assign for each batch in the job
+    :param use_serial: Process batches in serial mode
+    :param salesforce_conn_id: The :ref:`Salesforce Connection id <howto/connection:SalesforceHook>`.
+    """
+
+    def __init__(
+        self,
+        *,
+        operation: Optional[str] = None,
+        object_name: Optional[str] = None,
+        payload: list,
+        external_id_field: str = 'Id',
+        batch_size: int = 10000,
+        use_serial: bool = False,
+        salesforce_conn_id: str = 'salesforce_default',
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.operation = operation
+        self.object_name = object_name
+        self.payload = payload
+        self.external_id_field = external_id_field
+        self.batch_size = batch_size
+        self.use_serial = use_serial
+        self.salesforce_conn_id = salesforce_conn_id
+        self._validate_inputs()
+
+    def _validate_inputs(self) -> None:
+        if not self.object_name:
+            raise ValueError("The required parameter 'object_name' is missing.")
+
+        available_operations = ['insert', 'update', 'upsert', 'delete', 'hard_delete']
+        if self.operation not in available_operations:
+            raise ValueError(f"Operation not found! Available operations are {available_operations}.")
+
+    def execute(self, context: 'Context'):
+        """
+        Makes an HTTP request to Salesforce Bulk API.
+
+        :param context: The task context during execution.
+        :return: API response if do_xcom_push is True
+        """
+        sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+        conn = sf_hook.get_conn()
+
+        result = []
+        if self.operation == 'insert':
+            result = conn.bulk.__getattr__(self.object_name).insert(
+                data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+            )
+        elif self.operation == 'update':
+            result = conn.bulk.__getattr__(self.object_name).update(
+                data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+            )
+        elif self.operation == 'upsert':
+            result = conn.bulk.__getattr__(self.object_name).upsert(
+                data=self.payload,
+                external_id_field=self.external_id_field,
+                batch_size=self.batch_size,
+                use_serial=self.use_serial,
+            )
+        elif self.operation == 'delete':
+            result = conn.bulk.__getattr__(self.object_name).delete(
+                data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+            )
+        elif self.operation == 'hard_delete':
+            result = conn.bulk.__getattr__(self.object_name).hard_delete(
+                data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+            )
+
+        if self.do_xcom_push and result:
+            return result
+
+        return None

airflow/providers/salesforce/provider.yaml

@@ -46,13 +46,15 @@ integrations:
     external-doc-url: https://www.salesforce.com/
     how-to-guide:
       - /docs/apache-airflow-providers-salesforce/operators/salesforce_apex_rest.rst
+      - /docs/apache-airflow-providers-salesforce/operators/bulk.rst
     logo: /integration-logos/salesforce/Salesforce.png
     tags: [service]
 
 operators:
   - integration-name: Salesforce
     python-modules:
       - airflow.providers.salesforce.operators.salesforce_apex_rest
+      - airflow.providers.salesforce.operators.bulk
 
 hooks:
   - integration-name: Salesforce

docs/apache-airflow-providers-salesforce/operators/bulk.rst

@@ -0,0 +1,59 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _howto/operator:SalesforceBulkOperator:
+
+
+SalesforceBulkOperator
+======================
+
+Use the :class:`~airflow.providers.salesforce.operators.bulk.SalesforceBulkOperator` to execute Bulk API.
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+You can use this operator to access Bulk Insert API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_bulk.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_salesforce_bulk_insert_operation]
+    :end-before: [END howto_salesforce_bulk_insert_operation]
+
+You can use this operator to access Bulk Update API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_bulk.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_salesforce_bulk_update_operation]
+    :end-before: [END howto_salesforce_bulk_update_operation]
+
+You can use this operator to access Bulk Upsert API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_bulk.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_salesforce_bulk_upsert_operation]
+    :end-before: [END howto_salesforce_bulk_upsert_operation]
+
+You can use this operator to access Bulk Delete API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_bulk.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_salesforce_bulk_delete_operation]
+    :end-before: [END howto_salesforce_bulk_delete_operation]

docs/apache-airflow-providers-salesforce/operators/index.rst

@@ -24,3 +24,4 @@ Salesforce Operators
     :maxdepth: 1
 
     salesforce_apex_rest
+    bulk