fix(airflow): the init DAG helper writes what generate schedule writes, and Snowflake python tasks no longer lose the DAG

CHANGELOG.md

@@ -183,6 +183,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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.
+- **A python task no longer loses the Snowflake orchestration DAG.** `generate_airflow_dag`
+  in `providers/snowflake/orchestration`, a library function no `fluid` command calls
+  (`fluid export` uses `providers/snowflake/codegen`), emitted `python_callable=<callable>`
+  for a function it never defined, so Airflow raised `NameError` parsing the file and
+  dropped the whole DAG. It now defines a stub of that callable above the task: the DAG
+  loads, and the task fails with `AirflowFailException`, naming itself and skipping the
+  retries a stub cannot pass, until the function is written. A callable whose name the DAG
+  already uses (`dag`, an operator class, an earlier task) or a `__dunder__` name is refused
+  with a `ValueError` when the DAG is generated.
+- **The `fluid init` DAG helper writes what `fluid generate schedule` writes, or nothing.**
+  `generate_dag_for_project` ran `fluid generate-airflow --output-dir`, a flag that command
+  does not have, so it always wrote a basic validate/plan/apply DAG, or no DAG when `fluid`
+  was not on PATH. It now runs `fluid generate schedule --scheduler airflow --overwrite`
+  in-process through the CLI's own parser, and its `dags/README.md` names the DAGs that run
+  wrote and says to edit the contract, not the DAG. It writes no DAG, and says why, when the
+  contract schedules nothing (no build with a cron trigger and no `orchestration.tasks`, as
+  with customer-360's `manual` build) or when the command refuses the contract, an invalid
+  cron for instance, instead of a basic DAG on a schedule the contract never declared.
+  `create_basic_dag`, no longer a fallback, passes `fluid apply --yes` instead of the
+  rejected `--auto-approve` and quotes the project path. No CLI command reaches the helper
+  with DAGs enabled today: `fluid demo` turns them off.
 
 ## [0.21.0] — 2026-10-19
 

examples/generated_dags/financial_analytics_dag.py

@@ -8,33 +8,30 @@
 AWS Region: us-east-1
 
 Auto-generated by FLUID Forge - DO NOT EDIT MANUALLY
-Generated: 2026-01-30T14:02:10.320788
+Generated: 2026-10-21T21:45:23.515208
 """
 
-import json
-import logging
 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.athena import AthenaOperator
+try:  # Airflow 3
+    from airflow.sdk import DAG
+except ImportError:  # Airflow 2.x
+    from airflow import DAG
+try:  # Airflow 3, or Airflow 2 with apache-airflow-providers-standard
+    from airflow.providers.standard.operators.python import PythonOperator
+except ImportError:  # Airflow 2.x core
+    from airflow.operators.python 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.lambda_function import LambdaInvokeFunctionOperator
+from airflow.providers.amazon.aws.operators.athena import AthenaOperator
 from airflow.providers.amazon.aws.operators.redshift_data import RedshiftDataOperator
-from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator
+from airflow.providers.amazon.aws.operators.lambda_function import LambdaInvokeFunctionOperator
+import json
+import logging
 
 logger = logging.getLogger(__name__)
 
-
-def _ensure_glue_database(**kwargs):
-    return kwargs
-
-
-def _ensure_glue_table(**kwargs):
-    return kwargs
-
-
 # Default DAG arguments
 default_args = {
     "owner": "fluid-forge",
@@ -51,12 +48,100 @@ def _ensure_glue_table(**kwargs):
     dag_id="financial_analytics",
     default_args=default_args,
     description="Financial Analytics Pipeline",
-    schedule_interval="0 2 * * *",
+    schedule="0 2 * * *",
     start_date=datetime(2026, 1, 1, tzinfo=ZoneInfo("UTC")),
     catchup=False,
     tags=["fluid", "auto-generated", "financial_analytics"],
 )
 
+
+# Helper functions
+
+
+def _ensure_glue_database(database: str, region: str):
+    """Ensure Glue database exists."""
+    import boto3
+
+    client = boto3.client("glue", region_name=region)
+
+    try:
+        client.create_database(DatabaseInput={"Name": database})
+        logger.info(f"Created Glue database: {database}")
+    except client.exceptions.AlreadyExistsException:
+        logger.info(f"Glue database already exists: {database}")
+
+
+def _ensure_glue_table(database: str, table: str, params: dict, region: str):
+    """Ensure Glue table exists."""
+    import boto3
+
+    client = boto3.client("glue", region_name=region)
+
+    columns = params.get("columns", [])
+    location = params.get("location", "")
+    input_format = params.get("input_format", "parquet")
+
+    FORMAT_MAP = {
+        "parquet": {
+            "InputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
+            "OutputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat",
+            "SerDe": "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe",
+        },
+        "orc": {
+            "InputFormat": "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat",
+            "OutputFormat": "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat",
+            "SerDe": "org.apache.hadoop.hive.ql.io.orc.OrcSerde",
+        },
+        "json": {
+            "InputFormat": "org.apache.hadoop.mapred.TextInputFormat",
+            "OutputFormat": "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat",
+            "SerDe": "org.openx.data.jsonserde.JsonSerDe",
+        },
+        "csv": {
+            "InputFormat": "org.apache.hadoop.mapred.TextInputFormat",
+            "OutputFormat": "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat",
+            "SerDe": "org.apache.hadoop.hive.serde2.OpenCSVSerde",
+        },
+    }
+    fmt = FORMAT_MAP.get(input_format, FORMAT_MAP["parquet"])
+
+    table_input = {
+        "Name": table,
+        "StorageDescriptor": {
+            "Columns": columns,
+            "Location": location,
+            "InputFormat": fmt["InputFormat"],
+            "OutputFormat": fmt["OutputFormat"],
+            "SerdeInfo": {"SerializationLibrary": fmt["SerDe"]},
+        },
+        "TableType": "EXTERNAL_TABLE",
+    }
+    description = params.get("description")
+    if description:
+        table_input["Description"] = description
+
+    try:
+        client.get_table(DatabaseName=database, Name=table)
+        client.update_table(DatabaseName=database, TableInput=table_input)
+        logger.info(f"Updated Glue table: {database}.{table}")
+    except client.exceptions.EntityNotFoundException:
+        client.create_table(DatabaseName=database, TableInput=table_input)
+        logger.info(f"Created Glue table: {database}.{table}")
+
+
+def _execute_provider_action(action: str, params: dict, account_id: str, region: str):
+    """Provider-action tasks are retired.
+
+    forge-cli compiles the contract to OpenTofu and runs ``tofu`` — a
+    generated DAG no longer performs native cloud CRUD. Provision
+    infrastructure with ``fluid apply`` (the OpenTofu engine).
+    """
+    raise RuntimeError(
+        f"provider-action task {action!r} is retired: provision infrastructure "
+        "with `fluid apply` (the OpenTofu engine), then re-generate this DAG"
+    )
+
+
 # Task definitions
 create_bucket = S3CreateBucketOperator(
     task_id="create_bucket",
@@ -79,27 +164,9 @@ def _ensure_glue_table(**kwargs):
     python_callable=lambda: _ensure_glue_table(
         database="financial_analytics",
         table="transactions",
-        params={
-            "database": "financial_analytics",
-            "table": "transactions",
-            "s3_location": "s3://financial-data-lake-{{account_id}}/transactions/",
-            "format": "parquet",
-            "schema": {
-                "columns": [
-                    {"name": "transaction_id", "type": "string"},
-                    {"name": "customer_id", "type": "string"},
-                    {"name": "amount", "type": "double"},
-                    {"name": "currency", "type": "string"},
-                    {"name": "timestamp", "type": "timestamp"},
-                    {"name": "category", "type": "string"},
-                ],
-                "partition_keys": [
-                    {"name": "year", "type": "int"},
-                    {"name": "month", "type": "int"},
-                    {"name": "day", "type": "int"},
-                ],
-            },
-        },
+        params=json.loads(
+            '{"database": "financial_analytics", "format": "parquet", "s3_location": "s3://financial-data-lake-{{account_id}}/transactions/", "schema": {"columns": [{"name": "transaction_id", "type": "string"}, {"name": "customer_id", "type": "string"}, {"name": "amount", "type": "double"}, {"name": "currency", "type": "string"}, {"name": "timestamp", "type": "timestamp"}, {"name": "category", "type": "string"}], "partition_keys": [{"name": "year", "type": "int"}, {"name": "month", "type": "int"}, {"name": "day", "type": "int"}]}, "table": "transactions"}'
+        ),
         region="us-east-1",
     ),
     dag=dag,