Add possibility to run DAGs from system tests and see DAGs logs (#17868)

apache · Aug 31, 2021 · 5007806 · 5007806
1 parent 9befaee
commit 5007806
Show file tree

Hide file tree

Showing 99 changed files with 264 additions and 89 deletions.
diff --git a/airflow/providers/google/cloud/example_dags/example_automl_tables.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_azure_fileshare_to_gcs.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_dts.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_operations.py b/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,

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_queries.py b/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},

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_sensors.py b/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},

diff --git a/airflow/providers/google/cloud/example_dags/example_bigquery_to_bigquery.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_bigtable.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_cloud_build.py b/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]

diff --git a/airflow/providers/google/cloud/example_dags/example_cloud_memorystore.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_cloud_sql.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_cloud_sql_query.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_cloud_storage_transfer_service_gcp.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_compute.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_compute_igm.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_compute_ssh.py b/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]

diff --git a/airflow/providers/google/cloud/example_dags/example_datacatalog.py b/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(

diff --git a/airflow/providers/google/cloud/example_dags/example_dataflow.py b/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]

diff --git a/airflow/providers/google/cloud/example_dags/example_dataflow_flex_template.py b/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(

diff --git a/airflow/providers/google/cloud/example_dags/example_dataflow_sql.py b/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]

diff --git a/airflow/providers/google/cloud/example_dags/example_datafusion.py b/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]

diff --git a/airflow/providers/google/cloud/example_dags/example_dataprep.py b/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]

diff --git a/airflow/providers/google/cloud/example_dags/example_dataproc.py b/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",

diff --git a/airflow/providers/google/cloud/example_dags/example_datastore.py b/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]

diff --git a/airflow/providers/google/cloud/example_dags/example_dlp.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_facebook_ads_to_gcs.py b/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:
 

diff --git a/airflow/providers/google/cloud/example_dags/example_functions.py b/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:

diff --git a/airflow/providers/google/cloud/example_dags/example_gcs.py b/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(

diff --git a/airflow/providers/google/cloud/example_dags/example_gcs_timespan_file_transform.py b/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:
 

diff --git a/airflow/providers/google/cloud/example_dags/example_gcs_to_bigquery.py b/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'],
 )
 

diff --git a/airflow/providers/google/cloud/example_dags/example_gcs_to_gcs.py b/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(