Fix mypy changes and add env in example dag

Author: rajaths010494

astronomer/providers/google/cloud/example_dags/example_gcs.py

@@ -37,31 +37,34 @@
     GCSUploadSessionCompleteSensorAsync,
 )
 
-START_DATE = datetime(2021, 1, 1)
-
 PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "astronomer-airflow-providers")
-BUCKET_1 = "test-gcs-example-bucket"
+BUCKET_1 = os.environ.get("GCP_TEST_BUCKET", "test-gcs-example-bucket")
 PATH_TO_UPLOAD_FILE = "dags/example_gcs.py"
 PATH_TO_UPLOAD_FILE_PREFIX = "example_"
 
 BUCKET_FILE_LOCATION = "example_gcs.py"
 
 with models.DAG(
     "example_async_gcs_sensors",
-    start_date=START_DATE,
+    start_date=datetime(2021, 1, 1),
     catchup=False,
     schedule_interval='@once',
     tags=['example'],
 ) as dag:
+
+    # [START howto_create_bucket_task]
     create_bucket = GCSCreateBucketOperator(
         task_id="create_bucket", bucket_name=BUCKET_1, project_id=PROJECT_ID
     )
+    # [END howto_create_bucket_task]
+    # [START howto_upload_file_task]
     upload_file = LocalFilesystemToGCSOperator(
         task_id="upload_file",
         src=PATH_TO_UPLOAD_FILE,
         dst=BUCKET_FILE_LOCATION,
         bucket=BUCKET_1,
     )
+    # [END howto_upload_file_task]
     # [START howto_sensor_object_exists_task]
     gcs_object_exists = GCSObjectExistenceSensorAsync(
         bucket=BUCKET_1,
@@ -76,7 +79,7 @@
         task_id="gcs_object_with_prefix_exists_task",
     )
     # [END howto_sensor_object_with_prefix_exists_task]
-
+    # [START howto_sensor_gcs_upload_session_complete_task]
     gcs_upload_session_complete = GCSUploadSessionCompleteSensorAsync(
         bucket=BUCKET_1,
         prefix=PATH_TO_UPLOAD_FILE_PREFIX,
@@ -86,8 +89,10 @@
         previous_objects=set(),
         task_id="gcs_upload_session_complete_task",
     )
-
+    # [END howto_sensor_gcs_upload_session_complete_task]
+    # [START howto_delete_buckettask]
     delete_bucket = GCSDeleteBucketOperator(task_id="delete_bucket", bucket_name=BUCKET_1)
+    # [END howto_delete_buckettask]
 
     (
         create_bucket

astronomer/providers/google/cloud/sensors/gcs.py

@@ -1,6 +1,6 @@
 """This module contains Google Cloud Storage sensors."""
 
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional, Sequence, Union
 
 from airflow.exceptions import AirflowException
 from airflow.models.baseoperator import BaseOperator
@@ -115,7 +115,7 @@ def execute(self, context: "Context") -> None:
 
     def execute_complete(
         self, context: "Context", event: Optional[Dict[Any, Any]] = None
-    ) -> List[str]:  # pylint: disable=unused-argument
+    ) -> Any:  # pylint: disable=unused-argument
         """
         Callback for when the trigger fires - returns immediately.
         Relies on trigger to throw an exception, otherwise it assumes execution was
@@ -155,7 +155,7 @@ def execute(self, context: "Context") -> None:
 
     def execute_complete(
         self, context: "Context", event: Optional[Dict[Any, Any]] = None
-    ) -> None:  # pylint: disable=unused-argument
+    ) -> Optional[str]:  # pylint: disable=unused-argument
         """
         Callback for when the trigger fires - returns immediately.
         Relies on trigger to throw an exception, otherwise it assumes execution was

astronomer/providers/google/cloud/triggers/gcs.py

@@ -121,7 +121,7 @@ def __init__(
 
     def serialize(self) -> Tuple[str, Dict[str, Any]]:
         """
-        Serializes GCSBlobTrigger arguments and classpath.
+        Serializes GCSPrefixBlobTrigger arguments and classpath.
         """
         return (
             "astronomer.providers.google.cloud.triggers.gcs.GCSPrefixBlobTrigger",
@@ -162,8 +162,8 @@ async def _list_blobs_with_prefix(self, hook: GCSHookAsync, bucket_name: str, pr
         :param prefix: The prefix of the blob_names to match in the Google cloud
             storage bucket.
         """
-        async with Session() as s:
-            client = await hook.get_storage_client(s)
+        async with Session() as session:
+            client = await hook.get_storage_client(session)
             bucket = client.get_bucket(bucket_name)
             object_response = await bucket.list_blobs(prefix=prefix)
             return object_response