Cleanup dynamic start_date use for miscellaneous Google example DAGs (

#19400)

* Cleanup dynamic start_date use for misc Google example DAGs
* Updating task dependencies based on XComArgs

airflow/providers/google/ads/example_dags/example_ads.py

@@ -19,11 +19,11 @@
 Example Airflow DAG that shows how to use GoogleAdsToGcsOperator.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.ads.operators.ads import GoogleAdsListAccountsOperator
 from airflow.providers.google.ads.transfers.ads_to_gcs import GoogleAdsToGcsOperator
-from airflow.utils import dates
 
 # [START howto_google_ads_env_variables]
 CLIENT_IDS = ["1111111111", "2222222222"]
@@ -67,7 +67,8 @@
 with models.DAG(
     "example_google_ads",
     schedule_interval=None,  # Override to match your needs
-    start_date=dates.days_ago(1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
 ) as dag:
     # [START howto_google_ads_to_gcs_operator]
     run_operator = GoogleAdsToGcsOperator(

airflow/providers/google/firebase/example_dags/example_firestore.py

@@ -44,17 +44,18 @@
 """
 
 import os
+from datetime import datetime
 from urllib.parse import urlparse
 
 from airflow import models
+from airflow.models.baseoperator import chain
 from airflow.providers.google.cloud.operators.bigquery import (
     BigQueryCreateEmptyDatasetOperator,
     BigQueryCreateExternalTableOperator,
     BigQueryDeleteDatasetOperator,
     BigQueryInsertJobOperator,
 )
 from airflow.providers.google.firebase.operators.firestore import CloudFirestoreExportDatabaseOperator
-from airflow.utils import dates
 
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-gcp-project")
 FIRESTORE_PROJECT_ID = os.environ.get("G_FIRESTORE_PROJECT_ID", "example-firebase-project")
@@ -72,8 +73,9 @@
 
 with models.DAG(
     "example_google_firestore",
-    default_args=dict(start_date=dates.days_ago(1)),
+    start_date=datetime(2021, 1, 1),
     schedule_interval='@once',
+    catchup=False,
     tags=["example"],
 ) as dag:
     # [START howto_operator_export_database_to_gcs]
@@ -134,10 +136,12 @@
         },
     )
 
-    # Firestore
-    export_database_to_gcs >> create_dataset
-
-    # BigQuery
-    create_dataset >> create_external_table_multiple_types
-    create_external_table_multiple_types >> read_data_from_gcs_multiple_types
-    read_data_from_gcs_multiple_types >> delete_dataset
+    chain(
+        # Firestore
+        export_database_to_gcs,
+        # BigQuery
+        create_dataset,
+        create_external_table_multiple_types,
+        read_data_from_gcs_multiple_types,
+        delete_dataset,
+    )

airflow/providers/google/leveldb/example_dags/example_leveldb.py

@@ -19,29 +19,27 @@
 Example use of LevelDB operators.
 """
 
+from datetime import datetime
+
 from airflow import models
 from airflow.providers.google.leveldb.operators.leveldb import LevelDBOperator
-from airflow.utils.dates import days_ago
 
 with models.DAG(
     'example_leveldb',
-    start_date=days_ago(2),
+    start_date=datetime(2021, 1, 1),
     schedule_interval='@once',
+    catchup=False,
     tags=['example'],
 ) as dag:
     # [START howto_operator_leveldb_get_key]
-    get_key_leveldb_task = LevelDBOperator(
-        task_id='get_key_leveldb', leveldb_conn_id='leveldb_default', command='get', key=b'key', dag=dag
-    )
+    get_key_leveldb_task = LevelDBOperator(task_id='get_key_leveldb', command='get', key=b'key')
     # [END howto_operator_leveldb_get_key]
     # [START howto_operator_leveldb_put_key]
     put_key_leveldb_task = LevelDBOperator(
         task_id='put_key_leveldb',
-        leveldb_conn_id='leveldb_default',
         command='put',
         key=b'another_key',
         value=b'another_value',
-        dag=dag,
     )
     # [END howto_operator_leveldb_put_key]
     get_key_leveldb_task >> put_key_leveldb_task

airflow/providers/google/marketing_platform/example_dags/example_analytics.py

@@ -18,6 +18,7 @@
 Example Airflow DAG that shows how to use Google Analytics 360.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.marketing_platform.operators.analytics import (
@@ -28,7 +29,6 @@
     GoogleAnalyticsModifyFileHeadersDataImportOperator,
     GoogleAnalyticsRetrieveAdsLinksListOperator,
 )
-from airflow.utils import dates
 
 ACCOUNT_ID = os.environ.get("GA_ACCOUNT_ID", "123456789")
 
@@ -41,7 +41,8 @@
 with models.DAG(
     "example_google_analytics",
     schedule_interval='@once',  # Override to match your needs,
-    start_date=dates.days_ago(1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
 ) as dag:
     # [START howto_marketing_platform_list_accounts_operator]
     list_account = GoogleAnalyticsListAccountsOperator(task_id="list_account")

airflow/providers/google/marketing_platform/example_dags/example_campaign_manager.py

@@ -20,6 +20,7 @@
 """
 import os
 import time
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.marketing_platform.operators.campaign_manager import (
@@ -33,7 +34,6 @@
 from airflow.providers.google.marketing_platform.sensors.campaign_manager import (
     GoogleCampaignManagerReportSensor,
 )
-from airflow.utils import dates
 from airflow.utils.state import State
 
 PROFILE_ID = os.environ.get("MARKETING_PROFILE_ID", "123456789")
@@ -87,20 +87,21 @@
 with models.DAG(
     "example_campaign_manager",
     schedule_interval='@once',  # Override to match your needs,
-    start_date=dates.days_ago(1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
 ) as dag:
     # [START howto_campaign_manager_insert_report_operator]
     create_report = GoogleCampaignManagerInsertReportOperator(
         profile_id=PROFILE_ID, report=REPORT, task_id="create_report"
     )
-    report_id = "{{ task_instance.xcom_pull('create_report')['id'] }}"
+    report_id = create_report.output["report_id"]
     # [END howto_campaign_manager_insert_report_operator]
 
     # [START howto_campaign_manager_run_report_operator]
     run_report = GoogleCampaignManagerRunReportOperator(
         profile_id=PROFILE_ID, report_id=report_id, task_id="run_report"
     )
-    file_id = "{{ task_instance.xcom_pull('run_report')['id'] }}"
+    file_id = run_report.output["file_id"]
     # [END howto_campaign_manager_run_report_operator]
 
     # [START howto_campaign_manager_wait_for_operation]
@@ -129,7 +130,14 @@
     )
     # [END howto_campaign_manager_delete_report_operator]
 
-    create_report >> run_report >> wait_for_report >> get_report >> delete_report
+    wait_for_report >> get_report >> delete_report
+
+    # Task dependencies created via `XComArgs`:
+    #   create_report >> run_report
+    #   create_report >> wait_for_report
+    #   create_report >> get_report
+    #   run_report >> get_report
+    #   run_report >> wait_for_report
 
     # [START howto_campaign_manager_insert_conversions]
     insert_conversion = GoogleCampaignManagerBatchInsertConversionsOperator(
@@ -156,6 +164,7 @@
 
     insert_conversion >> update_conversion
 
+
 if __name__ == "__main__":
     dag.clear(dag_run_state=State.NONE)
     dag.run()

airflow/providers/google/marketing_platform/example_dags/example_display_video.py

@@ -19,6 +19,7 @@
 Example Airflow DAG that shows how to use DisplayVideo.
 """
 import os
+from datetime import datetime
 from typing import Dict
 
 from airflow import models
@@ -38,7 +39,6 @@
     GoogleDisplayVideo360GetSDFDownloadOperationSensor,
     GoogleDisplayVideo360ReportSensor,
 )
-from airflow.utils import dates
 
 # [START howto_display_video_env_variables]
 BUCKET = os.environ.get("GMP_DISPLAY_VIDEO_BUCKET", "gs://INVALID BUCKET NAME")
@@ -82,10 +82,13 @@
 DOWNLOAD_LINE_ITEMS_REQUEST: Dict = {"filterType": ADVERTISER_ID, "format": "CSV", "fileSpec": "EWF"}
 # [END howto_display_video_env_variables]
 
+START_DATE = datetime(2021, 1, 1)
+
 with models.DAG(
     "example_display_video",
     schedule_interval='@once',  # Override to match your needs,
-    start_date=dates.days_ago(1),
+    start_date=START_DATE,
+    catchup=False,
 ) as dag1:
     # [START howto_google_display_video_createquery_report_operator]
     create_report = GoogleDisplayVideo360CreateReportOperator(body=REPORT, task_id="create_report")
@@ -127,7 +130,8 @@
 with models.DAG(
     "example_display_video_misc",
     schedule_interval='@once',  # Override to match your needs,
-    start_date=dates.days_ago(1),
+    start_date=START_DATE,
+    catchup=False,
 ) as dag2:
     # [START howto_google_display_video_upload_multiple_entity_read_files_to_big_query]
     upload_erf_to_bq = GCSToBigQueryOperator(
@@ -160,7 +164,8 @@
 with models.DAG(
     "example_display_video_sdf",
     schedule_interval='@once',  # Override to match your needs,
-    start_date=dates.days_ago(1),
+    start_date=START_DATE,
+    catchup=False,
 ) as dag3:
     # [START howto_google_display_video_create_sdf_download_task_operator]
     create_sdf_download_task = GoogleDisplayVideo360CreateSDFDownloadTaskOperator(

airflow/providers/google/marketing_platform/example_dags/example_search_ads.py

@@ -19,14 +19,14 @@
 Example Airflow DAG that shows how to use SearchAds.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.marketing_platform.operators.search_ads import (
     GoogleSearchAdsDownloadReportOperator,
     GoogleSearchAdsInsertReportOperator,
 )
 from airflow.providers.google.marketing_platform.sensors.search_ads import GoogleSearchAdsReportSensor
-from airflow.utils import dates
 
 # [START howto_search_ads_env_variables]
 AGENCY_ID = os.environ.get("GMP_AGENCY_ID")
@@ -47,7 +47,8 @@
 with models.DAG(
     "example_search_ads",
     schedule_interval='@once',  # Override to match your needs,
-    start_date=dates.days_ago(1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
 ) as dag:
     # [START howto_search_ads_generate_report_operator]
     generate_report = GoogleSearchAdsInsertReportOperator(report=REPORT, task_id="generate_report")

airflow/providers/google/suite/example_dags/example_gcs_to_gdrive.py

@@ -19,17 +19,18 @@
 Example DAG using GoogleCloudStorageToGoogleDriveOperator.
 """
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.suite.transfers.gcs_to_gdrive import GCSToGoogleDriveOperator
-from airflow.utils.dates import days_ago
 
 GCS_TO_GDRIVE_BUCKET = os.environ.get("GCS_TO_DRIVE_BUCKET", "example-object")
 
 with models.DAG(
     "example_gcs_to_gdrive",
     schedule_interval=None,  # Override to match your needs,
-    start_date=days_ago(1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
     tags=['example'],
 ) as dag:
     # [START howto_operator_gcs_to_gdrive_copy_single_file]

airflow/providers/google/suite/example_dags/example_gcs_to_sheets.py

@@ -17,20 +17,21 @@
 # under the License.
 
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.providers.google.cloud.transfers.sheets_to_gcs import GoogleSheetsToGCSOperator
 from airflow.providers.google.suite.transfers.gcs_to_sheets import GCSToGoogleSheetsOperator
-from airflow.utils.dates import days_ago
 
 BUCKET = os.environ.get("GCP_GCS_BUCKET", "example-test-bucket3")
 SPREADSHEET_ID = os.environ.get("SPREADSHEET_ID", "example-spreadsheetID")
 NEW_SPREADSHEET_ID = os.environ.get("NEW_SPREADSHEET_ID", "1234567890qwerty")
 
 with models.DAG(
     "example_gcs_to_sheets",
-    start_date=days_ago(1),
+    start_date=datetime(2021, 1, 1),
     schedule_interval='@once',  # Override to match your needs
+    catchup=False,
     tags=["example"],
 ) as dag:
 

airflow/providers/google/suite/example_dags/example_sheets.py

@@ -17,13 +17,13 @@
 # under the License.
 
 import os
+from datetime import datetime
 
 from airflow import models
 from airflow.operators.bash import BashOperator
 from airflow.providers.google.cloud.transfers.sheets_to_gcs import GoogleSheetsToGCSOperator
 from airflow.providers.google.suite.operators.sheets import GoogleSheetsCreateSpreadsheetOperator
 from airflow.providers.google.suite.transfers.gcs_to_sheets import GCSToGoogleSheetsOperator
-from airflow.utils.dates import days_ago
 
 GCS_BUCKET = os.environ.get("SHEETS_GCS_BUCKET", "test28397ye")
 SPREADSHEET_ID = os.environ.get("SPREADSHEET_ID", "1234567890qwerty")
@@ -37,7 +37,8 @@
 with models.DAG(
     "example_sheets_gcs",
     schedule_interval='@once',  # Override to match your needs,
-    start_date=days_ago(1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
     tags=["example"],
 ) as dag:
     # [START upload_sheet_to_gcs]

airflow/providers/google/suite/example_dags/example_sql_to_sheets.py

@@ -16,17 +16,19 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from datetime import datetime
+
 from airflow import models
 from airflow.providers.google.suite.transfers.sql_to_sheets import SQLToGoogleSheetsOperator
-from airflow.utils.dates import days_ago
 
 SQL = "select 1 as my_col"
 NEW_SPREADSHEET_ID = "123"
 
 with models.DAG(
     "example_sql_to_sheets",
-    start_date=days_ago(1),
+    start_date=datetime(2021, 1, 1),
     schedule_interval=None,  # Override to match your needs
+    catchup=False,
     tags=["example"],
 ) as dag: