[airflow] branch main updated: Correct parameter typing in `SalesforceBulkOperator` (#24927)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 3f9414d0fd Correct parameter typing in `SalesforceBulkOperator` (#24927)
3f9414d0fd is described below

commit 3f9414d0fd67a2c508e105651de1105955f307bf
Author: Josh Fell <48...@users.noreply.github.com>
AuthorDate: Fri Jul 8 17:41:22 2022 -0400

    Correct parameter typing in `SalesforceBulkOperator` (#24927)
---
 airflow/providers/salesforce/operators/bulk.py    | 19 ++++++++++++-------
 tests/providers/salesforce/operators/test_bulk.py | 16 ++++++++++++++++
 2 files changed, 28 insertions(+), 7 deletions(-)

diff --git a/airflow/providers/salesforce/operators/bulk.py b/airflow/providers/salesforce/operators/bulk.py
index fc0ef154ee..110ed685ea 100644
--- a/airflow/providers/salesforce/operators/bulk.py
+++ b/airflow/providers/salesforce/operators/bulk.py
@@ -14,10 +14,11 @@
 # 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 typing import TYPE_CHECKING
 
 from airflow.models import BaseOperator
 from airflow.providers.salesforce.hooks.salesforce import SalesforceHook
+from airflow.typing_compat import Literal
 
 if TYPE_CHECKING:
     from airflow.utils.context import Context
@@ -41,11 +42,13 @@ class SalesforceBulkOperator(BaseOperator):
     :param salesforce_conn_id: The :ref:`Salesforce Connection id <howto/connection:SalesforceHook>`.
     """
 
+    available_operations = ('insert', 'update', 'upsert', 'delete', 'hard_delete')
+
     def __init__(
         self,
         *,
-        operation: Optional[str] = None,
-        object_name: Optional[str] = None,
+        operation: Literal[available_operations],
+        object_name: str,
         payload: list,
         external_id_field: str = 'Id',
         batch_size: int = 10000,
@@ -65,11 +68,13 @@ class SalesforceBulkOperator(BaseOperator):
 
     def _validate_inputs(self) -> None:
         if not self.object_name:
-            raise ValueError("The required parameter 'object_name' is missing.")
+            raise ValueError("The required parameter 'object_name' cannot have an empty value.")
 
-        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}.")
+        if self.operation not in self.available_operations:
+            raise ValueError(
+                f"Operation {self.operation!r} not found! "
+                f"Available operations are {self.available_operations}."
+            )
 
     def execute(self, context: 'Context'):
         """
diff --git a/tests/providers/salesforce/operators/test_bulk.py b/tests/providers/salesforce/operators/test_bulk.py
index 0a580582d7..b60b3da7e4 100644
--- a/tests/providers/salesforce/operators/test_bulk.py
+++ b/tests/providers/salesforce/operators/test_bulk.py
@@ -19,6 +19,7 @@ from unittest.mock import Mock, patch
 
 import pytest
 
+from airflow.exceptions import AirflowException
 from airflow.providers.salesforce.operators.bulk import SalesforceBulkOperator
 
 
@@ -31,6 +32,13 @@ class TestSalesforceBulkOperator:
         """
         Test execute missing operation
         """
+        with pytest.raises(AirflowException):
+            SalesforceBulkOperator(
+                task_id='no_missing_operation_arg',
+                object_name='Account',
+                payload=[],
+            )
+
         with pytest.raises(ValueError):
             SalesforceBulkOperator(
                 task_id='missing_operation',
@@ -43,10 +51,18 @@ class TestSalesforceBulkOperator:
         """
         Test execute missing object_name
         """
+        with pytest.raises(AirflowException):
+            SalesforceBulkOperator(
+                task_id='no_object_name_arg',
+                operation='insert',
+                payload=[],
+            )
+
         with pytest.raises(ValueError):
             SalesforceBulkOperator(
                 task_id='missing_object_name',
                 operation='insert',
+                object_name="",
                 payload=[],
             )