Show deprecation status of operators and sensors on the docs operator…

… list page (#1568)

This PR introduces enhancements to the documentation for operators and sensors, specifically to indicate their deprecation status and suggest replacements where applicable.

# Changes

## New Attributes
Added `is_deprecated` and `post_deprecation_replacement` attributes to operator and sensor classes.
1. Deprecation Information: The `is_deprecated` attribute is a boolean that flags whether a class is deprecated.
2. Replacement Suggestion: The `post_deprecation_replacement` attribute provides a string with the import path for the recommended replacement.

## Script changes that builds the operators list
The script `docs/_ext/traverse_operators_sensors.py` that builds in to the documentation the list of operators and sensors now uses the above newly added attributes to denote whether the operators have been deprecated

closes: #1567 

---------
Co-authored-by: Kaxil Naik <[email protected]>

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

@@ -23,6 +23,9 @@ class BatchOperatorAsync(BatchOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.operators.batch import BatchOperator"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (

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

@@ -9,6 +9,11 @@ class EmrContainerOperatorAsync(EmrContainerOperator):
     Please use :class: `~airflow.providers.amazon.aws.operators.emr.EmrContainerOperator`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.emr import EmrContainerOperator"
+    )
+
     def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
         warnings.warn(
             (

astronomer/providers/amazon/aws/operators/redshift_cluster.py

@@ -13,6 +13,11 @@ class RedshiftDeleteClusterOperatorAsync(RedshiftDeleteClusterOperator):
     Please use :class: `~airflow.providers.amazon.aws.operators.redshift_cluster.RedshiftDeleteClusterOperator`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.redshift_cluster import RedshiftDeleteClusterOperator"
+    )
+
     def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
         warnings.warn(
             (
@@ -32,6 +37,11 @@ class RedshiftResumeClusterOperatorAsync(RedshiftResumeClusterOperator):
     Please use :class: `~airflow.providers.amazon.aws.operators.redshift_cluster.RedshiftResumeClusterOperator`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.redshift_cluster import RedshiftResumeClusterOperator"
+    )
+
     def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
         warnings.warn(
             (
@@ -51,6 +61,11 @@ class RedshiftPauseClusterOperatorAsync(RedshiftPauseClusterOperator):
     Please use :class: `~airflow.providers.amazon.aws.operators.redshift_cluster.RedshiftPauseClusterOperator`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.redshift_cluster import RedshiftPauseClusterOperator"
+    )
+
     def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
         warnings.warn(
             (

astronomer/providers/amazon/aws/operators/redshift_data.py

@@ -11,6 +11,11 @@ class RedshiftDataOperatorAsync(RedshiftDataOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.redshift_data import RedshiftDataOperator"
+    )
+
     def __init__(
         self,
         **kwargs: Any,

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

@@ -24,6 +24,11 @@ class SageMakerProcessingOperatorAsync(SageMakerProcessingOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.sagemaker import SageMakerProcessingOperator"
+    )
+
     def __init__(self, **kwargs: Any) -> None:
         warnings.warn(
             (
@@ -44,6 +49,11 @@ class SageMakerTransformOperatorAsync(SageMakerTransformOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.sagemaker import SageMakerTransformOperator"
+    )
+
     def __init__(self, **kwargs: Any) -> None:
         warnings.warn(
             (
@@ -64,6 +74,11 @@ class SageMakerTrainingOperatorAsync(SageMakerTrainingOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.operators.sagemaker import SageMakerTrainingOperator"
+    )
+
     def __init__(self, **kwargs: Any) -> None:
         warnings.warn(
             (

astronomer/providers/amazon/aws/sensors/batch.py

@@ -9,6 +9,9 @@ class BatchSensorAsync(BatchSensor):
     Please use :class: `~airflow.providers.amazon.aws.sensors.batch.BatchSensor`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.batch import BatchSensor"
+
     def __init__(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
         warnings.warn(
             (

astronomer/providers/amazon/aws/sensors/emr.py

@@ -15,6 +15,9 @@ class EmrContainerSensorAsync(EmrContainerSensor):
     Please use :class: `~airflow.providers.amazon.aws.sensors.emr.EmrContainerSensor`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.emr import EmrContainerSensor"
+
     def __init__(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
         poll_interval = kwargs.pop("poll_interval")
         if poll_interval:
@@ -44,6 +47,9 @@ class EmrStepSensorAsync(EmrStepSensor):
     Please use :class: `~airflow.providers.amazon.aws.sensors.emr.EmrStepSensor`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.emr import EmrStepSensor"
+
     def __init__(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
         warnings.warn(
             (
@@ -63,6 +69,9 @@ class EmrJobFlowSensorAsync(EmrJobFlowSensor):
     Please use :class: `~airflow.providers.amazon.aws.sensors.emr.EmrJobFlowSensor`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.emr import EmrJobFlowSensor"
+
     def __init__(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
         warnings.warn(
             (

astronomer/providers/amazon/aws/sensors/redshift_cluster.py

@@ -11,6 +11,11 @@ class RedshiftClusterSensorAsync(RedshiftClusterSensor):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.amazon.aws.sensors.redshift_cluster import RedshiftClusterSensor"
+    )
+
     def __init__(
         self,
         **kwargs: Any,

astronomer/providers/amazon/aws/sensors/s3.py

@@ -14,6 +14,9 @@ class S3KeySensorAsync(S3KeySensor):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor"
+
     def __init__(self, **kwargs: Any) -> None:
         warnings.warn(
             (
@@ -30,9 +33,12 @@ def __init__(self, **kwargs: Any) -> None:
 class S3KeySizeSensorAsync(S3KeySensorAsync):
     """
     This class is deprecated.
-    Please use :class: `~astronomer.providers.amazon.aws.sensor.s3.S3KeySensorAsync`.
+    Please use :class: `~airflow.providers.amazon.aws.sensors.s3.S3KeySensor`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor"
+
     def __init__(
         self,
         *,
@@ -42,7 +48,7 @@ def __init__(
         warnings.warn(
             """
             S3KeySizeSensorAsync is deprecated.
-            Please use `astronomer.providers.amazon.aws.sensor.s3.S3KeySensorAsync`.
+            Please use `airflow.providers.amazon.aws.sensors.s3.S3KeySensor`.
             """,
             DeprecationWarning,
             stacklevel=2,
@@ -57,6 +63,9 @@ class S3KeysUnchangedSensorAsync(S3KeysUnchangedSensor):
     Please use :class: `~airflow.providers.amazon.aws.sensors.s3.S3KeysUnchangedSensor`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.s3 import S3KeysUnchangedSensor"
+
     def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
         warnings.warn(
             (
@@ -73,9 +82,12 @@ def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
 class S3PrefixSensorAsync(BaseSensorOperator):
     """
     This class is deprecated.
-    Please use :class: `~astronomer.providers.amazon.aws.sensor.s3.S3KeySensorAsync`.
+    Please use :class: `~airflow.providers.amazon.aws.sensors.s3.S3KeySensor`.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor"
+
     def __init__(
         self,
         *,
@@ -89,7 +101,7 @@ def __init__(
         warnings.warn(
             """
             S3PrefixSensor is deprecated.
-            Please use `astronomer.providers.amazon.aws.sensor.s3.S3KeySensorAsync`.
+            Please use `airflow.providers.amazon.aws.sensors.s3.S3KeySensor`.
             """,
             DeprecationWarning,
             stacklevel=2,

astronomer/providers/apache/livy/operators/livy.py

@@ -15,6 +15,9 @@ class LivyOperatorAsync(LivyOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.apache.livy.operators.livy import LivyOperator"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             "This class is deprecated. "

astronomer/providers/cncf/kubernetes/operators/kubernetes_pod.py

@@ -21,6 +21,11 @@ class KubernetesPodOperatorAsync(KubernetesPodOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator"
+    )
+
     def __init__(self, **kwargs: Any):
         warnings.warn(
             (

astronomer/providers/core/sensors/external_task.py

@@ -21,6 +21,9 @@
 
 
 class ExternalTaskSensorAsync(ExternalTaskSensor):  # noqa: D101
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.sensors.external_task import ExternalTaskSensor"
+
     def __init__(
         self,
         poke_interval: float = 5.0,

astronomer/providers/core/sensors/filesystem.py

@@ -26,6 +26,9 @@ class FileSensorAsync(FileSensor):
         ``**`` in glob filepath parameter. Defaults to ``False``.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.sensors.filesystem import FileSensor"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (

astronomer/providers/databricks/hooks/databricks.py

@@ -128,7 +128,7 @@ async def _do_api_call_async(
                         json=json if method in ("POST", "PATCH") else None,
                         params=json if method == "GET" else None,
                         headers=headers,
-                        timeout=self.timeout_seconds,
+                        timeout=self.timeout_seconds,  # type: ignore[arg-type]
                     )
                     response.raise_for_status()
                     return cast(Dict[str, Any], await response.json())

astronomer/providers/databricks/operators/databricks.py

@@ -16,6 +16,11 @@ class DatabricksSubmitRunOperatorAsync(DatabricksSubmitRunOperator):
     `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator"
+    )
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             "This class is deprecated."
@@ -32,6 +37,11 @@ class DatabricksRunNowOperatorAsync(DatabricksRunNowOperator):
     `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator"
+    )
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             "This class is deprecated."

astronomer/providers/dbt/cloud/operators/dbt.py

@@ -17,6 +17,11 @@ class DbtCloudRunJobOperatorAsync(DbtCloudRunJobOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = (
+        "from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator"
+    )
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (

astronomer/providers/dbt/cloud/sensors/dbt.py

@@ -13,6 +13,9 @@ class DbtCloudJobRunSensorAsync(DbtCloudJobRunSensor):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from airflow.providers.dbt.cloud.sensors import DbtCloudJobRunSensor"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (

astronomer/providers/google/cloud/operators/bigquery.py

@@ -23,6 +23,9 @@ class BigQueryInsertJobOperatorAsync(BigQueryInsertJobOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from google.cloud.operators.bigquery import BigQueryInsertJobOperator"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (
@@ -45,6 +48,9 @@ class BigQueryCheckOperatorAsync(BigQueryCheckOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from google.cloud.operators.bigquery import BigQueryCheckOperator"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (
@@ -65,6 +71,9 @@ class BigQueryGetDataOperatorAsync(BigQueryGetDataOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from google.cloud.operators.bigquery import BigQueryGetDataOperator"
+
     def __init__(self, *args: Any, use_legacy_sql: bool = False, **kwargs: Any) -> None:
         warnings.warn(
             (
@@ -86,6 +95,9 @@ class BigQueryIntervalCheckOperatorAsync(BigQueryIntervalCheckOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from google.cloud.operators.bigquery import BigQueryIntervalCheckOperator"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (
@@ -106,6 +118,9 @@ class BigQueryValueCheckOperatorAsync(BigQueryValueCheckOperator):
     and set `deferrable` param to `True` instead.
     """
 
+    is_deprecated = True
+    post_deprecation_replacement = "from google.cloud.operators.bigquery import BigQueryValueCheckOperator"
+
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         warnings.warn(
             (