Add possibility to run DAGs from system tests and see DAGs logs

airflow/providers/google/cloud/example_dags/example_automl_tables.py

@@ -84,7 +84,7 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 # Example DAG to create dataset, train model_id and deploy it.
 with models.DAG(
     "example_create_and_deploy",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     user_defined_macros={
         "get_target_column_spec": get_target_column_spec,
@@ -196,7 +196,7 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 # Example DAG for AutoML datasets operations
 with models.DAG(
     "example_automl_dataset",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     user_defined_macros={"extract_object_id": extract_object_id},
 ) as example_dag:
@@ -264,7 +264,7 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 
 with models.DAG(
     "example_gcp_get_deploy",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=["example"],
 ) as get_deploy_dag:
@@ -289,7 +289,7 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
 
 with models.DAG(
     "example_gcp_predict",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=["example"],
 ) as predict_dag:

airflow/providers/google/cloud/example_dags/example_azure_fileshare_to_gcs.py

@@ -36,7 +36,7 @@
         'retries': 1,
         'retry_delay': timedelta(minutes=5),
     },
-    schedule_interval=None,
+    schedule_interval='@once',
     start_date=datetime(2018, 11, 1),
     tags=['example'],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_bigquery_dts.py

@@ -64,7 +64,7 @@
 
 with models.DAG(
     "example_gcp_bigquery_dts",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=['example'],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_bigquery_operations.py

@@ -57,7 +57,7 @@
 
 with models.DAG(
     "example_bigquery_operations",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=["example"],
 ) as dag:
@@ -238,10 +238,10 @@
 
 with models.DAG(
     "example_bigquery_operations_location",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=["example"],
-):
+) as dag_with_location:
     create_dataset_with_location = BigQueryCreateEmptyDatasetOperator(
         task_id="create_dataset_with_location",
         dataset_id=LOCATION_DATASET_NAME,

airflow/providers/google/cloud/example_dags/example_bigquery_queries.py

@@ -65,7 +65,7 @@
 
     with models.DAG(
         dag_id,
-        schedule_interval=None,  # Override to match your needs
+        schedule_interval='@once',  # Override to match your needs
         start_date=days_ago(1),
         tags=["example"],
         user_defined_macros={"DATASET": DATASET_NAME, "TABLE": TABLE_1},

airflow/providers/google/cloud/example_dags/example_bigquery_sensors.py

@@ -54,7 +54,7 @@
 
 with models.DAG(
     dag_id,
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=["example"],
     user_defined_macros={"DATASET": DATASET_NAME, "TABLE": TABLE_NAME},

airflow/providers/google/cloud/example_dags/example_bigquery_to_bigquery.py

@@ -37,7 +37,7 @@
 
 with models.DAG(
     "example_bigquery_to_bigquery",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=["example"],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_bigtable.py

@@ -80,7 +80,7 @@
 
 with models.DAG(
     'example_gcp_bigtable_operators',
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=['example'],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_cloud_build.py

@@ -77,7 +77,7 @@
 with models.DAG(
     "example_gcp_cloud_build",
     default_args=dict(start_date=dates.days_ago(1)),
-    schedule_interval=None,
+    schedule_interval='@once',
     tags=['example'],
 ) as dag:
     # [START howto_operator_create_build_from_storage]

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.py

@@ -79,7 +79,7 @@
 
 with models.DAG(
     "gcp_cloud_memorystore_redis",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=dates.days_ago(1),
     tags=['example'],
 ) as dag:
@@ -255,7 +255,7 @@
 
 with models.DAG(
     "gcp_cloud_memorystore_memcached",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=dates.days_ago(1),
     tags=['example'],
 ) as dag_memcache:

airflow/providers/google/cloud/example_dags/example_cloud_sql.py

@@ -138,7 +138,7 @@
 
 with models.DAG(
     'example_gcp_sql',
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=['example'],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_cloud_sql_query.py

@@ -268,7 +268,7 @@ def get_absolute_path(path):
 
 with models.DAG(
     dag_id='example_gcp_sql_query',
-    schedule_interval=None,
+    schedule_interval='@once',
     start_date=days_ago(1),
     tags=['example'],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_cloud_storage_transfer_service_gcp.py

@@ -101,7 +101,7 @@
 
 with models.DAG(
     "example_gcp_transfer",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=["example"],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_compute.py

@@ -51,7 +51,7 @@
 
 with models.DAG(
     'example_gcp_compute',
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=['example'],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_compute_igm.py

@@ -91,7 +91,7 @@
 
 with models.DAG(
     'example_gcp_compute_igm',
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=['example'],
 ) as dag:

airflow/providers/google/cloud/example_dags/example_compute_ssh.py

@@ -31,7 +31,7 @@
 with models.DAG(
     'example_compute_ssh',
     default_args=dict(start_date=dates.days_ago(1)),
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     tags=['example'],
 ) as dag:
     # # [START howto_execute_command_on_remote1]

airflow/providers/google/cloud/example_dags/example_datacatalog.py

@@ -58,7 +58,7 @@
 FIELD_NAME_2 = "second"
 FIELD_NAME_3 = "first-rename"
 
-with models.DAG("example_gcp_datacatalog", start_date=days_ago(1), schedule_interval=None) as dag:
+with models.DAG("example_gcp_datacatalog", schedule_interval='@once', start_date=days_ago(1)) as dag:
     # Create
     # [START howto_operator_gcp_datacatalog_create_entry_group]
     create_entry_group = CloudDataCatalogCreateEntryGroupOperator(

airflow/providers/google/cloud/example_dags/example_dataflow.py

@@ -62,7 +62,7 @@
 
 with models.DAG(
     "example_gcp_dataflow_native_java",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=['example'],
 ) as dag_native_java:
@@ -111,7 +111,7 @@
     "example_gcp_dataflow_native_python",
     default_args=default_args,
     start_date=days_ago(1),
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     tags=['example'],
 ) as dag_native_python:
 
@@ -146,7 +146,7 @@
     "example_gcp_dataflow_native_python_async",
     default_args=default_args,
     start_date=days_ago(1),
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     tags=['example'],
 ) as dag_native_python_async:
     # [START howto_operator_start_python_job_async]
@@ -239,7 +239,7 @@ def check_autoscaling_event(autoscaling_events: List[dict]) -> bool:
     "example_gcp_dataflow_template",
     default_args=default_args,
     start_date=days_ago(1),
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     tags=['example'],
 ) as dag_template:
     # [START howto_operator_start_template_job]

airflow/providers/google/cloud/example_dags/example_dataflow_flex_template.py

@@ -46,7 +46,7 @@
 with models.DAG(
     dag_id="example_gcp_dataflow_flex_template_java",
     start_date=days_ago(1),
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
 ) as dag_flex_template:
     # [START howto_operator_start_template_job]
     start_flex_template = DataflowStartFlexTemplateOperator(

airflow/providers/google/cloud/example_dags/example_dataflow_sql.py

@@ -37,7 +37,7 @@
 with models.DAG(
     dag_id="example_gcp_dataflow_sql",
     start_date=days_ago(1),
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     tags=['example'],
 ) as dag_sql:
     # [START howto_operator_start_sql_job]

airflow/providers/google/cloud/example_dags/example_datafusion.py

@@ -147,7 +147,7 @@
 
 with models.DAG(
     "example_data_fusion",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=dates.days_ago(1),
 ) as dag:
     # [START howto_cloud_data_fusion_create_instance_operator]

airflow/providers/google/cloud/example_dags/example_dataprep.py

@@ -52,7 +52,7 @@
 
 with models.DAG(
     "example_dataprep",
-    schedule_interval=None,
+    schedule_interval='@once',
     start_date=dates.days_ago(1),  # Override to match your needs
 ) as dag:
     # [START how_to_dataprep_run_job_group_operator]

airflow/providers/google/cloud/example_dags/example_dataproc.py

@@ -151,7 +151,7 @@
 }
 
 
-with models.DAG("example_gcp_dataproc", start_date=days_ago(1), schedule_interval=None) as dag:
+with models.DAG("example_gcp_dataproc", schedule_interval='@once', start_date=days_ago(1)) as dag:
     # [START how_to_cloud_dataproc_create_cluster_operator]
     create_cluster = DataprocCreateClusterOperator(
         task_id="create_cluster",

airflow/providers/google/cloud/example_dags/example_datastore.py

@@ -42,7 +42,7 @@
 
 with models.DAG(
     "example_gcp_datastore",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=dates.days_ago(1),
     tags=["example"],
 ) as dag:
@@ -83,7 +83,7 @@
 with models.DAG(
     "example_gcp_datastore_operations",
     start_date=dates.days_ago(1),
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     tags=["example"],
 ) as dag2:
     # [START how_to_allocate_ids]

airflow/providers/google/cloud/example_dags/example_dlp.py

@@ -61,7 +61,7 @@
 
 with models.DAG(
     "example_gcp_dlp",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
     tags=['example'],
 ) as dag1:
@@ -110,7 +110,7 @@
 
 with models.DAG(
     "example_gcp_dlp_info_types",
-    schedule_interval=None,
+    schedule_interval='@once',
     start_date=days_ago(1),
     tags=["example", "dlp", "info-types"],
 ) as dag2:
@@ -152,7 +152,7 @@
 TRIGGER_ID = "example_trigger"
 
 with models.DAG(
-    "example_gcp_dlp_job", schedule_interval=None, start_date=days_ago(1), tags=["example", "dlp_job"]
+    "example_gcp_dlp_job", schedule_interval='@once', start_date=days_ago(1), tags=["example", "dlp_job"]
 ) as dag3:  # [START howto_operator_dlp_create_job_trigger]
     create_trigger = CloudDLPCreateJobTriggerOperator(
         project_id=GCP_PROJECT,
@@ -195,7 +195,7 @@
 
 with models.DAG(
     "example_gcp_dlp_deidentify_content",
-    schedule_interval=None,
+    schedule_interval='@once',
     start_date=days_ago(1),
     tags=["example", "dlp", "deidentify"],
 ) as dag4:

airflow/providers/google/cloud/example_dags/example_facebook_ads_to_gcs.py

@@ -56,7 +56,7 @@
 
 with models.DAG(
     "example_facebook_ads_to_gcs",
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=days_ago(1),
 ) as dag:
 

airflow/providers/google/cloud/example_dags/example_functions.py

@@ -96,7 +96,7 @@
 with models.DAG(
     'example_gcp_function',
     default_args=default_args,
-    schedule_interval=None,  # Override to match your needs
+    schedule_interval='@once',  # Override to match your needs
     start_date=dates.days_ago(1),
     tags=['example'],
 ) as dag:

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

@@ -60,7 +60,7 @@
 with models.DAG(
     "example_gcs",
     start_date=days_ago(1),
-    schedule_interval=None,
+    schedule_interval='@once',
     tags=['example'],
 ) as dag:
     create_bucket1 = GCSCreateBucketOperator(
@@ -160,7 +160,7 @@
 with models.DAG(
     "example_gcs_sensors",
     start_date=days_ago(1),
-    schedule_interval=None,
+    schedule_interval='@once',
     tags=['example'],
 ) as dag2:
     create_bucket = GCSCreateBucketOperator(

airflow/providers/google/cloud/example_dags/example_gcs_timespan_file_transform.py

@@ -41,7 +41,7 @@
 with models.DAG(
     "example_gcs_timespan_file_transform",
     start_date=days_ago(1),
-    schedule_interval=None,
+    schedule_interval='@once',
     tags=['example'],
 ) as dag:
 

airflow/providers/google/cloud/example_dags/example_gcs_to_bigquery.py

@@ -36,7 +36,7 @@
 dag = models.DAG(
     dag_id='example_gcs_to_bigquery_operator',
     start_date=days_ago(2),
-    schedule_interval=None,
+    schedule_interval='@once',
     tags=['example'],
 )
 

airflow/providers/google/cloud/example_dags/example_gcs_to_gcs.py

@@ -39,7 +39,7 @@
 OBJECT_2 = os.environ.get("GCP_GCS_OBJECT_2", "test-gcs-to-gcs-2")
 
 with models.DAG(
-    "example_gcs_to_gcs", start_date=days_ago(1), schedule_interval=None, tags=['example']
+    "example_gcs_to_gcs", schedule_interval='@once', start_date=days_ago(1), tags=['example']
 ) as dag:
     # [START howto_synch_bucket]
     sync_bucket = GCSSynchronizeBucketsOperator(