Deprecate CloudComposerEnvironmentSensor in favor of CloudComposerCreateEnvironmentOperator with defer mode

airflow/providers/google/cloud/operators/cloud_composer.py

@@ -112,7 +112,7 @@ class CloudComposerCreateEnvironmentOperator(GoogleCloudBaseOperator):
     :param metadata: Strings which should be sent along with the request as metadata.
     :param deferrable: Run operator in the deferrable mode
     :param pooling_period_seconds: Optional: Control the rate of the poll for the result of deferrable run.
-        By default the trigger will poll every 30 seconds.
+        By default, the trigger will poll every 30 seconds.
     """
 
     template_fields = (

airflow/providers/google/cloud/sensors/cloud_composer.py

@@ -19,9 +19,10 @@
 
 from __future__ import annotations
 
+import warnings
 from typing import TYPE_CHECKING, Any, Sequence
 
-from airflow.exceptions import AirflowException, AirflowSkipException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning, AirflowSkipException
 from airflow.providers.google.cloud.triggers.cloud_composer import CloudComposerExecutionTrigger
 from airflow.sensors.base import BaseSensorOperator
 
@@ -33,6 +34,11 @@ class CloudComposerEnvironmentSensor(BaseSensorOperator):
     """
     Check the status of the Cloud Composer Environment task.
 
+    This Sensor is deprecated. You can achieve the same functionality by using Cloud Composer Operators
+    CloudComposerCreateEnvironmentOperator, CloudComposerDeleteEnvironmentOperator and
+    CloudComposerUpdateEnvironmentOperator in  deferrable or non-deferrable mode, since every operator
+    gives user a possibility to wait (asynchronously or synchronously) until Operation will be finished.
+
     :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
     :param region: Required. The ID of the Google Cloud region that the service belongs to.
     :param operation_name: The name of the operation resource
@@ -59,6 +65,13 @@ def __init__(
         pooling_period_seconds: int = 30,
         **kwargs,
     ):
+        warnings.warn(
+            f"The `{self.__class__.__name__}` operator is deprecated. You can achieve the same functionality "
+            f"by using operators in deferrable or non-deferrable mode, since every operator for Cloud "
+            f"Composer will wait for the operation to complete.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
         super().__init__(**kwargs)
         self.project_id = project_id
         self.region = region

docs/apache-airflow-providers-google/operators/cloud/cloud_composer.rst

@@ -57,7 +57,7 @@ With this configuration we can create the environment:
 or you can define the same operator in the deferrable mode:
 :class:`~airflow.providers.google.cloud.operators.cloud_composer.CloudComposerCreateEnvironmentOperator`
 
-.. exampleinclude:: /../../tests/system/providers/google/cloud/composer/example_cloud_composer_deferrable.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/composer/example_cloud_composer.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_create_composer_environment_deferrable_mode]
@@ -116,7 +116,7 @@ To update a service you can use:
 or you can define the same operator in the deferrable mode:
 :class:`~airflow.providers.google.cloud.operators.cloud_composer.CloudComposerCreateEnvironmentOperator`
 
-.. exampleinclude:: /../../tests/system/providers/google/cloud/composer/example_cloud_composer_deferrable.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/composer/example_cloud_composer.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_update_composer_environment_deferrable_mode]
@@ -138,7 +138,7 @@ To delete a service you can use:
 or you can define the same operator in the deferrable mode:
 :class:`~airflow.providers.google.cloud.operators.cloud_composer.CloudComposerDeleteEnvironmentOperator`
 
-.. exampleinclude:: /../../tests/system/providers/google/cloud/composer/example_cloud_composer_deferrable.py
+.. exampleinclude:: /../../tests/system/providers/google/cloud/composer/example_cloud_composer.py
     :language: python
     :dedent: 4
     :start-after: [START howto_operator_delete_composer_environment_deferrable_mode]

tests/always/test_project_structure.py

