feat: Switch all class, functions, methods deprecations to decorators (…

…#36876)

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

@@ -29,7 +29,6 @@
 import json
 import logging
 import os
-import warnings
 from copy import deepcopy
 from functools import cached_property, wraps
 from pathlib import Path
@@ -44,6 +43,7 @@
 from botocore.config import Config
 from botocore.waiter import Waiter, WaiterModel
 from dateutil.tz import tzlocal
+from deprecated import deprecated
 from slugify import slugify
 
 from airflow.configuration import conf
@@ -1020,6 +1020,13 @@ def _parse_s3_config(config_file_name: str, config_format: str | None = "boto",
     pass
 
 
+@deprecated(
+    reason=(
+        "airflow.providers.amazon.aws.hook.base_aws.BaseAsyncSessionFactory "
+        "has been deprecated and will be removed in future"
+    ),
+    category=AirflowProviderDeprecationWarning,
+)
 class BaseAsyncSessionFactory(BaseSessionFactory):
     """
     Base AWS Session Factory class to handle aiobotocore session creation.
@@ -1029,12 +1036,6 @@ class BaseAsyncSessionFactory(BaseSessionFactory):
     """
 
     def __init__(self, *args, **kwargs):
-        warnings.warn(
-            "airflow.providers.amazon.aws.hook.base_aws.BaseAsyncSessionFactory has been deprecated and "
-            "will be removed in future",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
         super().__init__(*args, **kwargs)
 
     async def get_role_credentials(self) -> dict:
@@ -1113,6 +1114,13 @@ def create_session(self, deferrable: bool = False) -> AioSession:
         return self._get_session_with_assume_role()
 
 
+@deprecated(
+    reason=(
+        "airflow.providers.amazon.aws.hook.base_aws.AwsBaseAsyncHook "
+        "has been deprecated and will be removed in future"
+    ),
+    category=AirflowProviderDeprecationWarning,
+)
 class AwsBaseAsyncHook(AwsBaseHook):
     """Interacts with AWS using aiobotocore asynchronously.
 
@@ -1129,12 +1137,6 @@ class AwsBaseAsyncHook(AwsBaseHook):
     """
 
     def __init__(self, *args, **kwargs):
-        warnings.warn(
-            "airflow.providers.amazon.aws.hook.base_aws.AwsBaseAsyncHook has been deprecated and "
-            "will be removed in future",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
         super().__init__(*args, **kwargs)
 
     def get_async_session(self) -> AioSession:

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

@@ -18,10 +18,10 @@
 from __future__ import annotations
 
 import time
-import warnings
 from functools import cached_property
 
 from botocore.exceptions import ClientError
+from deprecated import deprecated
 
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
@@ -172,14 +172,15 @@ def wait_for_state(
         return status
 
     @cached_property
-    def sts_hook(self):
-        warnings.warn(
-            f"`{type(self).__name__}.sts_hook` property is deprecated and will be removed in the future. "
+    @deprecated(
+        reason=(
+            "`QuickSightHook.sts_hook` property is deprecated and will be removed in the future. "
             "This property used for obtain AWS Account ID, "
-            f"please consider to use `{type(self).__name__}.account_id` instead",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
+            "please consider to use `QuickSightHook.account_id` instead"
+        ),
+        category=AirflowProviderDeprecationWarning,
+    )
+    def sts_hook(self):
         from airflow.providers.amazon.aws.hooks.sts import StsHook
 
         return StsHook(aws_conn_id=self.aws_conn_id)

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

@@ -17,10 +17,10 @@
 from __future__ import annotations
 
 import asyncio
-import warnings
 from typing import Any, Sequence
 
 import botocore.exceptions
+from deprecated import deprecated
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseAsyncHook, AwsBaseHook
@@ -195,16 +195,17 @@ def get_cluster_snapshot_status(self, snapshot_identifier: str):
             return None
 
 
+@deprecated(
+    reason=(
+        "airflow.providers.amazon.aws.hook.base_aws.RedshiftAsyncHook "
+        "has been deprecated and will be removed in future"
+    ),
+    category=AirflowProviderDeprecationWarning,
+)
 class RedshiftAsyncHook(AwsBaseAsyncHook):
     """Interact with AWS Redshift using aiobotocore library."""
 
     def __init__(self, *args, **kwargs):
-        warnings.warn(
-            "airflow.providers.amazon.aws.hook.base_aws.RedshiftAsyncHook has been deprecated and "
-            "will be removed in future",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
         kwargs["client_type"] = "redshift"
         super().__init__(*args, **kwargs)
 

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

@@ -25,6 +25,7 @@
 from typing import TYPE_CHECKING, Any, List, Sequence, cast
 
 from botocore.exceptions import ClientError, WaiterError
+from deprecated import deprecated
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
@@ -263,13 +264,14 @@ def hook(self) -> EksHook:
         return EksHook(aws_conn_id=self.aws_conn_id, region_name=self.region)
 
     @property
-    def eks_hook(self):
-        warnings.warn(
+    @deprecated(
+        reason=(
             "`eks_hook` property is deprecated and will be removed in the future. "
-            "Please use `hook` property instead.",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
+            "Please use `hook` property instead."
+        ),
+        category=AirflowProviderDeprecationWarning,
+    )
+    def eks_hook(self):
         return self.hook
 
     def execute(self, context: Context):

airflow/providers/amazon/aws/sensors/quicksight.py

@@ -17,10 +17,11 @@
 # under the License.
 from __future__ import annotations
 
-import warnings
 from functools import cached_property
 from typing import TYPE_CHECKING, Sequence
 
+from deprecated import deprecated
+
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning, AirflowSkipException
 from airflow.providers.amazon.aws.hooks.quicksight import QuickSightHook
 from airflow.providers.amazon.aws.sensors.base_aws import AwsBaseSensor
@@ -80,24 +81,26 @@ def poke(self, context: Context) -> bool:
         return quicksight_ingestion_state == self.success_status
 
     @cached_property
+    @deprecated(
+        reason=(
+            "`QuickSightSensor.quicksight_hook` property is deprecated, "
+            "please use `QuickSightSensor.hook` property instead."
+        ),
+        category=AirflowProviderDeprecationWarning,
+    )
     def quicksight_hook(self):
-        warnings.warn(
-            f"`{type(self).__name__}.quicksight_hook` property is deprecated, "
-            f"please use `{type(self).__name__}.hook` property instead.",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
         return self.hook
 
     @cached_property
-    def sts_hook(self):
-        warnings.warn(
-            f"`{type(self).__name__}.sts_hook` property is deprecated and will be removed in the future. "
+    @deprecated(
+        reason=(
+            "`QuickSightSensor.sts_hook` property is deprecated and will be removed in the future. "
             "This property used for obtain AWS Account ID, "
-            f"please consider to use `{type(self).__name__}.hook.account_id` instead",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
+            "please consider to use `QuickSightSensor.hook.account_id` instead"
+        ),
+        category=AirflowProviderDeprecationWarning,
+    )
+    def sts_hook(self):
         from airflow.providers.amazon.aws.hooks.sts import StsHook
 
         return StsHook(aws_conn_id=self.aws_conn_id)

airflow/providers/amazon/aws/triggers/rds.py

@@ -16,10 +16,11 @@
 # under the License.
 from __future__ import annotations
 
-import warnings
 from functools import cached_property
 from typing import TYPE_CHECKING, Any
 
+from deprecated import deprecated
+
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.rds import RdsHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
@@ -31,6 +32,13 @@
     from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 
 
+@deprecated(
+    reason=(
+        "This trigger is deprecated, please use the other RDS triggers "
+        "such as RdsDbDeletedTrigger, RdsDbStoppedTrigger or RdsDbAvailableTrigger"
+    ),
+    category=AirflowProviderDeprecationWarning,
+)
 class RdsDbInstanceTrigger(BaseTrigger):
     """
     Deprecated Trigger for RDS operations. Do not use.
@@ -55,12 +63,6 @@ def __init__(
         region_name: str | None,
         response: dict[str, Any],
     ):
-        warnings.warn(
-            "This trigger is deprecated, please use the other RDS triggers "
-            "such as RdsDbDeletedTrigger, RdsDbStoppedTrigger or RdsDbAvailableTrigger",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
         self.db_instance_identifier = db_instance_identifier
         self.waiter_delay = waiter_delay
         self.waiter_max_attempts = waiter_max_attempts

airflow/providers/amazon/aws/utils/connection_wrapper.py

@@ -25,6 +25,7 @@
 
 from botocore import UNSIGNED
 from botocore.config import Config
+from deprecated import deprecated
 
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.utils import trim_none_values
@@ -462,6 +463,13 @@ def _get_assume_role_configs(
         return role_arn, assume_role_method, assume_role_kwargs
 
 
+@deprecated(
+    reason=(
+        "Use local credentials file is never documented and well tested. "
+        "Obtain credentials by this way deprecated and will be removed in a future releases."
+    ),
+    category=AirflowProviderDeprecationWarning,
+)
 def _parse_s3_config(
     config_file_name: str, config_format: str | None = "boto", profile: str | None = None
 ) -> tuple[str | None, str | None]:
@@ -474,13 +482,6 @@ def _parse_s3_config(
         Defaults to "boto"
     :param profile: profile name in AWS type config file
     """
-    warnings.warn(
-        "Use local credentials file is never documented and well tested. "
-        "Obtain credentials by this way deprecated and will be removed in a future releases.",
-        AirflowProviderDeprecationWarning,
-        stacklevel=4,
-    )
-
     import configparser
 
     config = configparser.ConfigParser()

airflow/providers/amazon/aws/utils/mixins.py

@@ -31,6 +31,7 @@
 from functools import cached_property
 from typing import Any, Generic, NamedTuple, TypeVar
 
+from deprecated import deprecated
 from typing_extensions import final
 
 from airflow.compat.functools import cache
@@ -160,9 +161,12 @@ def hook(self) -> AwsHookType:
 
     @property
     @final
+    @deprecated(
+        reason="`region` is deprecated and will be removed in the future. Please use `region_name` instead.",
+        category=AirflowProviderDeprecationWarning,
+    )
     def region(self) -> str | None:
         """Alias for ``region_name``, used for compatibility (deprecated)."""
-        warnings.warn(REGION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
         return self.region_name
 
 

airflow/providers/apache/beam/triggers/beam.py

@@ -17,9 +17,9 @@
 from __future__ import annotations
 
 import asyncio
-import warnings
 from typing import Any, AsyncIterator, Sequence
 
+from deprecated import deprecated
 from google.cloud.dataflow_v1beta3 import ListJobsRequest
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
@@ -252,6 +252,10 @@ async def run(self) -> AsyncIterator[TriggerEvent]:  # type: ignore[override]
         return
 
 
+@deprecated(
+    reason="`BeamPipelineTrigger` is deprecated. Please use `BeamPythonPipelineTrigger`.",
+    category=AirflowProviderDeprecationWarning,
+)
 class BeamPipelineTrigger(BeamPythonPipelineTrigger):
     """
     Trigger to perform checking the Python pipeline status until it reaches terminate state.
@@ -262,9 +266,4 @@ class BeamPipelineTrigger(BeamPythonPipelineTrigger):
     """
 
     def __init__(self, *args, **kwargs):
-        warnings.warn(
-            "`BeamPipelineTrigger` is deprecated. Please use `BeamPythonPipelineTrigger`.",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
         super().__init__(*args, **kwargs)

airflow/providers/apache/drill/operators/drill.py

@@ -17,13 +17,18 @@
 # under the License.
 from __future__ import annotations
 
-import warnings
 from typing import Sequence
 
+from deprecated import deprecated
+
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
 
 
+@deprecated(
+    reason="Please use `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator`.",
+    category=AirflowProviderDeprecationWarning,
+)
 class DrillOperator(SQLExecuteQueryOperator):
     """
     Executes the provided SQL in the identified Drill environment.
@@ -51,9 +56,3 @@ class DrillOperator(SQLExecuteQueryOperator):
 
     def __init__(self, *, drill_conn_id: str = "drill_default", **kwargs) -> None:
         super().__init__(conn_id=drill_conn_id, **kwargs)
-        warnings.warn(
-            """This class is deprecated.
-            Please use `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator`.""",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )