Fix stacklevel in warnings.warn into the providers

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

@@ -716,7 +716,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_countdown has been deprecated to standardize "
                 "naming conventions.  Please use waiter_max_attempts instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
             # waiter_countdown defaults to never timing out, which is not supported
             # by boto waiters, so we will set it here to "a very long time" for now.
@@ -725,7 +727,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_check_interval_seconds has been deprecated to "
                 "standardize naming conventions.  Please use waiter_delay instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
             waiter_delay = waiter_check_interval_seconds
         super().__init__(**kwargs)
@@ -1024,7 +1028,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_check_interval_seconds has been deprecated to standardize "
                 "naming conventions.  Please use waiter_delay instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         if waiter_countdown is NOTSET:
             waiter_max_attempts = 25 if waiter_max_attempts is NOTSET else waiter_max_attempts
@@ -1036,7 +1042,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_countdown has been deprecated to standardize "
                 "naming conventions.  Please use waiter_max_attempts instead. In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         self.aws_conn_id = aws_conn_id
         self.release_label = release_label
@@ -1205,7 +1213,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_check_interval_seconds has been deprecated to standardize "
                 "naming conventions.  Please use waiter_delay instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         if waiter_countdown is NOTSET:
             waiter_max_attempts = 25 if waiter_max_attempts is NOTSET else waiter_max_attempts
@@ -1217,7 +1227,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_countdown has been deprecated to standardize "
                 "naming conventions.  Please use waiter_max_attempts instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         self.aws_conn_id = aws_conn_id
         self.application_id = application_id
@@ -1408,7 +1420,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_check_interval_seconds has been deprecated to standardize "
                 "naming conventions.  Please use waiter_delay instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         if waiter_countdown is NOTSET:
             waiter_max_attempts = 25 if waiter_max_attempts is NOTSET else waiter_max_attempts
@@ -1420,7 +1434,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_countdown has been deprecated to standardize "
                 "naming conventions.  Please use waiter_max_attempts instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         self.aws_conn_id = aws_conn_id
         self.application_id = application_id
@@ -1569,7 +1585,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_check_interval_seconds has been deprecated to standardize "
                 "naming conventions.  Please use waiter_delay instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         if waiter_countdown is NOTSET:
             waiter_max_attempts = 25 if waiter_max_attempts is NOTSET else waiter_max_attempts
