将 Airflow 2.x DAG 代码迁移到…

当需要把 Airflow 2.x 项目升级到 3.x 并处理兼容性破坏变更时，按步骤迁移 DAG 代码、导入、调度与元数据访问方式。

适用场景

解决思路

1. 先升级到 Airflow 2.11，再升级到至少 Airflow 3.0.11（最好直接到 3.1），避免其他升级路径导致无法回滚；早期 3.0 版本 bug 较多，3.1 体验更好。
2. 运行 Ruff 的 Airflow 迁移规则自动修复可检测问题：`ruff check --preview --select AIR --fix --unsafe-fixes .`（AIR30/AIR301/AIR302/AIR31/AIR311/AIR312）。
3. 用 `ruff check --preview --select AIR .` 检查剩余未修复的 Airflow 问题。
4. 按 reference/migration-checklist.md 中的手动搜索清单扫描剩余问题，重点关注：直接访问元数据 DB、遗留导入、调度/上下文键、XCom pickling、datasets 转 assets、REST API/认证、plugins、文件路径。
5. 显式审查硬性行为/配置陷阱：cron 调度语义如需 Airflow 2 风格可设 `AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True`；`.airflowignore` 语法从 regexp 改为 glob，如需保留 regexp 设 `AIRFLOW__CORE__DAG_IGNORE_FILE_SYNTAX=regexp`；OAuth 回调 URL 增加 `/auth/` 前缀（如 `/auth/oauth-authorized/google`）；共享工具导入不再支持裸导入 `import common`，改用完全限定导入 `import dags.common`。
6. 按文件和问题类型规划变更：修复导入、更新 operators/hooks/providers、把元数据访问重构为使用 Airflow client 而非直接访问、修复过时的上下文变量、修复调度逻辑。
7. 增量实施变更，每次重大修改后重新运行 Ruff 和代码搜索。
8. 向用户解释变更，并提醒他们测试更新后的逻辑，例如重构后的元数据访问、调度逻辑和 Airflow 上下文使用。
9. 处理架构与元数据 DB 访问变化：Workers 不再直接连接元数据 DB，任务代码通过 API server 暴露的 Task Execution API 运行，DAG processor 作为独立于 scheduler 的进程运行，Triggerer 通过进程内 API server 使用任务执行机制。
10. 搜索并替换直接元数据 DB 访问模式：`provide_session`、`create_session`、`@provide_session`、`from airflow.settings import Session`、`from airflow.settings import engine`、`session.query(DagModel)...`、`session.query(DagRun)...`；这些在 Airflow 3 中会抛出 `RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.x`。
11. 优先使用 Airflow Python client 进行丰富的元数据访问：在 `requirements.txt` 添加 `apache-airflow-client==<your-airflow-runtime-version>`，用 `airflow_client.client.Configuration` 和 `DAGApi` 等调用 API。
12. 简单场景可直接用 `requests` 调用 REST API，例如 `GET {_HOST}/api/v2/dags`，带 `Authorization: Bearer <token>` 头。
13. 修复 Trigger 实现陷阱：如果 trigger 在 asyncio 事件循环中同步调用 hooks，可能失败或阻塞；优先通过 `sync_to_async(...)` 调用 hooks 或确保 hook 调用是 async-safe。
14. 按 Quick Reference 表更新关键导入：`airflow.operators.dummy_operator.DummyOperator` → `airflow.providers.standard.operators.empty.EmptyOperator`；`airflow.operators.bash.BashOperator` → `airflow.providers.standard.operators.bash.BashOperator`；`airflow.operators.python.PythonOperator` → `airflow.providers.standard.operators.python.PythonOperator`；`airflow.decorators.dag` → `airflow.sdk.dag`；`airflow.decorators.task` → `airflow.sdk.task`；`airflow.datasets.Dataset` → `airflow.sdk.Asset`。
15. 更新上下文键：`execution_date` → `context["dag_run"].logical_date`；`tomorrow_ds`/`yesterday_ds` → 用 `ds` 加日期运算 `macros.ds_add(ds, 1)`/`macros.ds_add(ds, -1)`；`prev_ds`/`next_ds` → `prev_start_date_success` 或 timetable API；`triggering_dataset_events` → `triggering_asset_events`；`templates_dict` → `context["params"]`。
16. 处理 asset 触发运行的 `logical_date` 可能为 `None`，使用 `context["dag_run"].logical_date` 时做防御性判断。
17. 不能用未来的 `logical_date` 触发，改用 `logical_date=None` 并依赖 `run_id`。
18. 注意默认行为变化：`schedule` 默认从 `timedelta(days=1)` 变为 `None`；`catchup` 默认从 `True` 变为 `False`。
19. 注意回调行为变化：`on_success_callback` 不再在 skip 时运行，如需处理用 `on_skipped_callback`；`@teardown` 不允许搭配 `TriggerRule.ALWAYS`，teardown 现在即使 DAG run 提前终止也会执行。
20. 参考详细文档：reference/config-changes.md（airflow.cfg 配置迁移）、reference/migration-patterns.md（导入、调度、XCom、Assets、DAG bundles、运行时行为示例）、reference/removed-methods.md（移除的模型方法及 SDK/API 迁移路径）、reference/migration-checklist.md（Ruff 未覆盖问题的搜索模式与修复）。

适用边界 · 注意事项

• 尚未升级到 Airflow 2.11 或无法按推荐路径升级时，不要直接跳到 Airflow 3，否则无法回滚。
• 如果项目仍依赖直接 ORM 访问元数据 DB 且无法改为 Airflow client 或 REST API，迁移会失败。
• 如果无法接受 Airflow 3 的默认行为变化（如 schedule 默认 None、catchup 默认 False、回调行为变化）且未做相应配置调整，不要贸然迁移。