fix(airflow): every generated DAG loads on Airflow 3 and on Airflow 2.7+

CHANGELOG.md

@@ -64,6 +64,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - An environment overlay cannot turn a scheduled build continuous: overlays cannot
   delete the base's `cron` or `schedule`, which a continuous trigger refuses. Declare the
   trigger in each environment's overlay instead of in the base.
+- Generated Airflow DAGs run Snowflake SQL through `SQLExecuteQueryOperator`
+  (apache-airflow-providers-common-sql, which the snowflake provider installs) with
+  `conn_id`, because apache-airflow-providers-snowflake 6.0 removed `SnowflakeOperator`;
+  a DAG generated before needs regenerating to load. A GCP `create_table` task uses
+  `BigQueryCreateTableOperator` where the google provider has it (13+; 17.0 removed the
+  old one), and `BigQueryCreateEmptyTableOperator` before that.
+- Both new operators template more of their arguments than the ones they replace:
+  `SQLExecuteQueryOperator` renders `conn_id` and `hook_params` (the warehouse and
+  database), and `BigQueryCreateTableOperator` renders `table_resource` (the schema). A
+  literal `{{`, `{%` or `{#` in a warehouse or database name, or in a schema field such
+  as a description, is now rendered as Jinja when the task runs, and fails it. Escape it
+  as `{{ '{{' }}`, `{{ '{%' }}` or `{{ '{#' }}`, or wrap the value in `{% raw %}` ...
+  `{% endraw %}`.
 
 ### Added
 
@@ -149,6 +162,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   so an escape sequence in a contract or an error message can no longer clear the
   terminal or plant a link. JSON and JUnit output are unchanged.
 
+### Fixed
+
+- **Generated Airflow DAGs load on Airflow 3.** `fluid export --engine airflow`, `fluid
+  generate schedule`, `fluid init` and the AWS, GCP and Snowflake code generators emitted
+  `DAG(schedule_interval=...)` and `start_date=days_ago(1)`, which Airflow 3 removed, so
+  their DAGs failed to import. They now pass `schedule=` with a fixed `start_date` and
+  `catchup=False`. Every generator, `fluid scaffold-composer` and `fluid schedule-sync`
+  included, imports `DAG`, `dag`/`task` and the core operators and sensors from their
+  Airflow 3 homes (`airflow.sdk`, apache-airflow-providers-standard), falling back to the
+  Airflow 2 paths, so the same DAG loads on Airflow 2.7+ and 3.x without the deprecation
+  warnings Airflow 3 gives the old import paths.
+- The AWS TaskFlow DAG wired its `@task` functions (`load >> publish`), which raised
+  `TypeError` when Airflow parsed it; it now wires each task's result, and an operator
+  task as itself.
+- The GCP DAG imports the Dataflow operator only when a task uses it: its module needs
+  apache-airflow-providers-apache-beam, and the unused import stopped every GCP DAG from
+  loading where that provider was absent.
+- `fluid init`'s basic DAG spliced `orchestration.retries` and `retry_delay` into the DAG
+  source as text, so a contract value ran as Python when Airflow parsed the DAG. Both are
+  now read as numbers (retries 0 to 100, the delay in whole minutes), falling back to 3
+  and 5.
+
 ## [0.21.0] — 2026-10-19
 
 ### Upgrade notes

fluid_build/cli/_acquisition_stage_ext.py

@@ -59,6 +59,7 @@
 from fluid_build.build_runners._ids import (
     validate_identifier as _validate_identifier,
 )
+from fluid_build.providers.common.codegen_utils import airflow_dag_import, airflow_standard_import
 from fluid_build.util.contract import is_unscheduled_build
 
 LOG = logging.getLogger("fluid.acquire.stage_ext")
