Apache Airflow 新增 REST API 端点完全指南从 FastAPI 路由实现到 prek 钩子验证【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本指南面向希望在 Apache Airflow 3当前仓库的 REST API 中新增端点的开发者完整讲解从接口设计决策、FastAPI 路由实现、Pydantic 数据模型定义、单元测试编写到 prek 钩子自动更新 OpenAPI 规范的端到端流程。读完本文你将掌握public与ui两套 API 的选型标准能够基于 AirflowRouter 注册带权限校验与查询参数的路由并让新端点自动进入 v2-rest-api-generated.yaml 与交互式文档。一、先理解 Airflow 3 的 API 分层public 与 uiAirflow 3 的 REST API 全部构建在 FastAPI 之上核心源码位于airflow-core/src/airflow/api_fastapi/core_api其中routes/public公共 API 端点路径以/api/v2为前缀。它们经过标准化、文档完善且保证向后兼容。外部用户、SDK 与集成方应只依赖这一层。routes/ui专门为前端Airflow UI定制的端点路径以/ui为前缀不承诺向后兼容可随时根据前端需要调整外部不应依赖。该分层在仓库中有着明确的落地证据生成的 OpenAPI 规范文件 v2-rest-api-generated.yaml 的info.description中写道/api/v2下的端点可以安全使用、稳定且向后兼容而/ui下的端点专为 UI 服务可能随前端需求发生破坏性变更。同时在 ui/init.py 中ui_router注册时显式设置了include_in_schemaFalse这意味着 UI 端点不会出现在公开的 OpenAPI schema 文档中。选型建议新增端点时应尽最大可能使其可被社区复用设计稳定、标准化因此优先放入public只有当数据类型过于特殊或频繁变动典型如 Grid、Gantt、Calendar 等页面专用数据结构时才放入ui。仓库中的 public 路由目录含dags.py、connections.py、variables.py、task_instances.py、xcom.py等 31 个模块与 ui 路由目录含grid.py、gantt.py、calendar.py、dashboard.py、dependencies.py等 16 个模块的对比就是这一决策的直观样例。二、Step 1实现端点逻辑2.1 确定接口归属并创建路由根据上面的分层决策导航到airflow-core/src/airflow/api_fastapi/core_api/routes/public或airflow-core/src/airflow/api_fastapi/core_api/routes/ui下的对应模块使用AirflowRouter注册新路由。AirflowRouter是 FastAPIAPIRouter的轻量扩展位于 airflow-core/src/airflow/api_fastapi/common/router.py其核心差异在于若不显式传operation_id会自动使用函数名作为operation_id这保证了 OpenAPI 规范中操作标识符的确定性。原文档给出的路由骨架如下dags_router.get(/dags) # permissions go in the dependencies parameter here async def get_dags( *, limit: int 100, offset: int 0, tags: Annotated[list[str] | None, Query()] None, dag_id_pattern: str | None None, only_active: bool True, paused: bool | None None, order_by: str dag_id, session: SessionDep, ) - DagCollectionResponse: pass2.2 对照真实源码理解各要素仓库中 public/dags.py 的get_dags是该骨架在生产环境的完整形态可逐项对照dags_router AirflowRouter(tags[DAG], prefix/dags) dags_router.get(, dependencies[Depends(requires_access_dag(methodGET))]) def get_dags( limit: QueryLimit, offset: QueryOffset, tags: QueryTagsFilter, dag_id_pattern: QueryDagIdPatternSearch, paused: QueryPausedFilter, ... order_by: Annotated[SortParam, Depends(...)], readable_dags_filter: ReadableDagsFilterDep, session: SessionDep, ) - DAGCollectionResponse: ...要点拆解HTTP 方法与路径dags_router.get()配合prefix/dags实际对外路径为/api/v2/dags。同文件中还演示了get、patch、post、delete各种方法的使用如patch_dag用dags_router.patch(/{dag_id})favorite_dag用dags_router.post(/{dag_id}/favorite)。权限声明放在装饰器的dependencies参数中。requires_access_dag(methodGET)是 security.py 提供的工厂函数它从请求的 path/query 中提取dag_id调用 AuthManager 的is_authorized_dag完成鉴权写操作需对应methodPUT、methodDELETE。查询参数的类型化生产代码不使用裸类型而是复用airflow.api_fastapi.common.parameters中的预定义类型例如QueryLimit、QueryOffset、QueryTagsFilter、QueryDagIdPatternSearch它们封装了默认值、校验规则与 OpenAPI 描述排序参数通过SortParam(...).dynamic_depends()动态构造Depends。可读性过滤ReadableDagsFilterDep对应 security.py 的ReadableDagsFilterDep Annotated[PermittedDagFilter, Depends(permitted_dag_filter_factory(GET))]会将当前用户无权读取的 DAG 在 SQL 层面过滤掉保证存在性不泄露。数据库会话session: SessionDep是 FastAPI 依赖注入的 SQLAlchemy 会话所有查询都经由它执行配合paginated_select统一实现分页与total_entries统计。返回类型注解- DagCollectionResponse是必须的FastAPI 依据该注解生成 OpenAPI 的 response schema。2.3 错误响应的 OpenAPI 文档化对于可能抛出多种 HTTP 异常的端点仓库推荐使用create_openapi_http_exception_doc显式声明响应码例如同文件中get_dag的定义dags_router.get( /{dag_id}, responsescreate_openapi_http_exception_doc( [status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_CONTENT] ), dependencies[Depends(requires_access_dag(methodGET))], )该工具函数位于 core_api/openapi/exceptions.py它会为每个列出的状态码生成符合 Airflow 统一错误结构的文档让 API 消费者提前知晓错误语义如 404 表示资源不存在、409 表示存在冲突如任务实例仍在运行。三、Step 2为端点编写测试3.1 手动验证在编写测试前先在本地启动 API 服务用 curl 或 FastAPI 自带文档手动验证端点行为符合预期包括查询参数组合、权限拦截与异常分支。3.2 初始化单元测试测试目录结构与源码一一对应公共端点测试位于 airflow-core/tests/unit/api_fastapi/core_api/routes/public每个端点模块都有对应测试文件如test_dags.py、test_connections.py、test_variables.py、test_xcom.py、test_task_instances.py等 30 个文件UI 端点测试位于 airflow-core/tests/unit/api_fastapi/core_api/routes/ui。测试应覆盖以下维度原文档要求extensive tests查询参数合法值、边界值、非法值应返回 422 校验错误权限无认证访问返回 401无授权访问返回 403不同 method 的权限差异错误处理资源不存在返回 404、并发/状态冲突返回 409 等。3.3 全局路由约束测试仓库还有一个值得借鉴的全局测试 test_routes.py它从public_router与authenticated_router的注册表出发做结构性断言test_no_auth_routes验证除NO_AUTH_PATHS/api/v2/auth/login、/api/v2/auth/logout、/api/v2/version、/api/v2/monitor/health外所有路由都必须经过认证test_routes_with_responses验证每个受保护路由都声明了 401/403 响应防止新增路由时意外漏掉鉴权。这意味着你新增的受保护端点若未配置 401/403 响应文档这一套测试会直接失败——这是 Airflow 通过测试强制保障 API 安全一致性的机制。四、Step 3文档自动生成与核对Airflow 的 API 文档由 FastAPI 自动生成并经过 prek 钩子见下一节固化。新增端点后启动服务并访问/docsFastAPI 内置 Swagger UI核对新端点的请求体类型、返回类型、查询参数、校验规则是否清晰、完整、符合预期确认错误响应码400/401/403/404/409/422已在文档中正确呈现。在仓库中这份自动生成的交互式文档的数据源就是 v2-rest-api-generated.yaml当前仓库中约 1.7 万行、覆盖所有 public 端点它同时被前端代码生成工具消费。五、Step 4运行 prek 钩子完成质量门禁5.1 prek 是什么prek是 Apache Airflow 项目的静态检查与代码生成工具链其依赖声明可见于 dev/breeze/pyproject.toml 中的prek0.4.14涵盖静态代码检查、格式化以及各类生成文件的校验。新增 API 端点后必须运行它以通过 CI 门禁。执行全部 prek 钩子prek --all-files5.2 OpenAPI 规范的持久化更新新增端点后持久化的 OpenAPI 规范文件v2-rest-api-generated.yaml需要同步更新。这一步由专用的 prek 钩子自动完成你只需运行 prek 钩子它会基于当前 FastAPI 应用重新生成规范文件将生成的变更git add并提交。该钩子的实现链路是scripts/ci/prek/generate_openapi_spec.py 通过 Breeze 容器调用 scripts/in_container/run_generate_openapi_spec.py后者以SimpleAuthManager初始化 FastAPI 应用并调用generate_openapi_file(appcreate_app(), file_pathOPENAPI_SPEC_FILE)写回 v2-rest-api-generated.yaml同时生成用于 UI 代码生成的私有规范_private_ui.yaml与 SimpleAuthManager 专属规范。这意味着你无需手写任何 OpenAPI YAML规范文件始终与代码保持同步。六、可选添加 Pydantic 模型当新端点涉及全新的数据结构时需要定义对应的 Pydantic 模型用于校验、序列化/反序列化请求与响应。原文档给出的最小示例class DagModelResponse(BaseModel): Dag serializer for responses. dag_id: str dag_display_name: str is_paused: bool is_active: bool last_parsed_time: datetime | None6.1 生产环境中的模型形态仓库中的模型比示例更复杂、更具参考性。响应模型统一放在airflow-core/src/airflow/api_fastapi/core_api/datamodels下以 datamodels/dags.py 为例DAGResponse核心序列化模型声明dag_id、dag_display_name、is_paused、is_stale、last_parsed_time等 20 字段通过field_validator将owners由逗号分隔字符串归一为列表、computed_field计算is_backfillable与file_token还通过AliasGenerator处理响应字段与模型字段的命名映射如next_dagrun_logical_date→next_dagrun。DAGCollectionResponse集合响应统一为dags: Iterable[DAGResponse]total_entries: int的结构这是全仓库所有列表端点Connections、Variables、Assets 等的通用约定可从 datamodels 目录 中大量XxxCollectionResponse得到印证。DAGPatchBody/DAGPatchBodyPartial请求体模型DAGPatchBodyPartial由make_partial_model工具生成全部字段可选的变体用于支持update_mask局部更新语义。6.2 模型自动进入 OpenAPI这些模型只要被某个端点实际引用作为参数或返回类型注解就会自动出现在 OpenAPI 规范文件中无需手动登记。因此新增或修改模型后重新运行 prek 钩子以更新所有生成文件包括v2-rest-api-generated.yaml中的components/schemas部分未被任何端点引用的模型不会进入规范——这保证了规范的整洁性。七、把新端点接入应用路由注册机制原文档没有展开、但对实际提交 PR 至关重要的环节是仅定义路由函数还不够必须把 Router 挂载到应用上。完整链路如下在routes/public/your_module.py中创建xxx_router AirflowRouter(...)在 routes/public/init.py 中导入该 router并通过authenticated_router.include_router(xxx_router)挂载——authenticated_router在路由器级别注入了Depends(get_user)作为防呆兜底即使某个路由忘了写自己的鉴权依赖也会被强制要求认证该设计动机在文件注释中有明确说明public_router AirflowRouter(prefix/api/v2)再统一挂载authenticated_router以及无需认证的monitor_router、version_router、auth_router最终由 core_api/app.py 的init_views调用app.include_router(ui_router)与app.include_router(public_router)完成应用级注册同时它还注册了/api/v1与未知/api/*的 404 兜底路由明确提示/api/v1在 Airflow 3 中已被移除、应改用/api/v2。同理UI 端点的注册路径是 routes/ui/init.pyui_router AirflowRouter(prefix/ui, include_in_schemaFalse, dependencies[Depends(get_user)])同样以 router 级依赖保证全部 UI 路由必须认证。八、安全基线新端点必须考虑的三件事结合 core_api/security.py 与 test_routes.py新增端点时请默念以下安全清单认证兜底通过authenticated_routerpublic或 router 级Depends(get_user)ui挂载或至少声明Depends(get_user)使未认证请求返回 401细粒度授权DAG 类资源使用requires_access_dag(method...)并按读写方法区分非 DAG 资源Pool、Connection、Variable 等使用各自的requires_access_*工厂列表类端点注入Readable*FilterDep做行级过滤响应码声明用create_openapi_http_exception_doc声明 401/403 及业务错误码否则test_routes_with_responses这类全局测试会拦截合入。九、小结新增端点的完整检查清单按原文档流程并结合仓库实践一个 API 端点的合入路径可归纳为判断归属标准化、向后兼容 →publicUI 专属、易变 →ui在对应routes/目录实现路由函数选好 HTTP 方法、查询参数类型、dependencies权限、responses错误码、Pydantic 返回类型注解将 Router 挂载到 public/init.py 或 ui/init.py在对应tests/unit/api_fastapi/core_api/routes/下补齐测试查询参数、权限、错误处理全覆盖访问/docs核对文档body/return 类型、query 参数、校验规则运行prek --all-files让钩子自动更新 v2-rest-api-generated.yaml 等生成文件git add后提交。如果你在此基础上进一步调整了 Airflow 的整体架构可参考仓库中的 架构图绘制指南 来同步更新架构文档。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考