在 Airflow 3.1+ 中把 FastAPI…

当需要在 Airflow 3.1+ 进程内构建自定义 API、UI 页面、React 组件、中间件、宏、operator 链接或定时表时使用。

适用场景

解决思路

1. 确认使用 Airflow 3.1+；fastapi_apps、react_apps、external_views 需要 Airflow 3.1+。
2. 绝不 import flask、flask_appbuilder，绝不使用 appbuilder_views / flask_blueprints；这些是 Airflow 2 模式，在 Airflow 3 不工作；如果现有代码使用它们，用 FastAPI 重写整个注册块。
3. 记住 FastAPI 插件端点默认不受 Airflow 认证保护；如果端点必须私有，使用 FastAPI 的安全工具（如 OAuth2、API key）显式实现认证。
4. Python 插件文件变更后重启 API server；HTML、JS、CSS 静态文件变更会立即生效。
5. 开发时设置 AIRFLOW__CORE__LAZY_LOAD_PLUGINS=False，使插件在启动时加载而不是惰性加载。
6. 在 external_views 中 href 必须没有前导斜杠；在 HTML 和 JavaScript 中所有资源和 fetch() 调用都用相对路径；绝对路径会在反向代理后失效。
7. 写代码前检查：是否使用 fastapi_apps / FastAPI，而不是 appbuilder_views / Flask。
8. 写代码前检查：所有 HTML/JS 资源路径和 fetch() 调用是否都是相对路径（无前导斜杠）。
9. 写代码前检查：所有同步 SDK 或 SQLAlchemy 调用是否包装在 asyncio.to_thread()。
10. 写代码前检查：static/ 和 assets/ 目录是否在 FastAPI app 挂载它们之前已经存在。
11. 写代码前检查：如果端点必须私有，是否添加显式 FastAPI 认证。
12. 选择插件组件：一个插件类可同时注册多个组件类型。
13. 用 fastapi_apps 注册自定义 API 端点：FastAPI app 挂载到 Airflow 进程。
14. 用 external_views 注册导航/页面链接：将 URL 作为 iframe 嵌入或链接出去。
15. 用 react_apps 注册 React 组件：自定义 React app 嵌入 Airflow UI。
16. 用 fastapi_root_middlewares 注册 API 中间件：拦截所有 Airflow API 请求/响应。
17. 用 macros 注册 Jinja 宏：DAG 模板中的可复用 Python 函数。
18. 用 operator_extra_links / global_operator_extra_links 注册任务实例按钮：task Detail 视图中的额外链接按钮。
19. 用 timetables 注册自定义调度逻辑。
20. 用 listeners 注册事件钩子：Airflow 事件的监听器回调。
21. 给每个插件在 plugins/ 下自己的子目录，保持 Python 文件、静态资源、模板在一起，便于多插件项目。
22. 使用 plugins/my-plugin/plugin.py 作为 AirflowPlugin 子类，Airflow 自动发现；放置 static/index.html、static/app.js、assets/icon.svg 等。
23. 在 plugin.py 中使用 BASE_DIR = Path(__file__).parent，使其解析到 plugins/my-plugin/，从而正确解析 static 和 asset 路径。
24. 在启动 Airflow 前创建插件子目录以及 static/ 和 assets/ 文件夹，否则 StaticFiles 会在导入时抛错。
25. 注册骨架导入：from pathlib import Path、from airflow.plugins_manager import AirflowPlugin、from fastapi import FastAPI、from fastapi.staticfiles import StaticFiles、from fastapi.responses import FileResponse。
26. 注册骨架创建 app：app = FastAPI(title='My Plugin')。
27. 注册骨架挂载 static：app.mount('/static', StaticFiles(directory=BASE_DIR / 'static'), name='static')。
28. 注册骨架挂载 assets：app.mount('/assets', StaticFiles(directory=BASE_DIR / 'assets'), name='assets')。
29. 注册骨架定义 class MyPlugin(AirflowPlugin)，设置 name = 'my_plugin'。
30. 注册骨架设置 fastapi_apps = [{'app': app, 'url_prefix': '/my-plugin', 'name': 'My Plugin'}]，插件可在 {AIRFLOW_HOST}/my-plugin/ 访问。
31. 注册骨架设置 external_views = [{'name': 'My Plugin', 'href': 'my-plugin/ui', 'destination': 'nav', 'category': 'browse', 'url_route': 'my-plugin', 'icon': '/my-plugin/static/icon.svg'}]；href 无前导斜杠，icon 使用前导斜杠并由 FastAPI 提供。
32. external_views 的 destination 设为 'nav' 时显示在左侧导航栏，并设置 category。
33. external_views 的 destination 设为 'dag' 时在每个 Dag 页添加额外 tab。
34. external_views 的 destination 设为 'dag_run' 时在每个 Dag run 页添加额外 tab。
35. external_views 的 destination 设为 'task' 时在每个 task 页添加额外 tab。
36. external_views 的 destination 设为 'task_instance' 时在每个 task instance 页添加额外 tab。
37. Nav bar category 可设 'browse' 或 'admin'，省略则顶级；仅用于 destination='nav'。
38. external_views 的 href 可以是内部端点的相对路径（如 'my-plugin/ui'）或完整外部 URL。
39. 插件可以只有 external_views 而没有 fastapi_apps，适用于简单链接或 tab，无需后端；示例 LearnViewPlugin 的 href 用 https://www.astronomer.io/docs/learn，destination 为 'dag'，url_route 为 'learn'。
40. 无前导斜杠规则仅适用于内部路径；完整 https:// URL 可以。
41. 提供 UI 入口点：定义 @app.get('/ui', response_class=FileResponse) async def serve_ui(): return FileResponse(BASE_DIR / 'static' / 'index.html')。
42. HTML 中始终使用相对路径，如 <link rel='stylesheet' href='static/app.css' /> 和 <script src='static/app.js?v=20240315'></script>；不要用 /my-plugin/static/app.js。
43. JavaScript 中始终使用相对路径，如 fetch('api/dags')；不要用 fetch('/my-plugin/api/dags')。
44. 仅当插件调用 Airflow REST API 时才需要后续 API 调用步骤；只服务静态文件、注册 external_views 或直接 DB 访问的插件可跳过。
45. 如果需要 REST API 通信，添加 apache-airflow-client 依赖。
46. 根据项目文件添加依赖：requirements.txt 中追加 apache-airflow-client。
47. pyproject.toml（uv / poetry）用 uv add apache-airflow-client 或 poetry add apache-airflow-client。
48. 如果以上文件都没有，告诉用户：在运行插件前把 apache-airflow-client 加到依赖。
49. 使用 apache-airflow-client 与 Airflow REST API 通信；SDK 是同步的，而 FastAPI 路由是异步的，绝不要在 async def 中直接调用阻塞 SDK 方法，否则会阻塞事件循环并冻结所有并发请求。
50. JWT token 管理：每个进程缓存一个 token。
51. JWT token 管理：在 1 小时过期前 5 分钟刷新。
52. JWT token 管理：使用双重检查锁定，避免多个并发请求同时刷新。
53. 环境变量前缀：将 MYPLUGIN_ 替换为从插件名派生的短大写前缀；如果还没有插件名，先问用户再写环境变量名。
54. JWT 代码：从环境变量读取 MYPLUGIN_HOST、MYPLUGIN_USERNAME、MYPLUGIN_PASSWORD、MYPLUGIN_TOKEN。
55. JWT 代码：_fetch_fresh_token 使用 requests.post(f'{AIRFLOW_HOST}/auth/token', json={'username': AIRFLOW_USER, 'password': AIRFLOW_PASS}, timeout=10)，raise_for_status()，返回 response.json()['access_token']。
56. JWT 代码：_get_token 如果 AIRFLOW_TOKEN 存在则直接返回。
57. JWT 代码：否则使用 fast path，无锁检查 _cached_token 和 now < _token_expires_at。
58. JWT 代码：slow path 加 _token_lock，再次检查，然后 _cached_token = _fetch_fresh_token()，_token_expires_at = now + 55 * 60。
59. JWT 代码：_make_config 返回 airflow_sdk.Configuration(host=AIRFLOW_HOST)，并设置 config.access_token = _get_token()。
60. 实现认证后告诉用户：本地开发在 .env 设置 MYPLUGIN_USERNAME 和 MYPLUGIN_PASSWORD，JWT 交换自动进行。
61. 实现认证后告诉用户：Astronomer Astro 生产环境创建 Deployment API token 并设为 MYPLUGIN_TOKEN，跳过 JWT 交换。
62. Astronomer Astro 创建 token 步骤：Astro UI -> 打开 Deployment -> Access -> API Tokens -> + Deployment API Token。
63. 复制 token 值（只显示一次）。
64. 运行 astro deployment variable create MYPLUGIN_TOKEN=<token>。
65. Astronomer Astro 上不需要 MYPLUGIN_USERNAME 和 MYPLUGIN_PASSWORD。
66. 包装 SDK 调用：定义普通内部 def _fetch() 包含所有 SDK 逻辑，然后 await asyncio.to_thread(_fetch)。
67. 示例 list_dags：with airflow_sdk.ApiClient(_make_config()) as client: return DAGApi(client).get_dags(limit=100).dags；然后 await asyncio.to_thread(_fetch)，返回所需字段，异常转 HTTPException(status_code=500, detail=str(e))。
68. API 字段名：绝不猜响应字段名，查 REST API reference。
69. 关键 DAGResponse 字段：dag_id、dag_display_name、description、is_paused、timetable_summary、timetable_description、fileloc、owners、tags。
70. 替代直接数据库访问：警告 Airflow metadb 不是公共接口；直接写或不良查询会损坏 scheduler 状态。
71. 使用直接 DB 访问时，明确告诉用户：这直接访问 Airflow 内部数据库；内部模型不是公共 API，可能随版本变化，错误查询会导致 metadb 问题；除非 REST API 未暴露，否则优先 apache-airflow-client。
72. 直接 DB 访问只用于 REST API 未暴露的读操作，或额外 HTTP 开销确实重要时。
73. 直接 DB 访问时，FastAPI 插件端点在 API server 进程内，可直接访问 Airflow 内部 SQLAlchemy 模型，无需 HTTP 往返或 JWT。
74. 直接 DB 访问时，总是用 asyncio.to_thread() 包装 DB 调用，因为 SQLAlchemy 查询是阻塞的。
75. 直接 DB 访问示例：使用 DagBag、DagModel、provide_session，在 _fetch() 内定义 @provide_session 的 _query(session=None)，统计总 DAG 和 paused 数量，然后 await asyncio.to_thread(_fetch)。
76. 如果需要未在示例中的 SDK 方法或字段，生成代码前验证，不要猜。
77. 验证 SDK 方法：运行 python3 -c "from airflow_client.client.api import <Class>; print([m for m in dir(<Class>) if not m.startswith('_')])"（在安装 SDK 的环境里），或在 apache/airflow-client-python 仓库搜索类定义。
78. 常见端点：触发 DAG 用 POST /api/dags/{dag_id}/trigger，DagRunApi.trigger_dag_run(dag_id, TriggerDAGRunPostBody())，返回 run_id 和 normalize_state(result.state)。
79. 常见端点：暂停/恢复 DAG 用 PATCH /api/dags/{dag_id}/pause，DAGApi.patch_dag(dag_id, DAGPatchBody(is_paused=is_paused))，返回 dag_id 和 is_paused。
80. 常见端点：删除 DAG 用 DELETE /api/dags/{dag_id}，DAGApi.delete_dag(dag_id)，返回 deleted。
81. 使用 normalize_state 将 SDK enum 对象转成普通字符串再发前端；raw 为 None 返回 'never_run'，否则 str(raw).lower()。
82. 常见端点：最新 DAG run 用 GET /api/dags/{dag_id}/runs/latest，DagRunApi.get_dag_runs(dag_id, limit=1, order_by='-start_date').dag_runs，取第一个；没有则返回 {'state': 'never_run'}。
83. 常见端点：某次 run 的 task instances 用 GET /api/dags/{dag_id}/runs/{run_id}/tasks，TaskInstanceApi.get_task_instances(dag_id, run_id).task_instances。
84. 常见端点：任务日志用 GET /api/dags/{dag_id}/runs/{run_id}/tasks/{task_id}/logs/{try_number}，TaskInstanceApi.get_log(dag_id, run_id, task_id, try_number, map_index=-1)；try_number 从 1 开始；返回 result.content 或 str(result)。
85. 流式代理：用 StreamingResponse 代理外部 URL 的二进制内容，适用于浏览器不能直接取（CORS、auth 等）。
86. 流式代理示例：GET /api/files/{filename}，在 asyncio.to_thread 中执行 requests.get(f'https://files.example.com/{filename}', stream=True)，raise_for_status()，然后返回 StreamingResponse(response.iter_content(chunk_size=8192), media_type='application/octet-stream', headers={'Content-Disposition': f'attachment; filename="{filename}"'})。
87. 注意 requests.get() 是阻塞的，必须在 asyncio.to_thread 中获取，避免等待远程服务器时阻塞事件循环。
88. Macros 由 scheduler（和 DAG processor）加载，不是 API server；修改后重启 scheduler。
89. 定义 macros：普通 Python 函数，如 format_confidence(confidence: float) -> str；class MyPlugin(AirflowPlugin) 中 macros = [format_confidence]。
90. 使用 macros：在任意模板字段用 {{ macros.my_plugin.format_confidence(0.95) }}，也可结合 ti.xcom_pull，如 {{ macros.my_plugin.format_confidence(ti.xcom_pull(task_ids='score_task')['confidence']) }}；命名模式 macros.{plugin_name}.{function_name}。
91. Middleware 应用到所有 Airflow API 请求，包括内置 REST API 和任意 FastAPI 插件；谨慎使用，并按需显式过滤请求。
92. 定义 middleware：继承 BaseHTTPMiddleware，实现 async dispatch(self, request: Request, call_next) -> Response，在调用前后运行逻辑。
93. 注册 middleware：fastapi_root_middlewares = [{'middleware': AuditMiddleware, 'args': [], 'kwargs': {}, 'name': 'Audit'}]。
94. 如果需要过滤，在 dispatch() 内按 request.url.path 过滤。
95. Operator extra links：继承 BaseOperatorLink，设置 name，如 'Open in Dashboard'。
96. 实现 get_link(self, operator, *, ti_key, **context) -> str，返回 URL，如 f'https://my-dashboard.example.com/tasks/{ti_key.task_id}'。
97. 注册 operator extra links：global_operator_extra_links = [MyDashboardLink()] 出现在每个 task；或 operator_extra_links = [MyDashboardLink()] 附加到特定 operator。
98. React apps：React app 作为 JavaScript bundle 通过 FastAPI 提供。
99. Bundle 必须暴露为与 plugin name 匹配的全局变量：globalThis['My Plugin'] = MyComponent；globalThis.AirflowPlugin = MyComponent 作为 fallback。
100. 注册 React app：fastapi_apps 提供 app 和 url_prefix；react_apps = [{'name': 'My Plugin', 'bundle_url': '/my-plugin/my-app.js', 'destination': 'nav', 'category': 'browse', 'url_route': 'my-plugin'}]。
101. 同一 bundle 可注册到多个 destinations，添加多个 entries，每个需要唯一 url_route，如 nav 和 dag 各一个。
102. React app 集成在 Airflow 3.1 是实验性的，接口可能在未来版本变化。
103. 环境变量和部署：绝不硬编码凭据。
104. 使用 os.environ.get 读取 MYPLUGIN_HOST、MYPLUGIN_USERNAME、MYPLUGIN_PASSWORD，默认如 http://localhost:8080、admin、admin。
105. 本地 Astro CLI：在 .env 设置 MYPLUGIN_HOST=http://localhost:8080、MYPLUGIN_USERNAME=admin、MYPLUGIN_PASSWORD=admin。
106. 任何 Python 插件变更后运行 astro dev restart。
107. 按组件检查日志：astro dev logs --api-server 查看 FastAPI apps、external_views，插件导入错误显示在这里。
108. astro dev logs --scheduler 查看 macros、timetables、listeners、operator links。
109. astro dev logs --dag-processor 查看 DAG 解析错误。
110. 非 Astro 环境运行 airflow plugins 列出所有加载的插件。
111. 生产 Astronomer 用 astro deployment variable create --deployment-id <id> MYPLUGIN_HOST=https://airflow.example.com 设置变量。
112. 开发时自动重载：设置 AIRFLOW__CORE__LAZY_LOAD_PLUGINS=False，跳过惰性加载。
113. 部署后为静态文件做缓存破坏，如 <script src='static/app.js?v=20240315-1'></script>。
114. 验证插件加载：打开 Admin > Plugins。
115. FastAPI 插件自动生成 OpenAPI 文档：Swagger UI 在 {AIRFLOW_HOST}/{url_prefix}/docs，OpenAPI JSON 在 {AIRFLOW_HOST}/{url_prefix}/openapi.json。
116. 避免 nav link 404：href 不要前导 /，用 'my-plugin/ui' 而不是 '/my-plugin/ui'。
117. 避免 nav icon 不显示：icon 需要前导 /，用 '/my-plugin/static/icon.svg'。
118. 避免高负载下事件循环冻结：不要在 async def 直接调用同步 SDK，用 asyncio.to_thread() 包装。
119. 避免 1 小时后 401：JWT 过期无刷新，使用 5 分钟预过期刷新模式。
120. 避免 StaticFiles 启动抛错：目录缺失，启动前创建 assets/ 和 static/。
121. 避免插件不显示：Python 文件变更未重启，运行 astro dev restart。
122. 避免端点未登录可访问：FastAPI apps 不自动认证，如需私有添加 FastAPI security（如 OAuth2、API key）。
123. 避免中间件影响错误路由：中间件应用到所有 API 流量，在 dispatch() 内按 request.url.path 过滤。
124. 避免 JS fetch() 在 Astro 坏：fetch() 中用绝对路径，改为相对路径 fetch('api/dags')。

适用边界 · 注意事项

• Airflow 版本低于 3.1 或仍使用 Airflow 2 的 Flask/appbuilder_views/flask_blueprints 模式时不适用。
• 不希望或不能重启 API server 加载 Python 插件变更时不适用，因为 Python 插件变更必须重启 API server。
• 只需要独立于 Airflow 进程的外部服务，不需要在 Airflow UI/API 进程内嵌入自定义界面或端点时不适用。