@@ -583,9 +584,10 @@ def _render_airflow_dag(
     contract_path = f"contracts/{contract_id}.fluid.yaml"
     return (
         '"""Auto-generated by fluid schedule-sync. Edit the contract, not this file."""\n\n'
-        "from datetime import datetime, timedelta\n"
-        "from airflow import DAG\n"
-        "from airflow.operators.bash import BashOperator\n\n"
+        "from datetime import datetime, timedelta\n\n"
+        + airflow_dag_import()
+        + airflow_standard_import("BashOperator")
+        + "\n"
         "default_args = {\n"
         '    "owner": "fluid",\n'
         '    "depends_on_past": False,\n'

fluid_build/cli/_init_dag_helpers.py

@@ -154,37 +154,49 @@ def create_basic_dag(project_dir: Path, contract: dict, logger):
 
     import re as _re
 
+    from fluid_build.providers.common.codegen_utils import (
+        airflow_dag_import,
+        airflow_standard_import,
+    )
+
     contract_name = contract.get("name", "my_product")
     orchestration = contract.get("orchestration", {})
     # Sanitize values to prevent code injection in generated Python.
     dag_id = _re.sub(r"[^a-zA-Z0-9_]", "_", contract_name)[:128]
     schedule = _re.sub(r"[^a-zA-Z0-9@*/, _-]", "", orchestration.get("schedule", "@daily"))[:64]
     contract_name = _re.sub(r"[^a-zA-Z0-9 _.-]", "_", contract_name)[:128]
+    # Numbers, never text: both are interpolated into the DAG's Python source, so a
+    # contract value such as `3, __import__('os').system('...')` would run at parse time.
+    try:
+        retries = max(0, min(int(orchestration.get("retries", 3)), 100))
+    except (TypeError, ValueError, OverflowError):  # OverflowError: `retries: .inf`
+        retries = 3
+    delay = _re.fullmatch(r"\s*(\d{1,4})\s*m?\s*", str(orchestration.get("retry_delay", "5m")))
+    retry_minutes = int(delay.group(1)) if delay else 5
 
     dag_content = f'''"""
 Airflow DAG for FLUID contract: {contract_name}
 Auto-generated by fluid init
 """
 
 from datetime import datetime, timedelta
-from airflow import DAG
-from airflow.operators.bash import BashOperator
 
+{airflow_dag_import()}{airflow_standard_import("BashOperator")}
 default_args = {{
     'owner': 'fluid',
     'depends_on_past': False,
     'start_date': datetime(2024, 1, 1),
     'email_on_failure': False,
     'email_on_retry': False,
-    'retries': {orchestration.get("retries", 3)},
-    'retry_delay': timedelta(minutes={orchestration.get("retry_delay", "5m").replace("m", "")}),
+    'retries': {retries},
+    'retry_delay': timedelta(minutes={retry_minutes}),
 }}
 
 with DAG(
     dag_id='{dag_id}',
     default_args=default_args,
     description='FLUID data product: {contract_name}',
-    schedule_interval='{schedule}',
+    schedule='{schedule}',
     catchup=False,
     tags=['fluid', 'data-product'],
 ) as dag:

fluid_build/cli/scaffold_composer.py

@@ -22,7 +22,12 @@
 from typing import Any, Dict, List
 
 from .._console import cprint
-from ..providers.common.codegen_utils import py_str_literal, sanitize_identifier
+from ..providers.common.codegen_utils import (
+    airflow_dag_import,
+    airflow_standard_import,
+    py_str_literal,
+    sanitize_identifier,
+)
 from ..util.contract import has_only_unscheduled_builds, is_unscheduled_build
 from ._acquisition_stage_ext import _validate_cron as _validate_cron_grammar
 from ._common import CLIError, load_contract_with_overlay
@@ -39,10 +44,13 @@ def register(subparsers: argparse._SubParsersAction):
     p.set_defaults(cmd=COMMAND, func=run)
 
 
-DAG_TMPL = """from __future__ import annotations
-from airflow import DAG
-from airflow.operators.bash import BashOperator
-from datetime import datetime
+# Airflow 2.7+ and 3.x: see codegen_utils.airflow_dag_import. The import blocks hold
+# no braces, so the template stays safe for str.format.
+DAG_TMPL = (
+    "from __future__ import annotations\n"
+    + airflow_dag_import()
+    + airflow_standard_import("BashOperator")
+    + """from datetime import datetime
 with DAG(
     dag_id={dag_id},
     start_date=datetime(2024,1,1),
@@ -65,6 +73,7 @@ def register(subparsers: argparse._SubParsersAction):
     )
     validate >> plan >> apply
 """
+)
 
 
 def _validate_cron(raw: object) -> str:

fluid_build/providers/aws/codegen/airflow.py

@@ -32,6 +32,8 @@
 from typing import Any, Dict, List, Optional
 
 from fluid_build.providers.common.codegen_utils import (
+    airflow_dag_import,
+    airflow_standard_import,
     escape_for_docstring,
     json_literal,
     py_str_literal,
@@ -41,11 +43,20 @@
 logger = logging.getLogger(__name__)
 
 
-# Module-level names the generated DAG binds itself. A task variable must not
-# collide with any of them.
+# Module-level names the generated DAG binds itself, including every name its
+# imports bind (the TaskFlow ``dag`` / ``task`` decorators and the operator
+# classes). A task variable must not collide with any of them: ``taskId: task``
+# emitted ``def task():`` inside the DAG function, which shadows the decorator.
 _EMITTED_MODULE_NAMES = frozenset(
     {
         "dag",
+        "task",
+        "PythonOperator",
+        "S3CreateBucketOperator",
+        "GlueJobOperator",
+        "AthenaOperator",
+        "RedshiftDataOperator",
+        "LambdaInvokeFunctionOperator",
         "default_args",
         "logger",
         "json",
@@ -187,11 +198,14 @@ def _generate_dag_header(
 
 def _generate_dag_imports() -> str:
     """Generate import statements."""
-    return """from datetime import datetime, timedelta
-from zoneinfo import ZoneInfo
-from airflow import DAG
-from airflow.operators.python import PythonOperator
-from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator
+    return (
+        (
+            "from datetime import datetime, timedelta\n"
+            "from zoneinfo import ZoneInfo\n"
+            + airflow_dag_import()
+            + airflow_standard_import("PythonOperator")
+        )
+        + """from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator
 from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
 from airflow.providers.amazon.aws.operators.athena import AthenaOperator
 from airflow.providers.amazon.aws.operators.redshift_data import RedshiftDataOperator
@@ -200,6 +214,7 @@ def _generate_dag_imports() -> str:
 import logging
 
 logger = logging.getLogger(__name__)"""
+    )
 
 
 def _generate_dag_definition(
@@ -227,7 +242,7 @@ def _generate_dag_definition(
     dag_id={py_str_literal(_sanitize_dag_id(contract_id))},
     default_args=default_args,
     description={py_str_literal(contract_name)},
-    schedule_interval={py_str_literal(airflow_schedule)},
+    schedule={py_str_literal(airflow_schedule)},
     start_date=datetime(2026, 1, 1, tzinfo=ZoneInfo({py_str_literal(timezone)})),
     catchup=False,
     tags={tags!r},
@@ -686,7 +701,10 @@ def _generate_taskflow_imports() -> str:
     """Generate import statements for TaskFlow API."""
     return """from datetime import datetime, timedelta
 from zoneinfo import ZoneInfo
-from airflow.decorators import dag, task
+try:  # Airflow 3
+    from airflow.sdk import dag, task
+except ImportError:  # Airflow 2.x
+    from airflow.decorators import dag, task
 from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator
 from airflow.providers.amazon.aws.operators.athena import AthenaOperator
 from airflow.providers.amazon.aws.operators.redshift_data import RedshiftDataOperator
@@ -736,6 +754,20 @@ def _generate_taskflow_dag(
 
     # Build per-task variable names for dependency wiring
     task_vars: Dict[str, str] = _unique_task_identifiers(tasks)
+    # What a dependency edge connects: the operator itself, or a @task's call result
+    # (``<var>_result``). A @task function is not a task: ``fn >> fn`` raises TypeError.
+    nodes: Dict[str, str] = {}
+    # A @task's result gets a name of its own, allocated against every task variable and
+    # every other result: a task named ``chk_result`` and ``chk``'s result must not share one.
+    taken = set(task_vars.values()) | set(_EMITTED_MODULE_NAMES)
+
+    def _result_name(var: str) -> str:
+        candidate, suffix = f"{var}_result", 2
+        while candidate in taken:
+            candidate = f"{var}_result_{suffix}"
+            suffix += 1
+        taken.add(candidate)
+        return candidate
 
     for t in tasks:
         tid = t.get("taskId", "")
@@ -755,6 +787,7 @@ def _generate_taskflow_dag(
             lines.append(f"        region_name={py_str_literal(params.get('region', region))},")
             lines.append("        aws_conn_id='aws_default',")
             lines.append("    )")
+            nodes[tid] = var
         elif service == "athena" and operation == "execute_query":
             query = params.get("query", "")
             db = params.get("database", "default")
@@ -769,6 +802,7 @@ def _generate_taskflow_dag(
             lines.append(f"        output_location={py_str_literal(out)},")
             lines.append("        aws_conn_id='aws_default',")
             lines.append("    )")
+            nodes[tid] = var
         elif service == "glue":
             # Glue operations use @task decorators
             lines.append("")
@@ -792,7 +826,9 @@ def _generate_taskflow_dag(
                     f"json.loads({json_literal(params)}), {py_str_literal(account_id)}, "
                     f"{py_str_literal(region)})"
                 )
-            lines.append(f"    {var}_result = {var}()")
+            result = _result_name(var)
+            lines.append(f"    {result} = {var}()")
+            nodes[tid] = result
         else:
             # Generic @task fallback
             lines.append("")
@@ -803,17 +839,19 @@ def _generate_taskflow_dag(
                 f"json.loads({json_literal(params)}), {py_str_literal(account_id)}, "
                 f"{py_str_literal(region)})"
             )
-            lines.append(f"    {var}_result = {var}()")
+            result = _result_name(var)
+            lines.append(f"    {result} = {var}()")
+            nodes[tid] = result
 
     # Dependencies
     for t in tasks:
         tid = t.get("taskId", "")
         depends_on = t.get("dependsOn", [])
         if depends_on:
-            var = task_vars[tid]
+            node = nodes[tid]
             for dep in depends_on:
-                dep_var = task_vars.get(dep)
-                if dep_var is None:
+                dep_node = nodes.get(dep)
+                if dep_node is None:
                     logger.warning(
                         "aws airflow: task %r dependsOn %r, which is not a "
                         "provider_action task in this DAG — edge skipped",
@@ -822,7 +860,7 @@ def _generate_taskflow_dag(
                     )
                     lines.append(f"    # dependency on non-DAG task {dep!r} skipped")
                     continue
-                lines.append(f"    {dep_var} >> {var}")
+                lines.append(f"    {dep_node} >> {node}")
 
     lines.append("")
     lines.append("")

fluid_build/providers/common/codegen_utils.py

@@ -166,15 +166,15 @@ def convert_schedule_to_cron(schedule: str) -> str:
 
 def convert_schedule_to_airflow(schedule: str) -> str:
     """
-    Convert FLUID schedule to Airflow schedule_interval format.
+    Convert FLUID schedule to an Airflow ``schedule`` value.
 
     Airflow accepts both cron and special keywords like @daily.
 
     Args:
         schedule: FLUID schedule string
 
     Returns:
-        Airflow schedule_interval value
+        Airflow ``schedule`` value (``DAG(schedule=...)``)
     """
     schedule_lower = schedule.lower().strip()
 
@@ -190,6 +190,45 @@ def convert_schedule_to_airflow(schedule: str) -> str:
     return "@daily"
 
 
+# ── Generated DAGs run on Airflow 2.7+ and 3.x ─────────────────────────────
+#
+# Airflow 3 removed ``DAG(schedule_interval=...)`` and ``airflow.utils.dates.days_ago``,
+# moved the core operators and sensors to apache-airflow-providers-standard, and
+# exports ``DAG`` from ``airflow.sdk``. A generated DAG therefore passes
+# ``schedule=`` (Airflow 2.4+), a fixed timezone-aware ``start_date`` with
+# ``catchup=False``, and imports each name from its Airflow 3 home first, falling
+# back to the Airflow 2 core path: the same shape schedulers/airflow/fluid_apply.py
+# already emits. The import paths are Airflow's own deprecation map
+# (airflow-core/src/airflow/operators/__init__.py and sensors/__init__.py at 3.2.2).
+_AIRFLOW_STANDARD_MODULES = {
+    "BashOperator": "operators.bash",
+    "PythonOperator": "operators.python",
+    "ExternalTaskSensor": "sensors.external_task",
+}
+
+
+def airflow_dag_import(indent: str = "") -> str:
+    """``DAG`` from ``airflow.sdk`` on Airflow 3, from ``airflow`` on Airflow 2."""
+    return (
+        f"{indent}try:  # Airflow 3\n"
+        f"{indent}    from airflow.sdk import DAG\n"
+        f"{indent}except ImportError:  # Airflow 2.x\n"
+        f"{indent}    from airflow import DAG\n"
+    )
+
+
+def airflow_standard_import(name: str, indent: str = "") -> str:
+    """``name`` from apache-airflow-providers-standard (Airflow 3, or Airflow 2 with the
+    provider installed), else from Airflow 2's core module."""
+    module = _AIRFLOW_STANDARD_MODULES[name]
+    return (
+        f"{indent}try:  # Airflow 3, or Airflow 2 with apache-airflow-providers-standard\n"
+        f"{indent}    from airflow.providers.standard.{module} import {name}\n"
+        f"{indent}except ImportError:  # Airflow 2.x core\n"
+        f"{indent}    from airflow.{module} import {name}\n"
+    )
+
+
 def json_literal(value: Any) -> str:
     """Emit ``value`` as a Python *string literal* holding its JSON encoding.