DAG Design for Airflow AI Prompt

Design and implement an Airflow DAG for this pipeline.

Pipeline requirements: {{pipeline_requirements}}
Schedule: {{schedule}}
Dependencies: {{upstream_dependencies}}

1. DAG structure best practices:
   - Use @dag decorator with explicit schedule, catchup=False, and max_active_runs=1
   - Set meaningful dag_id, description, and tags
   - Define default_args with retries=3, retry_delay=timedelta(minutes=5), retry_exponential_backoff=True
   - Set execution_timeout per task to prevent hung tasks from blocking slots

2. Task design:
   - Keep tasks small and single-responsibility (one logical operation per task)
   - Use TaskFlow API (@task decorator) for Python tasks — cleaner than PythonOperator
   - Use KubernetesPodOperator for heavy workloads — isolates dependencies and resources
   - Avoid pulling large datasets into Airflow worker memory — use Spark/dbt/SQL for heavy transforms

3. Dependencies and branching:
   - Define task dependencies with >> and << operators
   - Use BranchPythonOperator for conditional logic
   - Use TriggerDagRunOperator for cross-DAG dependencies (prefer sensors for blocking waits)

4. Idempotency:
   - All tasks must be safely re-runnable
   - Use execution_date in file paths and table partition filters to scope each run

5. Alerting:
   - on_failure_callback: send Slack alert with DAG name, task, execution_date, and log URL
   - SLA miss callback: alert if DAG does not complete within SLA

6. Testing:
   - Unit test task logic separately from the DAG
   - Use airflow dags test dag_id execution_date for local DAG runs

Return: complete DAG code with all operators, dependencies, and alerting.