[GitHub] [airflow] bhirsz commented on a diff in pull request #22808: Add example DAG for demonstrating usage of GCS sensors

Tue, 12 Apr 2022 00:34:58 -0700 


bhirsz commented on code in PR #22808:
URL: https://github.com/apache/airflow/pull/22808#discussion_r848080946


##########
airflow/providers/google/cloud/example_dags/example_gcs.py:
##########
@@ -37,35 +37,44 @@
 from airflow.providers.google.cloud.sensors.gcs import (
     GCSObjectExistenceSensor,
     GCSObjectsWithPrefixExistenceSensor,
+    GCSObjectUpdateSensor,
+    GCSUploadSessionCompleteSensor,
 )
 from airflow.providers.google.cloud.transfers.gcs_to_gcs import 
GCSToGCSOperator
 from airflow.providers.google.cloud.transfers.gcs_to_local import 
GCSToLocalFilesystemOperator
 from airflow.providers.google.cloud.transfers.local_to_gcs import 
LocalFilesystemToGCSOperator
 
 START_DATE = datetime(2021, 1, 1)
 
-PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-id")
-BUCKET_1 = os.environ.get("GCP_GCS_BUCKET_1", "test-gcs-example-bucket")
-GCS_ACL_ENTITY = os.environ.get("GCS_ACL_ENTITY", "allUsers")
+PROJECT_ID = os.getenv("GCP_PROJECT_ID", "example-id")
+BUCKET_1 = os.getenv("GCP_GCS_BUCKET_1", "test-gcs-example-bucket")
+GCS_ACL_ENTITY = os.getenv("GCS_ACL_ENTITY", "allUsers")
 GCS_ACL_BUCKET_ROLE = "OWNER"
 GCS_ACL_OBJECT_ROLE = "OWNER"
 
-BUCKET_2 = os.environ.get("GCP_GCS_BUCKET_2", "test-gcs-example-bucket-2")
+BUCKET_2 = os.getenv("GCP_GCS_BUCKET_2", "test-gcs-example-bucket-2")
 
 temp_dir_path = gettempdir()
-PATH_TO_TRANSFORM_SCRIPT = os.environ.get(
+PATH_TO_TRANSFORM_SCRIPT = os.getenv(
     "GCP_GCS_PATH_TO_TRANSFORM_SCRIPT", os.path.join(temp_dir_path, 
"transform_script.py")
 )
-PATH_TO_UPLOAD_FILE = os.environ.get(
+PATH_TO_UPLOAD_FILE = os.getenv(
     "GCP_GCS_PATH_TO_UPLOAD_FILE", os.path.join(temp_dir_path, 
"test-gcs-example-upload.txt")
 )
-PATH_TO_UPLOAD_FILE_PREFIX = 
os.environ.get("GCP_GCS_PATH_TO_UPLOAD_FILE_PREFIX", "test-gcs-")
-PATH_TO_SAVED_FILE = os.environ.get(
+PATH_TO_UPLOAD_FILE_PREFIX = os.getenv("GCP_GCS_PATH_TO_UPLOAD_FILE_PREFIX", 
"test-gcs-")
+PATH_TO_SAVED_FILE = os.getenv(
     "GCP_GCS_PATH_TO_SAVED_FILE", os.path.join(temp_dir_path, 
"test-gcs-example-download.txt")
 )
 
 BUCKET_FILE_LOCATION = PATH_TO_UPLOAD_FILE.rpartition("/")[-1]
 
+# Upload 'test-gcs-manual-example-upload.txt' manually in the <BUCKET_1> after 
triggering the DAG.

Review Comment:
   It's tricky in this case, indeed. How about starting sensors in parallel 
with upload_file? 
   
   Example:
   ```
       chain(
           # TEST SETUP
           create_bucket,
           upload_file,
           # TEST BODY
           [gcs_object_exists, gcs_object_with_prefix_exists],
           # TEST TEARDOWN
           delete_bucket,
       )
       chain(
           create_bucket,
           # TEST BODY
           [gcs_upload_session_complete, gcs_update_object_exists],
           delete_bucket
       )
   ```
   
   
![image](https://user-images.githubusercontent.com/8532066/162905393-61247c98-9661-40b3-97db-9a2f2aeb660c.png)
   
   We're starting the sensors and in meantime we're uploading the file - and 
sensors detect it:
   
![image](https://user-images.githubusercontent.com/8532066/162905681-eab7e6a1-967b-43ac-9582-aac5e21dea16.png)
   
   
![image](https://user-images.githubusercontent.com/8532066/162905856-f8a093b6-85cf-46d3-b2f6-461e2dbbe51a.png)
   
   I'm not sure though if it will not be flaky in some cases - running this in 
CI will show. I will update my PR with this change.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to