SuperAPI AI 工具集

airflow-plugins

部署与运维

查看项目

更新于 10-15 12:18

技能介绍

Airflow 3 插件允许将 FastAPI 应用、React UI、中间件、宏、操作符按钮和自定义时间表直接嵌入 Airflow 进程,无需 sidecar 或额外服务器。

【版本与兼容性】 插件组件(fastapi_apps、react_apps、external_views)要求 Airflow 3.1+。禁止导入 flask、flask_appbuilder,或使用 appbuilder_views / flask_blueprints,这些是 Airflow 2 的模式,在 Airflow 3 中无法工作。若现有代码使用它们,需用 FastAPI 重写整个注册块。

【安全与运行注意】 FastAPI 插件端点不会自动受 Airflow 认证保护。若端点需要私有,必须使用 FastAPI 的安全工具显式实现认证。修改 Python 插件文件需要重启 API 服务器;静态文件(HTML、JS、CSS)变更会立即生效。开发期间可设置 AIRFLOW__CORE__LAZY_LOAD_PLUGINS=False,使插件在启动时加载而非惰性加载。

【路径规则】 external_views 中的 href 不能有前导斜杠。HTML 和 JavaScript 中所有资源与 fetch() 调用应使用相对路径,绝对路径在反向代理后会失效。

【编写代码前的检查项】 是否使用 fastapi_apps / FastAPI 而非 appbuilder_views / Flask;所有 HTML/JS 资源路径与 fetch() 调用是否为相对路径(无前导斜杠);所有同步 SDK 或 SQLAlchemy 调用是否用 asyncio.to_thread() 包装;static/ 和 assets/ 目录是否在 FastAPI 应用挂载前已存在;若端点需私有,是否添加了显式 FastAPI 认证。

【插件组件类型】 单个插件类可同时注册多种组件:fastapi_apps(挂载在 Airflow 进程中的 FastAPI 应用,提供自定义 API 端点)、external_views(以 iframe 嵌入 URL 或外链,用于导航/页面链接)、react_apps(嵌入 Airflow UI 的自定义 React 应用)、fastapi_root_middlewares(拦截所有 Airflow API 请求/响应的中间件)、macros(DAG 模板中可复用的 Python 函数)、operator_extra_links / global_operator_extra_links(任务详情视图中的额外链接按钮)、timetables(自定义调度逻辑)、listeners(Airflow 事件监听回调)。

【项目结构】 建议每个插件在 plugins/ 下拥有独立子目录,将 Python 文件、静态资源和模板放在一起。例如 plugins/my-plugin/ 下包含 plugin.py、static/、assets/。plugin.py 中 BASE_DIR = Path(file).parent 解析到 plugins/my-plugin/。必须在启动 Airflow 前创建子目录及 static/assets 文件夹,否则 StaticFiles 会在导入时抛错。

【注册骨架】 通过继承 AirflowPlugin 定义插件类,设置 name,并配置 fastapi_apps(含 app、url_prefix、name)和 external_views(含 name、href、destination、category、url_route、icon)。external_views 的 destination 可选 "nav"(左侧导航栏,需同时设置 category)、"dag"、"dag_run"、"task"、"task_instance"。nav 的 category 可选 "browse"、"admin" 或省略为顶级。href 可为内部相对路径或完整外部 URL;仅含 external_views 而无 fastapi_apps 的插件也是有效的。

【UI 入口与 API 调用】 可用 FileResponse 提供 UI 入口。若插件需调用 Airflow REST API,需添加 apache-airflow-client 依赖(根据 requirements.txt 或 pyproject.toml 处理)。SDK 是同步的而 FastAPI 路由是异步的,绝不能在 async def 中直接调用阻塞 SDK 方法。JWT 令牌管理建议每进程缓存一个令牌,在 1 小时过期前 5 分钟刷新,使用双重检查锁定。环境变量前缀应基于插件名生成。本地开发设置用户名密码,Astronomer Astro 生产环境使用 Deployment API Token。SDK 调用需用 asyncio.to_thread() 包装。

【直接数据库访问】 FastAPI 插件端点运行在 API 服务器进程内,可直接访问 Airflow 内部 SQLAlchemy 模型。但 Airflow 元数据库不是公共接口,直接写入或不当查询可能损坏调度器状态,内部模型可能随版本变化,建议优先使用 apache-airflow-client,仅用于 REST API 未暴露的读操作。DB 调用同样需用 asyncio.to_thread() 包装。

【常见端点模式】 包括触发 DAG、暂停/恢复 DAG、删除 DAG、获取最新运行、获取任务实例、获取任务日志等。SDK 枚举对象需转换为普通字符串再发送给前端。还可使用 StreamingResponse 代理外部 URL 的二进制内容。

【限制】 插件组件需 Airflow 3.1+;FastAPI 端点默认无认证保护;Python 插件文件变更需重启 API 服务器;元数据库非公共接口,直接访问有风险;SDK 字段名需对照 REST API 参考验证,不可猜测。