Declarative Airflow DAG Generation - The FLUID Way

# bitcoin_tracker_manual.py - THE OLD WAY
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryCheckOperator
from datetime import datetime, timedelta
import sys
import os

# Add project to path
sys.path.insert(0, os.path.dirname(__file__))
from ingest_bitcoin_prices import fetch_bitcoin_price, insert_to_bigquery

# Default arguments
default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "email": ["alerts@example.com"],
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
}

# DAG definition
dag = DAG(
    dag_id="bitcoin_tracker_manual",
    default_args=default_args,
    description="Hourly Bitcoin price ingestion",
    schedule_interval="0 * * * *",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["crypto", "bitcoin", "manual"],
)

# Task 1: Validate contract
validate = BashOperator(
    task_id="validate_contract",
    bash_command="cd /path/to/project && python3 -m fluid_build.cli validate contract.fluid.yaml",
    dag=dag,
)

# Task 2: Fetch price
def fetch_btc_task(**context):
    price_data = fetch_bitcoin_price()
    context["ti"].xcom_push(key="price_usd", value=price_data["price_usd"])
    return price_data

fetch_price = PythonOperator(
    task_id="fetch_bitcoin_price",
    python_callable=fetch_btc_task,
    provide_context=True,
    dag=dag,
)

# Task 3: Insert to BigQuery
def insert_task(**context):
    price_data = context["ti"].xcom_pull(task_ids="fetch_bitcoin_price")
    success = insert_to_bigquery(
        price_data,
        "your-project-id",
        "crypto_data",
        "bitcoin_prices"
    )
    if not success:
        raise Exception("Insert failed")

insert_price = PythonOperator(
    task_id="insert_to_bigquery",
    python_callable=insert_task,
    provide_context=True,
    dag=dag,
)

# Task 4: Run dbt
run_dbt = BashOperator(
    task_id="run_dbt_models",
    bash_command="cd /path/to/project/dbt && dbt run --profiles-dir .",
    dag=dag,
)

# Task 5: Quality check
check_quality = BigQueryCheckOperator(
    task_id="check_data_quality",
    sql="""
        SELECT COUNT(*) > 0
        FROM `your-project.crypto_data.bitcoin_prices`
        WHERE DATE(timestamp) = CURRENT_DATE()
    """,
    use_legacy_sql=False,
    dag=dag,
)

# Define dependencies
validate >> fetch_price >> insert_price >> run_dbt >> check_quality

"""
Airflow DAG for FLUID Data Product: bitcoin-prices-gcp

Auto-generated from FLUID contract v0.7.1
Generated at: 2026-01-21T12:00:00

Domain: crypto
Description: Bitcoin price tracking data product
"""
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from datetime import datetime, timedelta

# DAG configuration (from contract metadata)
default_args = {
    'owner': 'fluid',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,  # ← From contract.builds[].execution.retries
    'retry_delay': timedelta(minutes=5),
}

# DAG definition
dag = DAG(
    dag_id="bitcoin_tracker",
    description="""Bitcoin price tracking data product""",
    schedule_interval="0 * * * *",  # ← From --schedule argument
    start_date=days_ago(1),
    catchup=False,
    tags=["fluid", "crypto", "bitcoin"],  # ← From contract.tags
    default_args=default_args
)

# Provision dataset: bitcoin_prices_table (from exposes[0])
provision_bitcoin_prices_table = BashOperator(
    task_id="provision_bitcoin_prices_table",
    bash_command="bq mk --project_id=<<YOUR_PROJECT_HERE>> --dataset crypto_data || true",
    dag=dag
)

# Schedule task: ingest_bitcoin_prices (from builds[0])
schedule_ingest_bitcoin_prices = BashOperator(
    task_id="schedule_ingest_bitcoin_prices",
    bash_command="python3 runtime/ingest_bitcoin_prices.py",
    dag=dag
)

# Schedule task: calculate_daily_summary (from builds[1])
schedule_calculate_daily_summary = BashOperator(
    task_id="schedule_calculate_daily_summary",
    bash_command="dbt run --models daily_price_summary",
    dag=dag
)

# Schedule task: calculate_price_trends (from builds[2])
schedule_calculate_price_trends = BashOperator(
    task_id="schedule_calculate_price_trends",
    bash_command="dbt run --models price_trends",
    dag=dag
)

# Task dependencies (inferred from builds.outputs)
provision_bitcoin_prices_table >> schedule_ingest_bitcoin_prices
schedule_ingest_bitcoin_prices >> schedule_calculate_daily_summary
schedule_ingest_bitcoin_prices >> schedule_calculate_price_trends