Use AirflowProviderDeprecationWarning in providers (#30975)

* Use `AirflowProviderDeprecationWarning` in providers
Now that Providers minimum Airflow version is 2.4.0 we can replace `DeprecationWarning` with `AirflowProviderDeprecationWarning`

Author: eladkal

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

@@ -40,7 +40,7 @@
 from boto3.s3.transfer import S3Transfer, TransferConfig
 from botocore.exceptions import ClientError
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.exceptions import S3HookUriParseFailure
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from airflow.providers.amazon.aws.utils.tags import format_tags
@@ -74,7 +74,7 @@ def wrapper(*args, **kwargs) -> T:
                     "s3 conn_type, and the associated schema field, is deprecated."
                     " Please use aws conn_type instead, and specify `bucket_name`"
                     " in `service_config.s3` within `extras`.",
-                    DeprecationWarning,
+                    AirflowProviderDeprecationWarning,
                     stacklevel=2,
                 )
                 bound_args.arguments["bucket_name"] = self.conn_config.schema

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

@@ -29,7 +29,7 @@
 from typing import TYPE_CHECKING, Any, Sequence
 
 from airflow.compat.functools import cached_property
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.batch_client import BatchClientHook
 from airflow.providers.amazon.aws.links.batch import (
@@ -163,7 +163,7 @@ def __init__(
             self.container_overrides = overrides
             warnings.warn(
                 "Parameter `overrides` is deprecated, Please use `container_overrides` instead.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
 

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

@@ -23,7 +23,7 @@
 
 from botocore.exceptions import ClientError, WaiterError
 
-from airflow import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.eks import EksHook
 
@@ -691,7 +691,7 @@ def __init__(
                 "Currently the default for this parameter is `False` but in a future release the default "
                 "will be changed to `True`. To ensure pods are not deleted in the future you will need to "
                 "set `is_delete_operator_pod=False` explicitly.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
             is_delete_operator_pod = False

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

@@ -22,7 +22,7 @@
 from typing import TYPE_CHECKING, Any, Sequence
 from uuid import uuid4
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.emr import EmrContainerHook, EmrHook, EmrServerlessHook
 from airflow.providers.amazon.aws.links.emr import EmrClusterLink
@@ -486,7 +486,7 @@ def __init__(
             warnings.warn(
                 f"Parameter `{self.__class__.__name__}.max_tries` is deprecated and will be removed "
                 "in a future release.  Please use method `max_polling_attempts` instead.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
             if max_polling_attempts and max_polling_attempts != max_tries:

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

@@ -24,7 +24,7 @@
 from botocore.exceptions import ClientError
 
 from airflow.compat.functools import cached_property
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from airflow.providers.amazon.aws.hooks.sagemaker import SageMakerHook
@@ -201,7 +201,7 @@ def __init__(
             warnings.warn(
                 "Action 'increment' on job name conflict has been deprecated for performance reasons."
                 "The alternative to 'fail' is now 'timestamp'.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
         self.action_if_job_exists = action_if_job_exists
@@ -486,7 +486,7 @@ def __init__(
                 warnings.warn(
                     "Action 'increment' on job name conflict has been deprecated for performance reasons."
                     "The alternative to 'fail' is now 'timestamp'.",
-                    DeprecationWarning,
+                    AirflowProviderDeprecationWarning,
                     stacklevel=2,
                 )
             self.action_if_job_exists = action_if_job_exists
@@ -718,7 +718,7 @@ def __init__(
                 warnings.warn(
                     "Action 'increment' on job name conflict has been deprecated for performance reasons."
                     "The alternative to 'fail' is now 'timestamp'.",
-                    DeprecationWarning,
+                    AirflowProviderDeprecationWarning,
                     stacklevel=2,
                 )
             self.action_if_job_exists = action_if_job_exists

airflow/providers/amazon/aws/secrets/secrets_manager.py

@@ -25,6 +25,7 @@
 from urllib.parse import unquote
 
 from airflow.compat.functools import cached_property
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.utils import trim_none_values
 from airflow.secrets import BaseSecretsBackend
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -149,7 +150,7 @@ def __init__(
                 "The `full_url_mode` kwarg is deprecated. Going forward, the `SecretsManagerBackend`"
                 " will support both URL-encoded and JSON-encoded secrets at the same time. The encoding"
                 " of the secret will be determined automatically.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
 
@@ -159,7 +160,7 @@ def __init__(
                 " migrating away from URL-encoding secret values for JSON secrets."
                 " To remove this warning, make sure your JSON secrets are *NOT* URL-encoded, and then"
                 " remove this kwarg from backend_kwargs.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
             self.are_secret_values_urlencoded = kwargs.pop("are_secret_values_urlencoded", None)

airflow/providers/amazon/aws/transfers/base.py

@@ -21,6 +21,7 @@
 import warnings
 from typing import Sequence
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from airflow.utils.types import NOTSET, ArgNotSet
@@ -60,7 +61,7 @@ def __init__(
     ) -> None:
         super().__init__(**kwargs)
         if not isinstance(aws_conn_id, ArgNotSet):
-            warnings.warn(_DEPRECATION_MSG, DeprecationWarning, stacklevel=3)
+            warnings.warn(_DEPRECATION_MSG, AirflowProviderDeprecationWarning, stacklevel=3)
             self.source_aws_conn_id = aws_conn_id
         else:
             self.source_aws_conn_id = source_aws_conn_id

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

@@ -25,7 +25,7 @@
 from botocore.config import Config
 
 from airflow.compat.functools import cached_property
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.utils import trim_none_values
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.log.secrets_masker import mask_secret
@@ -168,7 +168,7 @@ def __post_init__(self, conn: Connection):
                 f"{self.conn_repr} has connection type 's3', "
                 "which has been replaced by connection type 'aws'. "
                 "Please update your connection to have `conn_type='aws'`.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
         elif self.conn_type != "aws":
@@ -186,7 +186,7 @@ def __post_init__(self, conn: Connection):
             warnings.warn(
                 "'session_kwargs' in extra config is deprecated and will be removed in a future releases. "
                 f"Please specify arguments passed to boto3 Session directly in {self.conn_repr} extra.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
 
@@ -242,7 +242,7 @@ def __post_init__(self, conn: Connection):
             warnings.warn(
                 f"Host {conn.host} specified in the connection is not used."
                 " Please, set it on extra['endpoint_url'] instead",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
 
@@ -251,7 +251,7 @@ def __post_init__(self, conn: Connection):
             warnings.warn(
                 "extra['host'] is deprecated and will be removed in a future release."
                 " Please set extra['endpoint_url'] instead",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
         else:
@@ -385,7 +385,7 @@ def _get_assume_role_configs(
                 "Constructing 'role_arn' from extra['aws_account_id'] and extra['aws_iam_role'] is deprecated"
                 f" and will be removed in a future releases."
                 f" Please set 'role_arn' in {self.conn_repr} extra.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=3,
             )
             role_arn = f"arn:aws:iam::{aws_account_id}:role/{aws_iam_role}"
@@ -413,7 +413,7 @@ def _get_assume_role_configs(
             warnings.warn(
                 "'external_id' in extra config is deprecated and will be removed in a future releases. "
                 f"Please set 'ExternalId' in 'assume_role_kwargs' in {self.conn_repr} extra.",
-                DeprecationWarning,
+                AirflowProviderDeprecationWarning,
                 stacklevel=3,
             )
             assume_role_kwargs["ExternalId"] = external_id
@@ -436,7 +436,7 @@ def _parse_s3_config(
     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.",
-        DeprecationWarning,
+        AirflowProviderDeprecationWarning,
         stacklevel=4,
     )
 

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

@@ -20,6 +20,7 @@
 import warnings
 from typing import Sequence
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
 
 
@@ -49,6 +50,6 @@ def __init__(self, *, drill_conn_id: str = "drill_default", **kwargs) -> None:
         warnings.warn(
             """This class is deprecated.
             Please use `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator`.""",
-            DeprecationWarning,
+            AirflowProviderDeprecationWarning,
             stacklevel=2,
         )

airflow/providers/apache/druid/operators/druid_check.py

@@ -19,6 +19,7 @@
 
 import warnings
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.common.sql.operators.sql import SQLCheckOperator
 
 
@@ -32,7 +33,7 @@ def __init__(self, druid_broker_conn_id: str = "druid_broker_default", **kwargs)
         warnings.warn(
             """This class is deprecated.
             Please use `airflow.providers.common.sql.operators.sql.SQLCheckOperator`.""",
-            DeprecationWarning,
+            AirflowProviderDeprecationWarning,
             stacklevel=2,
         )
         super().__init__(conn_id=druid_broker_conn_id, **kwargs)