[fas89]

Every Airflow DAG forge-cli generates now loads on Airflow 3, and still on Airflow 2.7+. This is step 1 of the Command Center's move to Airflow 3: CC runs fluid export --engine airflow and ships its own Airflow image, and that image moves to 3.2.2 once a forge-cli release carries this.

What was broken

Airflow 3 removed two things the generators emitted, checked against apache/airflow 3.2.2:

• DAG(schedule_interval=...): the DAG takes only schedule, so the old keyword raises TypeError.
• airflow.utils.dates.days_ago: removed, so importing it raises ImportError.

Those two broke every DAG from:

• fluid export --engine airflow (runtimes/airflow_provider_actions.py);
• fluid generate schedule (schedulers/airflow);
• fluid init (cli/_init_dag_helpers.py);
• the AWS, GCP and Snowflake code generators;
• the Snowflake orchestration generator.

fluid scaffold-composer and fluid schedule-sync already passed schedule=. Like every other generator, though, they imported the operators from paths Airflow 3 has deprecated.

Loading every rendered DAG into a real DagBag found three more defects, each broken on Airflow 2.10 too:

• Snowflake: SQL tasks used SnowflakeOperator, which apache-airflow-providers-snowflake 6.0 removed.
• GCP: DAGs imported BigQueryCreateEmptyTableOperator, removed in google provider 17.0. They also imported the Dataflow operator unconditionally, and its module needs the apache-beam provider.
• AWS TaskFlow: the DAG wired its @task functions (load >> publish), which raises TypeError at parse.

What changed

• Shared helpers: codegen_utils.airflow_dag_import() and airflow_standard_import(name). Each emits a try/except that imports from the Airflow 3 home (airflow.sdk, apache-airflow-providers-standard) and falls back to the Airflow 2 path. That is the shape schedulers/airflow/fluid_apply.py already emitted. Every generator uses them, and passes schedule= with a fixed timezone-aware start_date and catchup=False.
• Snowflake: SQLExecuteQueryOperator (common-sql, installed by the snowflake provider) with conn_id, and hook_params carrying the warehouse and database. It keeps SnowflakeOperator's statement splitting and autocommit defaults.
• GCP:
  • create_table emits BigQueryCreateTableOperator where the provider has it (13+), else the old operator. The choice is made at runtime in the DAG.
  • The Dataflow import comes only with a <any>.dataflow.run_template task, the same test the dispatch uses.
  • The scheduler drops its unused empty-table import.
• AWS TaskFlow: dependencies wire each @task's result, and each operator as itself. Result names are allocated so they never collide with a task's variable.
• fluid init (security, pre-existing): orchestration.retries and retry_delay were spliced into the DAG source as text, so a contract value ran as Python at DAG-parse time. Both are read as bounded numbers now.
• templates/first-dag/README.md shows the new shape.
• CHANGELOG (Unreleased):
  • Upgrade notes: the Snowflake and BigQuery operator switch, and that the new operators template conn_id/hook_params/table_resource.
  • Fixed: the four defects.

Tests

tests/test_airflow3_dag_compat.py renders every generator, 15 renders in all, and fails on any of:

• schedule_interval;
• days_ago;
• an Airflow-2-only import (from x import y, import x, from airflow.models import DAG, …) outside an except ImportError fallback whose try imports the same names from their Airflow 3 home;
• SnowflakeOperator;
• a @task function wired as a task;
• a Dataflow import that doesn't follow the dispatch.

Against main, all 13 original renders fail. Two new injection tests cover a hostile GCP create_table schema and hostile fluid init retries. Three existing tests that pinned the old output now pin the new one.

Borrow before build

Checked locally:
- schedulers/airflow/fluid_apply.py already emitted the Airflow 2/3 import shape → generalised into codegen_utils
- tests/cli/_schedule_dag_fixtures.py: stub-Airflow loader; tests/test_airflow_codegen_injection.py: assert_inert
Searched / fetched (gh api, apache/airflow @ 3.2.2):
- airflow-core/src/airflow/operators/__init__.py, sensors/__init__.py → the deprecation map (core operators → providers.standard)
- airflow-core/src/airflow/utils/dates.py → no days_ago; task-sdk definitions/dag.py → `schedule` only
- providers google / snowflake changelogs (via installed versions) → BigQueryCreateTableOperator 13+, empty-table removed 17.0; SnowflakeOperator removed in snowflake 6.0
Reuse strategy: adapt the in-repo pattern; no new dependency

Validation

Every rendered DAG was loaded into a real DagBag.

Airflow	Setup	Result
3.2.2	constraints-3.2.2; providers amazon, google 22.0.0, snowflake, apache-beam	15 / 15 load
3.3.2 (latest)	the official apache/airflow:3.3.2-python3.11 image (amazon 9.36, google 22.5, snowflake 6.17, standard 1.19), plus the apache-beam provider	17 / 17 load: the 15 renders and two DAGs with tasks named task, dag, AthenaOperator, S3CreateBucketOperator and PythonOperator (classic and TaskFlow), with the right edges
2.10.5	constraints-2.10.5; google 12.0.0, snowflake 6.0.0, no standard provider (the fallbacks used)	14 / 15. The Dataflow DAG needs the apache-beam provider, which would not build locally.

• Dependency edges checked in the loaded DAGs (load >> publish), including an operator upstream of a @task.
• 1751 Airflow-related tests pass.
• ruff check fluid_build/ tests/ and black==24.10.0 --check fluid_build/ tests/ (CI's pins) are clean.
• tests/perf/test_startup_budget.py passes.

Review

An independent reviewer ran security review, world-class inspection and defect scan. Its own DagBag loads covered every generator branch on both versions, and it checked the compat shims against Airflow 3.0, 3.1 and 3.2.2 and providers-standard 1.0.

• Round 1: PASS-WITH-LOWS, fixed in dca4461:
  • a @task result could reuse another task's variable name;
  • the Dataflow import used a different test from the dispatch;
  • version facts;
  • the templating note;
  • test gaps;
  • the first-dag README;
  • the pre-existing fluid init injection.
• Round 2: PASS-WITH-LOWS, fixed in 24442d8:
  • OverflowError on retries: .inf;
  • the Jinja escape note for {% and {#;
  • a pre-existing collision: taskId: task shadowed the TaskFlow @task decorator. The AWS generators now reserve task and every operator class they import.
• Final re-check: PASS. A DAG with tasks named task, dag, AthenaOperator, S3CreateBucketOperator and PythonOperator loads on 3.2.2 and 2.10.5, classic and TaskFlow alike, with the right edges.

The security pass found no new injection path. Every new interpolation goes through py_str_literal, repr or sanitize_identifier.

Not in this PR

Pre-existing, filed separately:

• fluid init calls generate-airflow --output-dir, a flag that command doesn't accept, so it always uses the basic fallback DAG.
• Snowflake orchestration python tasks reference a callable that is never defined.
• examples/generated_dags/ (not shipped) still shows the old shape.

Release

The Command Center's Airflow 3 change needs a forge-cli release that carries this.

[github-actions]

📄 Documentation Reminder

This PR appears to be missing a documentation reference. Our docs live in a separate repo.

Please update the PR description with one of:

• Link a docs PR — check the "Docs PR linked" box and paste the URL
• Mark as no docs needed — check "No docs needed" with a justification
• Acknowledge docs TODO — check "Docs TODO" and create the docs PR before merge

See the Contributing Guide for details.