Airflow implementation of import automation workflow

[vish-cs]

Implements the Apache Airflow (Cloud Composer) orchestration pipeline for Data Commons data imports, replacing the legacy Cloud Workflows implementation.

Key Changes

• Dynamic DAG Factory (import_dags_factory.py & build_manifest_catalog.py): Scans repository manifest.json files at build time and dynamically registers an independent Airflow DAG for each import specification (plus a generic import_automation_workflow fallback DAG). All DAGs are created in a disabled (is_paused_upon_creation=True) state with skipProdIngestion=True by default.
• 4-Stage Pipeline (import_automation_workflow.py):
  1. Import Execution (run_import_job): Triggers the dc-import-executor Cloud Batch job (with inline validation and differ disabled).
  2. Standalone Validation & Differ (run_validation_job): Invokes the Cloud Run v2 Job (import-validator-job) with a 2-hour timeout to run BigQuery differ and validation checks.
  3. Staging Ingestion & Golden Verification Gate (golden_verification.py): Triggers Spanner staging ingestion, runs automated golden diff verification via Cloud Build (ingestion-golden-verification.cloudbuild.yaml), and pauses for human approval (HumanApprovalSensor) if diffs are detected.
  4. Production Ingestion (ingest_prod): Promotes validated imports to production Spanner when enabled.
• CI/CD & E2E Testing (cloudbuild.workflow.yaml & e2e_dag_test.py): Automates catalog compilation, GCS DAG synchronization to Cloud Composer, and live End-to-End DAG execution verification via the Airflow REST API.

TAG=agy
CONV=fd554db8-40c1-4b71-b86a-83b4aa199d65

[gemini-code-assist]

Code Review

This pull request introduces an automated workflow for Data Commons imports using Apache Airflow and Cloud Composer. It includes a catalog builder script, a dynamic DAG factory, and a core automation DAG that orchestrates Cloud Batch jobs and Spanner ingestion. The review identifies several critical anti-patterns, including unsafe path manipulation, global environment variable modification, synchronous operator execution within tasks, and blocking sleep loops. Actionable feedback was provided to improve path handling, enforce strict duplicate ID management, and implement robust retry mechanisms for network requests.