💻 Coding
Apache Airflow 2 to 3 DAG Migration Planner: Ruff AIR301 and AIR302 Scans, airflow.sdk Imports, schedule and logical_date Renames, Standard Provider Operators, No Direct Metadata DB Access, and a Staging Cutover Checklist
Plan an Airflow 2.x to 3.0 upgrade DAG by DAG: run Ruff's AIR rules to find removed and moved names, switch imports to airflow.sdk and the standard provider, replace schedule_interval, execution_date, days_ago, and SubDAGs, move task code off direct metadata database sessions to the REST API, and finish with a config update and staging cutover checklist.
0Reviews
Prompt
Act as a data platform engineer who has upgraded production Apache Airflow deployments from 2.x to 3.0, maintains a few hundred DAGs, and has been paged for DAGs that vanished from the UI after an import error, for templates still using execution_date, and for tasks that opened ORM sessions against the metadata database. Inputs: - Current Airflow version, Python version, and how it is installed (pip constraints file, official Docker image, Helm chart, managed service): [CurrentVersion] - DAG inventory: file names with the imports, DAG arguments, operators, and Jinja templates they use (paste code or excerpts): [DagInventory] - Installed provider packages and versions: [Providers] - Executor and deployment layout (scheduler, webserver, workers, triggerer): [Deployment] - Any task code that touches the metadata database, airflow.settings.Session, or internal models: [DbAccessInTasks] - Output format: [Format] Generate: 1. A readiness gate: move to the latest 2.x release first (2.10 or the 2.11 bridge release), clear every deprecation warning in the scheduler and DAG processor logs, back up the metadata database, and confirm every provider in Providers has a release that supports Airflow 3. 2. A Ruff scan plan: the exact commands to run the AIR301 (removed in Airflow 3) and AIR302 (moved to a provider) rules over the dags folder with preview mode, how to review autofixes before accepting them, and how to read each rule code in the output. 3. A rename table built from DagInventory: from airflow.decorators and airflow.models.DAG to airflow.sdk (dag, task, DAG, Variable, TaskGroup), Dataset to Asset, schedule_interval and timetable to schedule, airflow.utils.dates.days_ago to a fixed pendulum start_date, and BashOperator, PythonOperator, and the core sensors to apache-airflow-providers-standard import paths. 4. A template and context fix list: execution_date, next_ds, prev_ds, tomorrow_ds, and yesterday_ds replaced with logical_date or data_interval_start and data_interval_end, with each changed template line shown before and after. 5. Behavior changes to decide per DAG: catchup now defaults to False, SubDAGs are removed so each one becomes a TaskGroup, SLA callbacks are removed, and the REST API moves from v1 to v2. 6. A DbAccessInTasks rewrite: task code may no longer open metadata database sessions, so each case moves to the Airflow REST API through the Python client, an Airflow Variable or Connection read through the SDK, or an external table you own. 7. A Deployment change list: the webserver is replaced by airflow api-server, the DAG processor runs as its own component, run airflow config update to review renamed options and add --fix only after review, then airflow db migrate. 8. A staging cutover checklist: parse every DAG with zero import errors, trigger one run per DAG, compare task counts and durations with the 2.x baseline, then a rollback note that names the database backup. Constraints: - Only rewrite code that appears in DagInventory; never invent DAG names, tasks, or connection IDs. - Mark any provider version you cannot confirm as "check the provider changelog". No em dashes.
Instructions
Replace every [bracket] with your details before running. Works on ChatGPT, Claude, and Gemini.
Generated Output
This image was generated using the prompt above.
Examples
Example Input
CurrentVersion: Airflow 2.9.3 on Python 3.11, official Docker image with docker compose, constraints file pinned
DagInventory: daily_sales.py uses from airflow.models import DAG, schedule_interval="0 6 * * *", start_date=days_ago(1), BashOperator with bash_command "python load.py {{ execution_date.strftime('%Y-%m-%d') }}"; refresh_dash.py uses @dag from airflow.decorators, Dataset("s3://lake/sales"), and a SubDagOperator for three report tasks
Providers: apache-airflow-providers-amazon 8.25.0, apache-airflow-providers-postgres 5.11.0
Deployment: CeleryExecutor, Redis broker, one scheduler, two workers, webserver behind nginx
DbAccessInTasks: cleanup_runs.py opens settings.Session() and deletes DagRun rows older than 90 days
Format: phased plan with code diffs and commandsExample Output
PHASE 0: READINESS
- Upgrade 2.9.3 to the latest 2.x (2.10.x or the 2.11 bridge release) with its constraints file. Run one week and fix every deprecation warning in scheduler logs first.
- Back up the metadata database: pg_dump of the airflow DB, stored as airflow-pre3.dump.
- Providers: amazon 8.25.0 and postgres 5.11.0 predate Airflow 3. Check the provider changelog for the first versions that support Airflow 3 and pin those.
PHASE 1: RUFF SCAN
ruff check dags/ --select AIR301,AIR302 --preview
ruff check dags/ --select AIR301,AIR302 --preview --diff
Read AIR301 as "removed in Airflow 3, must change" and AIR302 as "moved to a provider package, install it and change the import". Apply --fix only after reading the --diff.
PHASE 2: CODE CHANGES
daily_sales.py
- from airflow.models import DAG
+ from airflow.sdk import DAG
- from airflow.operators.bash import BashOperator
+ from airflow.providers.standard.operators.bash import BashOperator
- schedule_interval="0 6 * * *",
- start_date=days_ago(1),
+ schedule="0 6 * * *",
+ start_date=pendulum.datetime(2026, 10, 1, tz="UTC"),
+ catchup=False, # stated on purpose; it is now the default
- bash_command="python load.py {{ execution_date.strftime('%Y-%m-%d') }}"
+ bash_command="python load.py {{ data_interval_start | ds }}"
Note: for a daily schedule, data_interval_start matches the old execution_date. Confirm with one run in staging.
refresh_dash.py
- from airflow.decorators import dag, task
- from airflow.datasets import Dataset
+ from airflow.sdk import dag, task, Asset, TaskGroup
- Dataset("s3://lake/sales")
+ Asset("s3://lake/sales")
SubDagOperator is removed. Replace it with:
with TaskGroup("reports") as reports:
build_daily() >> build_weekly() >> build_monthly()
Task IDs gain the prefix reports. so update any alerts or downstream sensors that reference the old IDs.
cleanup_runs.py
settings.Session() inside a task is blocked in Airflow 3. Options:
1. Preferred: delete this DAG and schedule airflow db clean --clean-before-timestamp from the host or a maintenance job outside task code.
2. If it must stay a DAG: call the REST API v2 with the apache-airflow-client and an account limited to this job.
PHASE 3: DEPLOYMENT
- Add apache-airflow-providers-standard to the image.
- docker compose: replace the webserver service command with airflow api-server, add a dag-processor service.
- Point nginx at the api-server port.
- airflow config update (review the list), then airflow config update --fix
- airflow db migrate
PHASE 4: STAGING CUTOVER
[ ] airflow dags list-import-errors returns nothing
[ ] Both DAGs show in the UI with the expected schedule
[ ] One manual run each; daily_sales passes the same date string as the 2.x run
[ ] reports TaskGroup task counts match the old SubDAG
[ ] Durations within the range of the last 2.x runs
[ ] Rollback: restore airflow-pre3.dump and redeploy the 2.x image tag