Rename DataSync Hook and Operator (#20328)

Author: ferruzzi

airflow/providers/amazon/aws/example_dags/example_datasync_1.py

@@ -30,7 +30,7 @@
 from os import getenv
 
 from airflow import models
-from airflow.providers.amazon.aws.operators.datasync import AWSDataSyncOperator
+from airflow.providers.amazon.aws.operators.datasync import DataSyncOperator
 
 # [START howto_operator_datasync_1_args_1]
 TASK_ARN = getenv("TASK_ARN", "my_aws_datasync_task_arn")
@@ -52,7 +52,7 @@
 ) as dag:
 
     # [START howto_operator_datasync_1_1]
-    datasync_task_1 = AWSDataSyncOperator(task_id="datasync_task_1", task_arn=TASK_ARN)
+    datasync_task_1 = DataSyncOperator(task_id="datasync_task_1", task_arn=TASK_ARN)
     # [END howto_operator_datasync_1_1]
 
 with models.DAG(
@@ -61,7 +61,7 @@
     schedule_interval=None,  # Override to match your needs
 ) as dag:
     # [START howto_operator_datasync_1_2]
-    datasync_task_2 = AWSDataSyncOperator(
+    datasync_task_2 = DataSyncOperator(
         task_id="datasync_task_2",
         source_location_uri=SOURCE_LOCATION_URI,
         destination_location_uri=DESTINATION_LOCATION_URI,

airflow/providers/amazon/aws/example_dags/example_datasync_2.py

@@ -39,7 +39,7 @@
 from os import getenv
 
 from airflow import models
-from airflow.providers.amazon.aws.operators.datasync import AWSDataSyncOperator
+from airflow.providers.amazon.aws.operators.datasync import DataSyncOperator
 
 # [START howto_operator_datasync_2_args]
 SOURCE_LOCATION_URI = getenv("SOURCE_LOCATION_URI", "smb://hostname/directory/")
@@ -78,7 +78,7 @@
 ) as dag:
 
     # [START howto_operator_datasync_2]
-    datasync_task = AWSDataSyncOperator(
+    datasync_task = DataSyncOperator(
         task_id="datasync_task",
         source_location_uri=SOURCE_LOCATION_URI,
         destination_location_uri=DESTINATION_LOCATION_URI,

airflow/providers/amazon/aws/hooks/datasync.py

@@ -18,13 +18,14 @@
 """Interact with AWS DataSync, using the AWS ``boto3`` library."""
 
 import time
+import warnings
 from typing import List, Optional
 
 from airflow.exceptions import AirflowBadRequest, AirflowException, AirflowTaskTimeout
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 
 
-class AWSDataSyncHook(AwsBaseHook):
+class DataSyncHook(AwsBaseHook):
     """
     Interact with AWS DataSync.
 
@@ -33,7 +34,7 @@ class AWSDataSyncHook(AwsBaseHook):
 
     .. seealso::
         :class:`~airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook`
-        :class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator`
+        :class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator`
 
     :param wait_interval_seconds: Time to wait between two
         consecutive calls to check TaskExecution status. Defaults to 30 seconds.
@@ -316,3 +317,18 @@ def wait_for_task_execution(self, task_execution_arn: str, max_iterations: int =
         if iterations <= 0:
             raise AirflowTaskTimeout("Max iterations exceeded!")
         raise AirflowException(f"Unknown status: {status}")  # Should never happen
+
+
+class AWSDataSyncHook(DataSyncHook):
+    """
+    This hook is deprecated.
+    Please use :class:`airflow.providers.amazon.aws.hooks.datasync.DataSyncHook`.
+    """
+
+    def __init__(self, *args, **kwargs):
+        warnings.warn(
+            "This hook is deprecated. Please use `airflow.providers.amazon.aws.hooks.datasync.DataSyncHook`.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(*args, **kwargs)

airflow/providers/amazon/aws/operators/datasync.py

@@ -19,22 +19,23 @@
 
 import logging
 import random
+import warnings
 from typing import List, Optional
 
 from airflow.exceptions import AirflowException, AirflowTaskTimeout
 from airflow.models import BaseOperator
-from airflow.providers.amazon.aws.hooks.datasync import AWSDataSyncHook
+from airflow.providers.amazon.aws.hooks.datasync import DataSyncHook
 
 
-class AWSDataSyncOperator(BaseOperator):
+class DataSyncOperator(BaseOperator):
     r"""Find, Create, Update, Execute and Delete AWS DataSync Tasks.
 
     If ``do_xcom_push`` is True, then the DataSync TaskArn and TaskExecutionArn
     which were executed will be pushed to an XCom.
 
     .. seealso::
         For more information on how to use this operator, take a look at the guide:
-        :ref:`howto/operator:AWSDataSyncOperator`
+        :ref:`howto/operator:DataSyncOperator`
 
     .. note:: There may be 0, 1, or many existing DataSync Tasks defined in your AWS
         environment. The default behavior is to create a new Task if there are 0, or
@@ -185,7 +186,7 @@ def __init__(
             )
 
         # Others
-        self.hook: Optional[AWSDataSyncHook] = None
+        self.hook: Optional[DataSyncHook] = None
         # Candidates - these are found in AWS as possible things
         # for us to use
         self.candidate_source_location_arns: Optional[List[str]] = None
@@ -196,15 +197,15 @@ def __init__(
         self.destination_location_arn: Optional[str] = None
         self.task_execution_arn: Optional[str] = None
 
-    def get_hook(self) -> AWSDataSyncHook:
-        """Create and return AWSDataSyncHook.
+    def get_hook(self) -> DataSyncHook:
+        """Create and return DataSyncHook.
 
-        :return AWSDataSyncHook: An AWSDataSyncHook instance.
+        :return DataSyncHook: An DataSyncHook instance.
         """
         if self.hook:
             return self.hook
 
-        self.hook = AWSDataSyncHook(
+        self.hook = DataSyncHook(
             aws_conn_id=self.aws_conn_id,
             wait_interval_seconds=self.wait_interval_seconds,
         )
@@ -345,7 +346,7 @@ def _update_datasync_task(self) -> None:
         self.log.info("Updated TaskArn %s", self.task_arn)
 
     def _execute_datasync_task(self) -> None:
-        """Create and monitor an AWSDataSync TaskExecution for a Task."""
+        """Create and monitor an AWS DataSync TaskExecution for a Task."""
         if not self.task_arn:
             raise AirflowException("Missing TaskArn")
 
@@ -407,3 +408,19 @@ def _get_location_arns(self, location_uri) -> List[str]:
         location_arns = self.get_hook().get_location_arns(location_uri)
         self.log.info("Found LocationArns %s for LocationUri %s", location_arns, location_uri)
         return location_arns
+
+
+class AWSDataSyncOperator(DataSyncOperator):
+    """
+    This operator is deprecated.
+    Please use :class:`airflow.providers.amazon.aws.operators.datasync.DataSyncOperator`.
+    """
+
+    def __init__(self, *args, **kwargs):
+        warnings.warn(
+            "This operator is deprecated. Please use "
+            "`airflow.providers.amazon.aws.operators.datasync.DataSyncHook`.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(*args, **kwargs)

docs/apache-airflow-providers-amazon/operators/datasync.rst

@@ -16,7 +16,7 @@
     under the License.
 
 
-.. _howto/operator:AWSDataSyncOperator:
+.. _howto/operator:DataSyncOperator:
 
 AWS DataSync Operator
 =====================
@@ -25,13 +25,13 @@ Overview
 --------
 
 Two example_dags are provided which showcase the
-:class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator`
+:class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator`
 in action.
 
  - example_datasync_1.py
  - example_datasync_2.py
 
-Both examples use the :class:`~airflow.providers.amazon.aws.hooks.datasync.AWSDataSyncHook`
+Both examples use the :class:`~airflow.providers.amazon.aws.hooks.datasync.DataSyncHook`
 to create a boto3 DataSync client. This hook in turn uses the :class:`~airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook`
 
 Note this guide differentiates between an *Airflow task* (identified by a task_id on Airflow),
@@ -65,7 +65,7 @@ These examples rely on the following variables, which can be passed via OS envir
 Get DataSync Tasks
 """"""""""""""""""
 
-The :class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator` can execute a specific
+The :class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator` can execute a specific
 TaskArn by specifying the ``task_arn`` parameter. This is useful when you know the TaskArn you want to execute.
 
 .. exampleinclude:: /../../airflow/providers/amazon/aws/example_dags/example_datasync_1.py
@@ -80,7 +80,7 @@ Alternatively, the operator can search in AWS DataSync for a Task based on
 
 In AWS, DataSync Tasks are linked to source and destination Locations. A location has a LocationURI and
 is referenced by a LocationArn much like other AWS resources.
-The :class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator`
+The :class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator`
 can iterate all DataSync Tasks for their source and destination LocationArns. Then it checks
 each LocationArn to see if its the URIs match the desired source / destination URI.
 
@@ -103,7 +103,7 @@ Purpose
 """""""
 
 Show how DataSync Tasks and Locations can be automatically created, deleted and updated using the
-:class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator`.
+:class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator`.
 
 Find and update a DataSync Task, or create one if it doesn't exist. Update the Task, then execute it.
 Finally, delete it.
@@ -121,7 +121,7 @@ This example relies on the following variables, which can be passed via OS envir
 Get, Create, Update, Run and Delete DataSync Tasks
 """"""""""""""""""""""""""""""""""""""""""""""""""
 
-The :class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator` is used
+The :class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator` is used
 as before but with some extra arguments.
 
 Most of the arguments (``CREATE_*_KWARGS``) provide a way for the operator to automatically create a Task
@@ -149,7 +149,7 @@ Operator behaviour
 DataSync Task execution behaviour
 """""""""""""""""""""""""""""""""
 
-Once the :class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator` has identified
+Once the :class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator` has identified
 the correct TaskArn to run (either because you specified it, or because it was found), it will then be
 executed. Whenever an AWS DataSync Task is executed it creates an AWS DataSync TaskExecution, identified
 by a TaskExecutionArn.
@@ -163,7 +163,7 @@ for inspection.
 Finally, both the TaskArn and the TaskExecutionArn are returned from the ``execute()`` method, and pushed to
 an XCom automatically if ``do_xcom_push=True``.
 
-The :class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator` supports
+The :class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator` supports
 optional passing of additional kwargs to the underlying ``boto3.start_task_execution()`` API.
 This is done with the ``task_execution_kwargs`` parameter.
 This is useful for example to limit bandwidth or filter included files - refer to the boto3 Datasync
@@ -172,12 +172,12 @@ documentation for more details.
 TaskArn selection behaviour
 """""""""""""""""""""""""""
 
-The :class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator`
+The :class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator`
 may find 0, 1, or many AWS DataSync Tasks with a matching ``source_location_uri`` and
 ``destination_location_uri``. The operator must decide what to do in each of these scenarios.
 
 To override the default behaviour, simply create an operator which inherits
-:class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator`
+:class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator`
 and re-implement the ``choose_task`` and ``choose_location`` methods
 to suit your use case.
 
@@ -204,7 +204,7 @@ TaskArn creation behaviour
 """""""""""""""""""""""""""
 
 When creating a Task, the
-:class:`~airflow.providers.amazon.aws.operators.datasync.AWSDataSyncOperator` will try to find
+:class:`~airflow.providers.amazon.aws.operators.datasync.DataSyncOperator` will try to find
 and use existing LocationArns rather than creating new ones. If multiple LocationArns match the
 specified URIs then we need to choose one to use. In this scenario, the operator behaves similarly
 to how it chooses a single Task from many Tasks:

tests/providers/amazon/aws/hooks/test_datasync.py

@@ -24,22 +24,22 @@
 from moto import mock_datasync
 
 from airflow.exceptions import AirflowTaskTimeout
-from airflow.providers.amazon.aws.hooks.datasync import AWSDataSyncHook
+from airflow.providers.amazon.aws.hooks.datasync import DataSyncHook
 
 
 @mock_datasync
-class TestAwsDataSyncHook(unittest.TestCase):
+class TestDataSyncHook(unittest.TestCase):
     def test_get_conn(self):
-        hook = AWSDataSyncHook(aws_conn_id="aws_default")
+        hook = DataSyncHook(aws_conn_id="aws_default")
         assert hook.get_conn() is not None
 
 
-# Explanation of: @mock.patch.object(AWSDataSyncHook, 'get_conn')
+# Explanation of: @mock.patch.object(DataSyncHook, 'get_conn')
 # base_aws.py fiddles with config files and changes the region
 # If you have any ~/.credentials then aws_hook uses it for the region
 # This region might not match us-east-1 used for the mocked self.client
 
-# Once patched, the AWSDataSyncHook.get_conn method is mocked and passed to the test as
+# Once patched, the DataSyncHook.get_conn method is mocked and passed to the test as
 # mock_get_conn. We then override it to just return the locally created self.client instead of
 # the one created by the AWS self.hook.
 
@@ -48,8 +48,8 @@ def test_get_conn(self):
 
 
 @mock_datasync
-@mock.patch.object(AWSDataSyncHook, "get_conn")
-class TestAWSDataSyncHookMocked(unittest.TestCase):
+@mock.patch.object(DataSyncHook, "get_conn")
+class TestDataSyncHookMocked(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.source_server_hostname = "host"
@@ -65,7 +65,7 @@ def __init__(self, *args, **kwargs):
 
     def setUp(self):
         self.client = boto3.client("datasync", region_name="us-east-1")
-        self.hook = AWSDataSyncHook(aws_conn_id="aws_default", wait_interval_seconds=0)
+        self.hook = DataSyncHook(aws_conn_id="aws_default", wait_interval_seconds=0)
 
         # Create default locations and tasks
         self.source_location_arn = self.client.create_location_smb(