Airflow如何限制DAG每日手动触发次数 无需定时调度的实现方案

Airflow 目前没有原生参数可以直接实现手动触发的24小时频次限制，你提到的depends_on_past参数仅校验上一个相邻调度实例的执行状态，不限制指定时间窗口内的总触发次数，确实无法匹配你的需求，以下是两种可落地的实现方案：

方案1：DAG内置前置检查节点（轻量化适配单DAG） #

直接在现有DAG的最前面新增一个校验任务，查询该DAG过去24小时内的运行记录，符合拦截条件则直接抛出异常终止DAG运行，无需额外改造Airflow组件。
代码示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import DagRun
from datetime import datetime, timedelta
from airflow.utils.dates import days_ago

def check_trigger_frequency(**context):
    current_dag_id = context["dag"].dag_id
    # 24小时时间阈值，若需要限制自然日单日1次，可替换为当日0点
    time_threshold = datetime.now() - timedelta(hours=24)
    # 查询过去24小时内该DAG的成功运行记录，若要包含运行中/失败的记录可删除state参数
    recent_valid_runs = DagRun.find(
        dag_id=current_dag_id,
        state="success",
        execution_date=time_threshold
    )
    if recent_valid_runs:
        raise Exception(f"该DAG24小时内仅允许触发1次，最近一次成功运行时间为{recent_valid_runs[0].execution_date}，请稍后再试")

with DAG(
    dag_id="your_target_dag",
    schedule_interval=None, # 关闭定时调度，仅允许手动触发
    start_date=days_ago(1),
    catchup=False
) as dag:
    # 校验任务作为第一个节点
    check_task = PythonOperator(
        task_id="check_24h_trigger_limit",
        python_callable=check_trigger_frequency,
        provide_context=True
    )

    # 原有DAG的任务接在校验节点之后
    check_task >> your_original_first_task

• 优势：无额外依赖，改造成本极低，适配单DAG的个性化限制需求
• 可灵活调整规则：修改时间阈值、校验的运行状态即可适配不同的限制要求

方案2：自定义触发校验插件（适合多DAG全局管控） #

如果需要对平台内多个DAG统一配置频次限制规则，可以自定义Airflow插件，扩展DAG参数新增max_runs_per_24h配置项，在DAG触发的前置钩子逻辑中做统一校验，不符合规则的触发请求会被直接拦截，不会生成DAG运行实例。
该方案适合平台级统一管控场景，无需每个DAG单独编写校验逻辑，统一配置参数即可生效。

内容的提问来源于stack exchange，提问作者torxx666