@@ -1581,7 +1599,9 @@ def __init__(
             warnings.warn(
                 "The parameter waiter_countdown has been deprecated to standardize "
                 "naming conventions.  Please use waiter_max_attempts instead.  In the "
-                "future this will default to None and defer to the waiter's default value."
+                "future this will default to None and defer to the waiter's default value.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
         self.wait_for_delete_completion = wait_for_completion
         # super stops the app

airflow/providers/celery/executors/celery_executor_utils.py

@@ -41,7 +41,7 @@
 
 import airflow.settings as settings
 from airflow.configuration import conf
-from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.executors.base_executor import BaseExecutor
 from airflow.stats import Stats
 from airflow.utils.dag_parsing_context import _airflow_parsing_context_manager
@@ -88,7 +88,8 @@ def _get_celery_app() -> Celery:
             "Change it to `airflow.providers.celery.executors.celery_executor`, and "
             "update the `-app` flag in your Celery Health Checks "
             "to use `airflow.providers.celery.executors.celery_executor.app`.",
-            RemovedInAirflow3Warning,
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
 
     return Celery(celery_app_name, config_source=celery_configuration)

airflow/providers/cncf/kubernetes/pod_generator.py

@@ -39,7 +39,7 @@
 from airflow.exceptions import (
     AirflowConfigException,
     AirflowException,
-    RemovedInAirflow3Warning,
+    AirflowProviderDeprecationWarning,
 )
 from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
     POD_NAME_MAX_LENGTH,
@@ -155,7 +155,7 @@ def __init__(
 
     def gen_pod(self) -> k8s.V1Pod:
         """Generate pod."""
-        warnings.warn("This function is deprecated. ", RemovedInAirflow3Warning)
+        warnings.warn("This function is deprecated. ", AirflowProviderDeprecationWarning, stacklevel=2)
         result = self.ud_pod
 
         result.metadata.name = add_pod_suffix(pod_name=result.metadata.name)
@@ -170,7 +170,9 @@ def add_xcom_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
         """Add sidecar."""
         warnings.warn(
             "This function is deprecated. "
-            "Please use airflow.providers.cncf.kubernetes.utils.xcom_sidecar.add_xcom_sidecar instead"
+            "Please use airflow.providers.cncf.kubernetes.utils.xcom_sidecar.add_xcom_sidecar instead",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         pod_cp = copy.deepcopy(pod)
         pod_cp.spec.volumes = pod.spec.volumes or []
@@ -207,7 +209,8 @@ def from_obj(obj) -> dict | k8s.V1Pod | None:
                 "Using a dictionary for the executor_config is deprecated and will soon be removed."
                 'please use a `kubernetes.client.models.V1Pod` class with a "pod_override" key'
                 " instead. ",
-                category=RemovedInAirflow3Warning,
+                category=AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
             return PodGenerator.from_legacy_obj(obj)
         else:
@@ -386,7 +389,10 @@ def construct_pod(
         """
         if len(pod_id) > POD_NAME_MAX_LENGTH:
             warnings.warn(
-                f"pod_id supplied is longer than {POD_NAME_MAX_LENGTH} characters; truncating and adding unique suffix."
+                f"pod_id supplied is longer than {POD_NAME_MAX_LENGTH} characters; "
+                f"truncating and adding unique suffix.",
+                UserWarning,
+                stacklevel=2,
             )
             pod_id = add_pod_suffix(pod_name=pod_id, max_len=POD_NAME_MAX_LENGTH)
         try:
@@ -583,7 +589,8 @@ def make_unique_pod_id(pod_id: str) -> str | None:
         """
         warnings.warn(
             "This function is deprecated. Use `add_pod_suffix` in `kubernetes_helper_functions`.",
-            RemovedInAirflow3Warning,
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
 
         if not pod_id:

airflow/providers/dbt/cloud/operators/dbt.py

@@ -184,6 +184,8 @@ def execute(self, context: Context):
                 warnings.warn(
                     "Argument `wait_for_termination` is False and `deferrable` is True , hence "
                     "`deferrable` parameter doesn't have any effect",
+                    UserWarning,
+                    stacklevel=2,
                 )
             return self.run_id
 

airflow/providers/elasticsearch/log/es_task_handler.py

@@ -171,6 +171,7 @@ def __init__(
             warnings.warn(
                 "Passing log_id_template to ElasticsearchTaskHandler is deprecated and has no effect",
                 AirflowProviderDeprecationWarning,
+                stacklevel=2,
             )
 
         self.log_id_template = log_id_template  # Only used on Airflow < 2.3.2.

airflow/providers/google/cloud/hooks/bigquery.py

@@ -145,6 +145,7 @@ def get_service(self) -> Resource:
         warnings.warn(
             "This method will be deprecated. Please use `BigQueryHook.get_client` method",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         http_authorized = self._authorize()
         return build("bigquery", "v2", http=http_authorized, cache_discovery=False)
@@ -624,6 +625,7 @@ def create_external_table(
             "This method is deprecated. Please use `BigQueryHook.create_empty_table` method with "
             "passing the `table_resource` object. This gives more flexibility than this method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         location = location or self.location
         src_fmt_configs = src_fmt_configs or {}
@@ -808,6 +810,7 @@ def patch_table(
         warnings.warn(
             "This method is deprecated, please use ``BigQueryHook.update_table`` method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         table_resource: dict[str, Any] = {}
 
@@ -958,7 +961,9 @@ def patch_dataset(self, dataset_id: str, dataset_resource: dict, project_id: str
         :param project_id: The Google Cloud Project ID
         """
         warnings.warn(
-            "This method is deprecated. Please use ``update_dataset``.", AirflowProviderDeprecationWarning
+            "This method is deprecated. Please use ``update_dataset``.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         project_id = project_id or self.project_id
         if not dataset_id or not isinstance(dataset_id, str):
@@ -1007,7 +1012,9 @@ def get_dataset_tables_list(
         :return: List of tables associated with the dataset
         """
         warnings.warn(
-            "This method is deprecated. Please use ``get_dataset_tables``.", AirflowProviderDeprecationWarning
+            "This method is deprecated. Please use ``get_dataset_tables``.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         project_id = project_id or self.project_id
         tables = self.get_client().list_tables(
@@ -1197,7 +1204,9 @@ def run_table_delete(self, deletion_dataset_table: str, ignore_if_missing: bool
         :return:
         """
         warnings.warn(
-            "This method is deprecated. Please use `delete_table`.", AirflowProviderDeprecationWarning
+            "This method is deprecated. Please use `delete_table`.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         return self.delete_table(table_id=deletion_dataset_table, not_found_ok=ignore_if_missing)
 
@@ -1250,7 +1259,11 @@ def get_tabledata(
         :param start_index: zero based index of the starting row to read.
         :return: list of rows
         """
-        warnings.warn("This method is deprecated. Please use `list_rows`.", AirflowProviderDeprecationWarning)
+        warnings.warn(
+            "This method is deprecated. Please use `list_rows`.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
         rows = self.list_rows(
             dataset_id=dataset_id,
             table_id=table_id,
@@ -1458,6 +1471,7 @@ def cancel_query(self) -> None:
         warnings.warn(
             "This method is deprecated. Please use `BigQueryHook.cancel_job`.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         if self.running_job_id:
             self.cancel_job(job_id=self.running_job_id)
@@ -1617,6 +1631,7 @@ def run_with_configuration(self, configuration: dict) -> str:
         warnings.warn(
             "This method is deprecated. Please use `BigQueryHook.insert_job`",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         job = self.insert_job(configuration=configuration, project_id=self.project_id)
         self.running_job_id = job.job_id
@@ -1714,6 +1729,7 @@ def run_load(
         warnings.warn(
             "This method is deprecated. Please use `BigQueryHook.insert_job` method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
 
         if not self.project_id:
@@ -1901,6 +1917,7 @@ def run_copy(
         warnings.warn(
             "This method is deprecated. Please use `BigQueryHook.insert_job` method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         if not self.project_id:
             raise ValueError("The project_id should be set")
@@ -1982,6 +1999,7 @@ def run_extract(
         warnings.warn(
             "This method is deprecated. Please use `BigQueryHook.insert_job` method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         if not self.project_id:
             raise ValueError("The project_id should be set")
@@ -2109,6 +2127,7 @@ def run_query(
         warnings.warn(
             "This method is deprecated. Please use `BigQueryHook.insert_job` method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         if not self.project_id:
             raise ValueError("The project_id should be set")

airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py

@@ -250,7 +250,11 @@ def list_transfer_job(self, request_filter: dict | None = None, **kwargs) -> lis
                 request_filter = kwargs["filter"]
                 if not isinstance(request_filter, dict):
                     raise ValueError(f"The request_filter should be dict and is {type(request_filter)}")
-                warnings.warn("Use 'request_filter' instead of 'filter'", AirflowProviderDeprecationWarning)
+                warnings.warn(
+                    "Use 'request_filter' instead of 'filter'",
+                    AirflowProviderDeprecationWarning,
+                    stacklevel=2,
+                )
             else:
                 raise TypeError("list_transfer_job missing 1 required positional argument: 'request_filter'")
 
@@ -374,7 +378,11 @@ def list_transfer_operations(self, request_filter: dict | None = None, **kwargs)
                 request_filter = kwargs["filter"]
                 if not isinstance(request_filter, dict):
                     raise ValueError(f"The request_filter should be dict and is {type(request_filter)}")
-                warnings.warn("Use 'request_filter' instead of 'filter'", AirflowProviderDeprecationWarning)
+                warnings.warn(
+                    "Use 'request_filter' instead of 'filter'",
+                    AirflowProviderDeprecationWarning,
+                    stacklevel=2,
+                )
             else:
                 raise TypeError(
                     "list_transfer_operations missing 1 required positional argument: 'request_filter'"

airflow/providers/google/cloud/hooks/kubernetes_engine.py

@@ -102,6 +102,7 @@ def get_conn(self) -> container_v1.ClusterManagerClient:
         warnings.warn(
             "The get_conn method has been deprecated. You should use the get_cluster_manager_client method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         return self.get_cluster_manager_client()
 
@@ -111,6 +112,7 @@ def get_client(self) -> ClusterManagerClient:
         warnings.warn(
             "The get_client method has been deprecated. You should use the get_conn method.",
             AirflowProviderDeprecationWarning,
+            stacklevel=2,
         )
         return self.get_conn()