@@ -398,6 +398,7 @@ class TestGoogleProviderProjectStructure(ExampleCoverageTest, AssetsCoverageTest
         "airflow.providers.google.cloud.operators.bigquery.BigQueryPatchDatasetOperator",
         "airflow.providers.google.cloud.operators.dataflow.DataflowCreatePythonJobOperator",
         "airflow.providers.google.cloud.operators.bigquery.BigQueryExecuteQueryOperator",
+        "airflow.providers.google.cloud.sensors.cloud_composer.CloudComposerEnvironmentSensor",
         "airflow.providers.google.marketing_platform.operators.GoogleDisplayVideo360CreateQueryOperator",
         "airflow.providers.google.marketing_platform.operators.GoogleDisplayVideo360RunQueryOperator",
         "airflow.providers.google.marketing_platform.operators.GoogleDisplayVideo360DownloadReportV2Operator",

tests/system/providers/google/cloud/composer/example_cloud_composer.py

@@ -32,31 +32,31 @@
 )
 from airflow.utils.trigger_rule import TriggerRule
 
-ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID", "default")
 PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT", "default")
 
 DAG_ID = "example_composer"
-
 REGION = "us-central1"
 
 # [START howto_operator_composer_simple_environment]
 
 ENVIRONMENT_ID = f"test-{DAG_ID}-{ENV_ID}".replace("_", "-")
+ENVIRONMENT_ID_ASYNC = f"test-deferrable-{DAG_ID}-{ENV_ID}".replace("_", "-")
 
 ENVIRONMENT = {
     "config": {
-        "software_config": {"image_version": "composer-2.0.28-airflow-2.2.5"},
+        "software_config": {"image_version": "composer-2.5.0-airflow-2.5.3"},
     }
 }
 # [END howto_operator_composer_simple_environment]
 
 # [START howto_operator_composer_update_environment]
 UPDATED_ENVIRONMENT = {
     "labels": {
-        "label1": "testing",
+        "label": "testing",
     }
 }
-UPDATE_MASK = {"paths": ["labels.label1"]}
+UPDATE_MASK = {"paths": ["labels.label"]}
 # [END howto_operator_composer_update_environment]
 
 
@@ -85,6 +85,17 @@
     )
     # [END howto_operator_create_composer_environment]
 
+    # [START howto_operator_create_composer_environment_deferrable_mode]
+    defer_create_env = CloudComposerCreateEnvironmentOperator(
+        task_id="defer_create_env",
+        project_id=PROJECT_ID,
+        region=REGION,
+        environment_id=ENVIRONMENT_ID_ASYNC,
+        environment=ENVIRONMENT,
+        deferrable=True,
+    )
+    # [END howto_operator_create_composer_environment_deferrable_mode]
+
     # [START howto_operator_list_composer_environments]
     list_envs = CloudComposerListEnvironmentsOperator(
         task_id="list_envs", project_id=PROJECT_ID, region=REGION
@@ -111,6 +122,18 @@
     )
     # [END howto_operator_update_composer_environment]
 
+    # [START howto_operator_update_composer_environment_deferrable_mode]
+    defer_update_env = CloudComposerUpdateEnvironmentOperator(
+        task_id="defer_update_env",
+        project_id=PROJECT_ID,
+        region=REGION,
+        environment_id=ENVIRONMENT_ID_ASYNC,
+        update_mask=UPDATE_MASK,
+        environment=UPDATED_ENVIRONMENT,
+        deferrable=True,
+    )
+    # [END howto_operator_update_composer_environment_deferrable_mode]
+
     # [START howto_operator_delete_composer_environment]
     delete_env = CloudComposerDeleteEnvironmentOperator(
         task_id="delete_env",
@@ -121,7 +144,25 @@
     # [END howto_operator_delete_composer_environment]
     delete_env.trigger_rule = TriggerRule.ALL_DONE
 
-    chain(image_versions, create_env, list_envs, get_env, update_env, delete_env)
+    # [START howto_operator_delete_composer_environment_deferrable_mode]
+    defer_delete_env = CloudComposerDeleteEnvironmentOperator(
+        task_id="defer_delete_env",
+        project_id=PROJECT_ID,
+        region=REGION,
+        environment_id=ENVIRONMENT_ID_ASYNC,
+        deferrable=True,
+    )
+    # [END howto_operator_delete_composer_environment_deferrable_mode]
+    defer_delete_env.trigger_rule = TriggerRule.ALL_DONE
+
+    chain(
+        image_versions,
+        [create_env, defer_create_env],
+        list_envs,
+        get_env,
+        [update_env, defer_update_env],
+        [delete_env, defer_delete_env],
+    )
 
     from tests.system.utils.watcher import watcher