Apache Airflow 系列教程 | 番外篇：通过 REST API 动态创建 DAG
导读（Introduction）
在 Apache Airflow 的标准使用模式中，DAG 的定义通常以 Python 文件的形式存放在 DAGs 文件夹中，由 DagFileProcessorManager 周期性解析并持久化到数据库。然而在实际的平台化场景中，用户往往希望通过 Web 界面或 API 接口以更友好的方式（如填写表单、提交 JSON 配置）来动态创建工作流，而不是手写 Python 代码。
本文将设计并实现一个通过 REST API 创建 DAG的扩展功能。我们将深入分析 Airflow 3.x 现有的 FastAPI 路由体系、DAG 序列化机制和数据库模型结构，设计一个完整的技术方案，使得通过 HTTP POST 请求提交的 JSON DAG 定义能够被转换为 Airflow DAG 对象、序列化并持久化到数据库，最终被 Scheduler 正常调度执行。
注意：本文所有代码产出仅作为技术方案参考，不修改 Airflow 源代码。实际部署时可作为独立的 Airflow Plugin 或扩展模块使用。
学习目标（Learning Objectives）
完成本番外篇后，你将能够：
- 理解 Airflow 3.x 中 DAG 从 Python 对象到数据库持久化的完整链路
- 掌握
LazyDeserializedDAG、SerializedDagModel、DagModel之间的关系 - 设计符合 Airflow FastAPI 风格的 REST API 接口
- 实现 JSON DAG 定义到 Airflow DAG 对象的转换逻辑
- 理解如何绕过文件系统解析，直接将 DAG 写入数据库
- 确保 API 创建的 DAG 能被现有 Scheduler 正常调度
正文内容（Main Content）
1. 现有架构分析
1.1 DAG 持久化的标准路径
在 Airflow 3.x 中，DAG 从文件到数据库的标准持久化路径如下：
Python 文件 → DagFileProcessorManager → DagBag.process_file() → DAG 对象 → DagSerialization.to_dict(dag) → LazyDeserializedDAG → SerializedDagModel.write_dag() → 数据库 (serialized_dag 表) → DagModel (dag 表) → DagVersion + DagCode关键组件的职责：
| 组件 | 职责 |
|---|---|
DagSerialization.to_dict(dag) | 将 DAG 对象序列化为字典格式 |
LazyDeserializedDAG | 轻量级 DAG 代理，延迟反序列化 |
SerializedDagModel.write_dag() | 将序列化数据写入serialized_dag表 |
DagModel | 存储 DAG 元数据（调度状态、标签等） |
DagVersion | DAG 版本管理 |
DagCode | 存储 DAG 源代码 |
1.2 数据库模型结构
DagModel(dag表) 的关键字段：
# 源码位置: airflow-core/src/airflow/models/dag.py:340classDagModel(Base):__tablename__="dag"dag_id:Mapped[str]# 主键is_paused:Mapped[bool]# 是否暂停is_stale:Mapped[bool]# 是否过期last_parsed_time:Mapped[datetime]# 最后解析时间fileloc:Mapped[str|None]# 文件路径bundle_name:Mapped[str]# Bundle 名称（外键）bundle_version:Mapped[str|None]# Bundle 版本owners:Mapped[str|None]# DAG 拥有者description:Mapped[str|None]# 描述timetable_type:Mapped[str]# 时间表类型max_active_tasks:Mapped[int]# 最大活跃任务数max_active_runs:Mapped[int|None]# 最大活跃运行数next_dagrun:Mapped[datetime|None]# 下次执行时间next_dagrun_create_after:Mapped[datetime|None]# 下次执行创建时间SerializedDagModel(serialized_dag表) 的关键字段：
# 源码位置: airflow-core/src/airflow/models/serialized_dag.py:281classSerializedDagModel(Base):__tablename__="serialized_dag"id:Mapped[UUID]# 主键 (UUID7)dag_id:Mapped[str]# DAG ID_data:Mapped[dict|None]# JSON 序列化数据_data_compressed:Mapped[bytes|None]# 压缩的序列化数据created_at:Mapped[datetime]# 创建时间last_updated:Mapped[datetime]# 最后更新时间dag_hash:Mapped[str]# 数据哈希值 (MD5)dag_version_id:Mapped[UUID]# 关联的 DagVersion1.3 现有 API 路由模式
Airflow 3.x 使用 FastAPI 构建 REST API，路由定义遵循统一模式：
# 源码位置: airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.pyconnections_router=AirflowRouter(tags=["Connection"],prefix="/connections")@connections_router.post("",status_code=status.HTTP_201_CREATED,responses=create_openapi_http_exception_doc([status.HTTP_409_CONFLICT]),dependencies=[Depends(requires_access_connection(method="POST")),Depends(action_logging())],)defpost_connection(post_body:ConnectionBody,session:SessionDep,)->ConnectionResponse:"""Create connection entry."""connection=Connection(**post_body.model_dump(by_alias=True))session.add(connection)returnconnection2. 技术方案设计
2.1 整体架构
我们的方案绕过文件解析流程，直接通过 API 接收 JSON DAG 定义，在服务端构建 DAG 对象并持久化到数据库：
HTTP POST (JSON) → FastAPI 路由 → Pydantic 校验 → 构建 DAG 对象 → DagSerialization.to_dict() → LazyDeserializedDAG → SerializedDagModel.write_dag() → 创建/更新 DagModel → Scheduler 自动发现并调度2.2 关键设计决策
| 决策点 | 选择 | 理由 |
|---|---|---|
| Bundle 归属 | 使用专用的api_createdbundle | 区分 API 创建的 DAG 与文件系统 DAG |
| 序列化方式 | 复用DagSerialization.to_dict() | 保证与现有系统完全兼容 |
| Operator 支持 | 初始支持 BashOperator、PythonOperator、EmptyOperator | 可扩展设计 |
| 调度集成 | 写入标准数据库表 | Scheduler 无需修改即可调度 |
| 幂等性 | 基于 dag_id 进行 upsert | 重复提交不会创建重复 DAG |
2.3 JSON DAG 定义格式
{"dag_id":"api_created_etl_pipeline","description":"通过 API 创建的 ETL 工作流","schedule":"0 2 * * *","start_date":"2024-01-01T00:00:00+00:00","end_date":null,"catchup":false,"tags":["etl","api-created"],"default_args":{"owner":"data-team","retries":2,"retry_delay_seconds":300},"max_active_tasks":16,"max_active_runs":1,"tasks":[{"task_id":"extract","operator":"BashOperator","params":{"bash_command":"echo 'extracting data'"},"downstream":["transform"]},{"task_id":"transform","operator":"PythonOperator","params":{"python_callable_name":"my_module.transform_func","python_callable_source":"def transform_func():\n print('transforming')"},"downstream":["load"]},{"task_id":"load","operator":"BashOperator","params":{"bash_command":"echo 'loading data'"},"downstream":[]}]}3. 完整实现代码
3.1 Pydantic 数据模型定义
# file: airflow_api_dag_creator/datamodels.py""" Pydantic 数据模型：定义 API 请求/响应的 JSON 结构。 """from__future__importannotationsfromdatetimeimportdatetimefromtypingimportAnyfrompydanticimportBaseModel,Field,field_validatorclassTaskDefinition(BaseModel):"""单个任务的定义。"""task_id:str=Field(...,description="任务唯一标识",pattern=r"^[a-zA-Z_][a-zA-Z0-9_\-\.]*$")operator:str=Field(...,description="Operator 类型名称")params:dict[str,Any]=Field(default_factory=dict,description="Operator 参数")downstream:list[str]=Field(default_factory=list,description="下游任务 ID 列表")retries:int|None=Field(None,description="重试次数")retry_delay_seconds:int|None=Field(None,description="重试间隔（秒）")trigger_rule:str=Field("all_success",description="触发规则")pool:str=Field("default_pool",description="资源池名称")@field_validator("operator")@classmethoddefvalidate_operator(cls,v:str)->str:allowed_operators={"BashOperator","PythonOperator","EmptyOperator","EmailOperator",}ifvnotinallowed_operators:raiseValueError(f"不支持的 Operator:{v}。当前支持:{', '.join(sorted(allowed_operators))}")returnvclassDefaultArgs(BaseModel):"""DAG 默认参数。"""owner:str=Field("airflow",description="DAG 拥有者")retries:int=Field(0,ge=0,le=10,description="默认重试次数")retry_delay_seconds:int=Field(300,ge=0,description="默认重试间隔（秒）")email:list[str]|None=Field(None,description="告警邮件列表")email_on_failure:bool=Field(False,description="失败时是否发送邮件")email_on_retry:bool=Field(False,description="重试时是否发送邮件")classDagCreateRequest(BaseModel):"""创建 DAG 的请求体。"""dag_id:str=Field(...,description="DAG 唯一标识符",pattern=r"^[a-zA-Z_][a-zA-Z0-9_\-\.]*$",max_length=250,)description:str|None=Field(None,description="DAG 描述")schedule:str|No