You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/06/15 13:41:34 UTC
[GitHub] [airflow] ishiis opened a new pull request, #24473: Add support for Salesforce bulk api
ishiis opened a new pull request, #24473:
URL: https://github.com/apache/airflow/pull/24473
Add SalesforceBulkInsertOperator, SalesforceBulkUpdateOperator, SalesforceBulkUpsertOperator and SalesforceBulkDeleteOperator.
These operators allow to access Bulk API functions.
related: #16921
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragement file, named `{pr_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r901671843
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
@mariotaddeucci
I moved the validation to else clause. This is better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r899082162
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,245 @@
+# 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
+
+from airflow.models import BaseOperator
+from airflow.providers.salesforce.hooks.salesforce import SalesforceHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class SalesforceBulkInsertOperator(BaseOperator):
Review Comment:
@eladkal
Thank you for feedback.
I think so too. These operators are very similar. Instead of creating these operators, make them `SalesforceBulkOperator`, add `operation` parameter to `SalesforceBulkOperator`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r902649279
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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']
+ :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: str,
+ object_name: str,
+ 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 AirflowException("The required parameter 'object_name' is missing.")
+
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f"Operation not found! Available operations are f{available_operations}.")
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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
+ )
+
+ if self.do_xcom_push:
+ return result
+ else:
+ return {}
Review Comment:
~~This empty dict unneeded. I removed this.~~
I could not remove this. So I changed the way I wrote return statement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mariotaddeucci commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
mariotaddeucci commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r901821551
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
Both are good, a validation on constructor will help to find errors when the dag is generated, not on execution time, becoming easier to get errors on unit tests for example
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal merged pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal merged PR #24473:
URL: https://github.com/apache/airflow/pull/24473
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r903615350
##########
docs/apache-airflow-providers-salesforce/operators/salesforce_bulk.rst:
##########
@@ -0,0 +1,55 @@
+ .. 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.salesforce_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_salesforce_bulk.py
+ :language: python
Review Comment:
Added `:dedent:` param.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on PR #24473:
URL: https://github.com/apache/airflow/pull/24473#issuecomment-1173701770
@ishiis can you fix conflicts and rebase? I will merge after
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r913132147
##########
airflow/providers/salesforce/operators/bulk.py:
##########
@@ -0,0 +1,113 @@
+# 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') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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 not self.do_xcom_push:
+ result = None
+
+ return result
Review Comment:
@eladkal
I referred to [this code](https://github.com/apache/airflow/blob/main/airflow/providers/salesforce/operators/salesforce_apex_rest.py#L55-L69).
I removed the return type because None cannot be returned if return type is specified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r901671843
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
@mariotaddeucci
~~I moved the validation to else clause. This is better.~~
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
@mariotaddeucci
I moved the validation to constructor and add validation for object_name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r905056648
##########
tests/providers/salesforce/operators/test_salesforce_bulk.py:
##########
@@ -0,0 +1,221 @@
+# 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.
+
+import unittest
+from unittest.mock import Mock, patch
+
+import pytest
+
+from airflow.exceptions import AirflowException
+from airflow.providers.salesforce.operators.salesforce_bulk import SalesforceBulkOperator
+
+
+class TestSalesforceBulkOperator(unittest.TestCase):
Review Comment:
Changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on PR #24473:
URL: https://github.com/apache/airflow/pull/24473#issuecomment-1168608981
@mariotaddeucci @eladkal @josh-fell
What do you think my changes? Feel free to comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r913061206
##########
airflow/providers/salesforce/operators/bulk.py:
##########
@@ -0,0 +1,113 @@
+# 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') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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 not self.do_xcom_push:
+ result = None
+
+ return result
Review Comment:
```suggestion
if self.do_xcom_push and result:
return result
return None
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r898985348
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,245 @@
+# 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
+
+from airflow.models import BaseOperator
+from airflow.providers.salesforce.hooks.salesforce import SalesforceHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class SalesforceBulkInsertOperator(BaseOperator):
Review Comment:
I think we may need only 1 operator for all actions.
The bulk api functions are very similar
Insert:
https://github.com/simple-salesforce/simple-salesforce/blob/cfd191c547256ebca176ef35df8bc0f3770d7030/simple_salesforce/bulk.py#L272
Upsert:
https://github.com/simple-salesforce/simple-salesforce/blob/master/simple_salesforce/bulk.py#L280
Update:
https://github.com/simple-salesforce/simple-salesforce/blob/cfd191c547256ebca176ef35df8bc0f3770d7030/simple_salesforce/bulk.py#L289
etc...
Instead of creating `SalesforceBulkInsertOperator`, `SalesforceBulkUpdateOperator` , `SalesforceBulkUpsertOperator`, etc..
We can create a unified operator that accept method where method can be `update`, `insert`, `upsert` etc and `method_kwargs` where contains all parameters needed for the method. This way it also automatically supports any new parameter that will introduced by the API.
In some way it's similar to what we do with method `parameter` of `TableauOperator`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r902550181
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
I think we should just call it `bulk.py` ?
@josh-fell WDYT?
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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']
+ :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: str,
+ object_name: str,
+ 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 AirflowException("The required parameter 'object_name' is missing.")
+
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f"Operation not found! Available operations are f{available_operations}.")
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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
+ )
+
+ if self.do_xcom_push:
+ return result
+ else:
+ return {}
Review Comment:
Why do we need to return empty dict?
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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']
Review Comment:
lets also add `hard_delete`
https://github.com/simple-salesforce/simple-salesforce/blob/cfd191c547256ebca176ef35df8bc0f3770d7030/simple_salesforce/bulk.py#L296
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r913148471
##########
airflow/providers/salesforce/operators/bulk.py:
##########
@@ -0,0 +1,113 @@
+# 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') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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 not self.do_xcom_push:
+ result = None
+
+ return result
Review Comment:
The API response type was wrong. Fixed type of result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r901147763
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
Check line 74
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r903614492
##########
airflow/providers/salesforce/operators/salesforce_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
+
+from airflow.exceptions import AirflowException
+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: str,
+ object_name: str,
+ 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 AirflowException("The required parameter 'object_name' is missing.")
Review Comment:
Changed AirflowException to ValueError
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r901147922
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
Would be good to have a test case for this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r898985348
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,245 @@
+# 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
+
+from airflow.models import BaseOperator
+from airflow.providers.salesforce.hooks.salesforce import SalesforceHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class SalesforceBulkInsertOperator(BaseOperator):
Review Comment:
I think we may need only 1 operator for all actions.
The bulk api functions are very similar
Insert:
https://github.com/simple-salesforce/simple-salesforce/blob/cfd191c547256ebca176ef35df8bc0f3770d7030/simple_salesforce/bulk.py#L272
Upsert:
https://github.com/simple-salesforce/simple-salesforce/blob/master/simple_salesforce/bulk.py#L280
Update:
https://github.com/simple-salesforce/simple-salesforce/blob/cfd191c547256ebca176ef35df8bc0f3770d7030/simple_salesforce/bulk.py#L289
etc...
Instead of creating `SalesforceBulkInsertOperator`, `SalesforceBulkUpdateOperator` , `SalesforceBulkUpsertOperator`, etc..
We can create a unified operator that accept method where method can be `update`, `insert`, `upsert` etc and `method_kwargs` where contains all parameters needed for the method. This way it also automatically supports any new parameter that will introduced by the API.
In some way it's similar to what we do with method `parameter` of `TableauOperator`
WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r902649279
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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']
+ :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: str,
+ object_name: str,
+ 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 AirflowException("The required parameter 'object_name' is missing.")
+
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f"Operation not found! Available operations are f{available_operations}.")
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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
+ )
+
+ if self.do_xcom_push:
+ return result
+ else:
+ return {}
Review Comment:
This empty dict unneeded. I removed this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r903615933
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Renamed `salesforce_apex_rest.py` to `bulk.py`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r902645880
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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']
Review Comment:
added a hard_delete operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r901630418
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
There is already a test case for this.
https://github.com/apache/airflow/pull/24473/files#diff-4db0aba80d593c6846d4ab4b91327f59986e7b7e304aa0cc0e6df9b85f76b0ebR32-R44
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
josh-fell commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r902739430
##########
airflow/providers/salesforce/operators/salesforce_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
+
+from airflow.exceptions import AirflowException
+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: str,
+ object_name: str,
+ 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 AirflowException("The required parameter 'object_name' is missing.")
+
+ available_operations = ['insert', 'update', 'upsert', 'delete', 'hard_delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f"Operation not found! Available operations are f{available_operations}.")
Review Comment:
Same comment here too also looks like some leftovers from a previous f-string.
```suggestion
raise ValueError(f"Operation not found! Available operations are {available_operations}.")
```
##########
docs/apache-airflow-providers-salesforce/operators/salesforce_bulk.rst:
##########
@@ -0,0 +1,55 @@
+ .. 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.salesforce_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_salesforce_bulk.py
+ :language: python
+ :start-after: [START howto_salesforce_bulk_insert_operator]
+ :end-before: [END howto_salesforce_bulk_insert_operator]
+
+You can use this operator to access Bulk Update API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_salesforce_bulk.py
+ :language: python
+ :start-after: [START howto_salesforce_bulk_update_operator]
+ :end-before: [END howto_salesforce_bulk_update_operator]
+
+You can use this operator to access Bulk Upsert API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_salesforce_bulk.py
+ :language: python
Review Comment:
```suggestion
:language: python
:dedent: 4
```
##########
docs/apache-airflow-providers-salesforce/operators/salesforce_bulk.rst:
##########
@@ -0,0 +1,55 @@
+ .. 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.salesforce_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_salesforce_bulk.py
+ :language: python
+ :start-after: [START howto_salesforce_bulk_insert_operator]
+ :end-before: [END howto_salesforce_bulk_insert_operator]
+
+You can use this operator to access Bulk Update API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_salesforce_bulk.py
+ :language: python
Review Comment:
```suggestion
:language: python
:dedent: 4
```
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
I agree. Slightly unfortunate we already have `salesforce_apex_rest.py`. The naming is a little redundant and I don't think we need to continue to pattern.
##########
tests/providers/salesforce/operators/test_salesforce_bulk.py:
##########
@@ -0,0 +1,221 @@
+# 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.
+
+import unittest
+from unittest.mock import Mock, patch
+
+import pytest
+
+from airflow.exceptions import AirflowException
+from airflow.providers.salesforce.operators.salesforce_bulk import SalesforceBulkOperator
+
+
+class TestSalesforceBulkOperator(unittest.TestCase):
Review Comment:
For net-new unit tests, `pytest` is preferable over `unittest`.
##########
docs/apache-airflow-providers-salesforce/operators/salesforce_bulk.rst:
##########
@@ -0,0 +1,55 @@
+ .. 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.salesforce_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_salesforce_bulk.py
+ :language: python
Review Comment:
```suggestion
:language: python
:dedent: 4
```
Since the `START` marker is indented we should use the `:dedent:` param for rendering the code snippet appropriately.
##########
airflow/providers/salesforce/operators/salesforce_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
+
+from airflow.exceptions import AirflowException
+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: str,
+ object_name: str,
+ 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 AirflowException("The required parameter 'object_name' is missing.")
Review Comment:
```suggestion
raise ValueError("The required parameter 'object_name' is missing.")
```
I think `AirflowException` is used _too_ broadly throughout the code base. `ValueError` seems like a better choice.
##########
airflow/providers/salesforce/operators/salesforce_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
+
+from airflow.exceptions import AirflowException
+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):
Review Comment:
WDYT about adding some `template_fields` for the operator? Not required, but just thinking about developer use cases for dynamic/templatized values, pulling values from Airflow Variables, etc.
##########
docs/apache-airflow-providers-salesforce/operators/salesforce_bulk.rst:
##########
@@ -0,0 +1,55 @@
+ .. 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.salesforce_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_salesforce_bulk.py
+ :language: python
+ :start-after: [START howto_salesforce_bulk_insert_operator]
+ :end-before: [END howto_salesforce_bulk_insert_operator]
+
+You can use this operator to access Bulk Update API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_salesforce_bulk.py
+ :language: python
+ :start-after: [START howto_salesforce_bulk_update_operator]
+ :end-before: [END howto_salesforce_bulk_update_operator]
+
+You can use this operator to access Bulk Upsert API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_salesforce_bulk.py
+ :language: python
+ :start-after: [START howto_salesforce_bulk_upsert_operator]
+ :end-before: [END howto_salesforce_bulk_upsert_operator]
+
+You can use this operator to access Bulk Delete API:
+
+.. exampleinclude:: /../../tests/system/providers/salesforce/example_salesforce_bulk.py
+ :language: python
Review Comment:
```suggestion
:language: python
:dedent: 4
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mariotaddeucci commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
mariotaddeucci commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r901146227
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,103 @@
+# 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
+
+from airflow.exceptions import AirflowException
+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
+ :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: str,
+ object_name: str,
+ 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
+
+ def execute(self, context: 'Context') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ available_operations = ['insert', 'update', 'upsert', 'delete']
+ if self.operation not in available_operations:
+ raise AirflowException(f'Operation not found! Available operations are {available_operations}')
+
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ 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,
+ )
+ else:
+ result = conn.bulk.__getattr__(self.object_name).delete(
+ data=self.payload, batch_size=self.batch_size, use_serial=self.use_serial
+ )
+
Review Comment:
Hey @ishiis, a simple suggestion. Add a validation on operators constructor about what operation will execute and don't execute the delete on `else`, instead of this, raise an exception.
If I what to update and make a mistake writing, `updat`, this operator will delete the passed ids.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r913038315
##########
airflow/providers/salesforce/operators/bulk.py:
##########
@@ -0,0 +1,113 @@
+# 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') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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 not self.do_xcom_push:
+ result = {}
Review Comment:
One last comment...
Didn't we say we don't want to push empty dict?
https://github.com/apache/airflow/pull/24473#discussion_r902548285
We should return None if nothing to push
see for example
https://github.com/apache/airflow/blob/ec6761a5c0d031221d53ce213c0e42813606c55d/airflow/providers/amazon/aws/operators/ecs.py#L312-L315
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on PR #24473:
URL: https://github.com/apache/airflow/pull/24473#issuecomment-1173751277
@eladkal I fixed the conflicts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r913050580
##########
airflow/providers/salesforce/operators/bulk.py:
##########
@@ -0,0 +1,113 @@
+# 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') -> dict:
+ """
+ Makes an HTTP request to Salesforce Bulk API.
+
+ :param context: The task context during execution.
+ :return: API response if do_xcom_push is True
+ :rtype: dict
+ """
+ sf_hook = SalesforceHook(salesforce_conn_id=self.salesforce_conn_id)
+ conn = sf_hook.get_conn()
+
+ result: dict = {}
+ 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 not self.do_xcom_push:
+ result = {}
Review Comment:
fixed this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r899082162
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,245 @@
+# 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
+
+from airflow.models import BaseOperator
+from airflow.providers.salesforce.hooks.salesforce import SalesforceHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class SalesforceBulkInsertOperator(BaseOperator):
Review Comment:
@eladkal
Thank you for feedback.
I think so too. These operators are very similar. Instead of creating these operators, make them `SalesforceBulkOperator`. Then add `operation` parameter to `SalesforceBulkOperator`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r898985348
##########
airflow/providers/salesforce/operators/salesforce_bulk.py:
##########
@@ -0,0 +1,245 @@
+# 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
+
+from airflow.models import BaseOperator
+from airflow.providers.salesforce.hooks.salesforce import SalesforceHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class SalesforceBulkInsertOperator(BaseOperator):
Review Comment:
I think we may need only 1 operator for all actions.
The bulk api functions are very similar
Insert:
https://github.com/simple-salesforce/simple-salesforce/blob/cfd191c547256ebca176ef35df8bc0f3770d7030/simple_salesforce/bulk.py#L272
Upsert:
https://github.com/simple-salesforce/simple-salesforce/blob/master/simple_salesforce/bulk.py#L280
Update:
https://github.com/simple-salesforce/simple-salesforce/blob/cfd191c547256ebca176ef35df8bc0f3770d7030/simple_salesforce/bulk.py#L289
etc...
Instead of creating `SalesforceBulkInsertOperator`, `SalesforceBulkUpdateOperator` , `SalesforceBulkUpsertOperator`, etc..
We can create a unified operator that accept method where method can be `update`, `insert`, `upsert` etc and `method_kwargs` where contains all parameters needed for the method. This way it also automatically supports any new parameter that will introduced by the API.
In some way it's similar to what we do with `method` parameter of `TableauOperator`
WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ishiis commented on a diff in pull request #24473: Add support for Salesforce bulk api
Posted by GitBox <gi...@apache.org>.
ishiis commented on code in PR #24473:
URL: https://github.com/apache/airflow/pull/24473#discussion_r903611750
##########
airflow/providers/salesforce/operators/salesforce_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
+
+from airflow.exceptions import AirflowException
+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):
Review Comment:
I don't think there is anything to add to template_fields. I can't think of any good use cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org