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 元数据（调度状态、标签等）DagVersionDAG 版本管理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重复提交不会创建重复 DAG2.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