airflow-plugins

Airflow 3 插件

Airflow 3 插件可让你直接在 Airflow 进程中嵌入 FastAPI 应用、React UI、中间件、宏、操作按钮和自定义时间表。无需 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() 调用都使用相对路径。绝对路径在反向代理后无法正常工作。

编写任何代码之前，请确认

1. 我是否使用了 fastapi_apps / FastAPI，而不是 appbuilder_views / Flask？
2. 所有 HTML/JS 资源路径和 fetch() 调用是否都是相对路径（不以斜杠开头）？
3. 所有同步 SDK 或 SQLAlchemy 调用是否都用 asyncio.to_thread() 包装？
4. FastAPI 应用挂载 static/ 和 assets/ 目录之前，这些目录是否都已存在？
5. 如果端点必须设为私有，我是否添加了显式的 FastAPI 身份验证？

---

步骤 1：选择插件组件

单个插件类可以同时注册多种组件类型。

组件	功能	字段
自定义 API 端点	在 Airflow 进程中挂载 FastAPI 应用	fastapi_apps
导航 / 页面链接	将 URL 嵌入为 iframe 或链接到外部页面	external_views
React 组件	嵌入 Airflow UI 的自定义 React 应用	react_apps
API 中间件	拦截所有 Airflow API 请求/响应	fastapi_root_middlewares
Jinja 宏	DAG 模板中可复用的 Python 函数	macros
任务实例按钮	任务详情视图中的额外链接按钮	operator_extra_links / global_operator_extra_links
自定义时间表	自定义调度逻辑	timetables
事件钩子	Airflow 事件的监听器回调	listeners

---

步骤 2：插件注册骨架

项目文件结构

为每个插件在 plugins/ 下创建独立子目录——这样 Python 文件、静态资源和模板都能放在一起，也便于管理包含多个插件的项目：

plugins/
  my-plugin/
    plugin.py       # AirflowPlugin subclass — auto-discovered by Airflow
    static/
      index.html
      app.js
    assets/
      icon.svg

plugin.py 中的 BASE_DIR = Path(__file__).parent 会解析为 plugins/my-plugin/——静态文件和资源路径相对于该目录都能正确解析。启动 Airflow 之前，请创建子目录以及任何 static/assets 目录，否则 StaticFiles 会在导入时引发异常。

from pathlib import Path
from airflow.plugins_manager import AirflowPlugin
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse

BASE_DIR = Path(__file__).parent

app = FastAPI(title="My Plugin")

# Both directories must exist before Airflow starts or FastAPI raises on import
app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")
app.mount("/assets", StaticFiles(directory=BASE_DIR / "assets"), name="assets")


class MyPlugin(AirflowPlugin):
    name = "my_plugin"

    fastapi_apps = [
        {
            "app": app,
            "url_prefix": "/my-plugin",   # plugin available at {AIRFLOW_HOST}/my-plugin/
            "name": "My Plugin",
        }
    ]

    external_views = [
        {
            "name": "My Plugin",
            "href": "my-plugin/ui",              # NO leading slash — breaks on Astro and reverse proxies
            "destination": "nav",                # see locations table below
            "category": "browse",                # nav bar category (nav destination only)
            "url_route": "my-plugin",            # unique route name (required for React apps)
            "icon": "/my-plugin/static/icon.svg" # DOES use a leading slash — served by FastAPI
        }
    ]

外部视图位置

destination	显示位置
"nav"	左侧导航栏（也要设置 category）
"dag"	每个 Dag 页面上的额外标签页
"dag_run"	每个 Dag 运行页面上的额外标签页
"task"	每个任务页面上的额外标签页
"task_instance"	每个任务实例页面上的额外标签页

导航栏类别（destination: "nav"）

将 "category" 设为特定的导航分组，让链接显示在该分组下："browse"、"admin"；也可以省略该字段，将链接放在顶层。

外部 URL 和最小插件

href 可以是指向内部端点的相对路径（"my-plugin/ui"），也可以是完整的外部 URL。只包含 external_views 而不包含 fastapi_apps 的插件也是有效的——简单的链接或标签页无需后端：

from airflow.plugins_manager import AirflowPlugin

class LearnViewPlugin(AirflowPlugin):
    name = "learn_view_plugin"

    external_views = [
        {
            "name": "Learn Airflow 3",
            "href": "https://www.astronomer.io/docs/learn",
            "destination": "dag",   # adds a tab to every Dag page
            "url_route": "learn"
        }
    ]

不以斜杠开头的规则仅适用于内部路径——完整的 https:// URL 可以正常使用。

---

第 3 步：提供 UI 入口

@app.get("/ui", response_class=FileResponse)
async def serve_ui():
    return FileResponse(BASE_DIR / "static" / "index.html")

在 HTML 中，始终使用相对路径。如果 Airflow 挂载在子路径下，绝对路径会失效：

<link rel="stylesheet" href="static/app.css" />
<script src="static/app.js?v=20240315"></script>


<script src="/my-plugin/static/app.js"></script>

JavaScript 中也遵循同一规则：

fetch('api/dags')           // correct — relative to current page
fetch('/my-plugin/api/dags') // breaks on Astro and sub-path deploys

---

第 4 步：从插件调用 Airflow API

仅当插件调用 Airflow REST API 时才需要。 仅提供静态文件、注册 external_views 或直接访问数据库的插件不需要此步骤——跳到第 5 步或第 6 步。

添加依赖项

仅在实现 REST API 通信时，才将 apache-airflow-client 添加到项目依赖项中。检查存在哪个文件，并采取相应操作：

找到的文件	操作
requirements.txt	追加 apache-airflow-client
pyproject.toml（uv / poetry）	uv add apache-airflow-client 或 poetry add apache-airflow-client
以上文件均不存在	告诉用户：“运行插件前，请将 apache-airflow-client 添加到依赖项中。”

使用 apache-airflow-client 调用 Airflow 自身的 REST API。该 SDK 是同步的，但 FastAPI 路由是异步的——切勿在 async def 中直接调用会阻塞的 SDK 方法，否则会阻塞事件循环并冻结所有并发请求。

JWT 令牌管理

每个进程缓存一个令牌。在 1 小时有效期到期前 5 分钟刷新。使用双重检查锁定，避免多个并发请求同时争相刷新：

将 MYPLUGIN_ 替换为根据插件名称生成的简短大写前缀（例如，若插件名为 “Trip Analyzer”，则使用 TRIP_ANALYZER_）。如果尚未提供插件名称，请先询问用户，再编写环境变量名称。

import asyncio
import os
import threading
import time
import airflow_client.client as airflow_sdk
import requests

AIRFLOW_HOST  = os.environ.get("MYPLUGIN_HOST",     "http://localhost:8080")
AIRFLOW_USER  = os.environ.get("MYPLUGIN_USERNAME", "admin")
AIRFLOW_PASS  = os.environ.get("MYPLUGIN_PASSWORD", "admin")
AIRFLOW_TOKEN = os.environ.get("MYPLUGIN_TOKEN")    # Astronomer Astro: Deployment API token

_cached_token: str | None = None
_token_expires_at: float  = 0.0
_token_lock = threading.Lock()


def _fetch_fresh_token() -> str:
    """Exchange username/password for a JWT via Airflow's auth endpoint."""
    response = requests.post(
        f"{AIRFLOW_HOST}/auth/token",
        json={"username": AIRFLOW_USER, "password": AIRFLOW_PASS},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()["access_token"]


def _get_token() -> str:
    # Astronomer Astro production: use static Deployment API token directly
    if AIRFLOW_TOKEN:
        return AIRFLOW_TOKEN
    global _cached_token, _token_expires_at
    now = time.monotonic()
    # Fast path — no lock if still valid
    if _cached_token and now < _token_expires_at:
        return _cached_token
    # Slow path — one thread refreshes, others wait
    with _token_lock:
        if _cached_token and now < _token_expires_at:
            return _cached_token
        _cached_token = _fetch_fresh_token()
        _token_expires_at = now + 55 * 60  # refresh 5 min before 1-hour expiry
    return _cached_token


def _make_config() -> airflow_sdk.Configuration:
    config = airflow_sdk.Configuration(host=AIRFLOW_HOST)
    config.access_token = _get_token()
    return config

实现身份验证后，告诉用户：

• 本地开发：在 .env 中设置 MYPLUGIN_USERNAME 和 MYPLUGIN_PASSWORD——JWT 交换会自动进行。

• Astronomer Astro（生产环境）：创建 Deployment API 令牌，并将其设置为 MYPLUGIN_TOKEN——完全跳过 JWT 交换：

  1. Astro UI → 打开 Deployment → Access → API Tokens → + Deployment API Token
  2. 复制令牌值（只显示一次）
  3. astro deployment variable create MYPLUGIN_TOKEN=<token>

  在 Astro 上无需设置 MYPLUGIN_USERNAME 和 MYPLUGIN_PASSWORD。

使用 asyncio.to_thread 封装 SDK 调用

from fastapi import HTTPException
from airflow_client.client.api import DAGApi

@app.get("/api/dags")
async def list_dags():
    try:
        def _fetch():
            with airflow_sdk.ApiClient(_make_config()) as client:
                return DAGApi(client).get_dags(limit=100).dags
        dags = await asyncio.to_thread(_fetch)
        return [{"dag_id": d.dag_id, "is_paused": d.is_paused, "timetable_summary": d.timetable_summary} for d in dags]
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

API 字段名称：切勿猜测响应字段名称——请对照 REST API 参考 验证。DAGResponse 的关键字段：dag_id、dag_display_name、description、is_paused、timetable_summary、timetable_description、fileloc、owners、tags。

始终遵循以下模式：定义一个普通的内部 def _fetch()，包含所有 SDK 逻辑，然后执行 await asyncio.to_thread(_fetch)。

替代方案：直接访问数据库

警告——请谨慎使用，并告知用户。 Airflow 元数据库不是公共接口。直接写入或格式错误的查询可能破坏调度器状态。每当使用此模式时，都要明确告诉用户：“这会直接访问 Airflow 的内部数据库。内部模型不属于公共 API，可能会随 Airflow 版本变化，错误的查询还可能导致元数据库出现问题。除非该操作未通过 REST API 提供，否则请优先使用 apache-airflow-client。”

由于 FastAPI 插件端点运行在API 服务器进程中（而不是任务工作进程中），因此可以直接访问 Airflow 的内部 SQLAlchemy 模型——无需 HTTP 往返或 JWT。仅用于 REST API 未提供的读取操作，或确实需要避免额外 HTTP 开销的情况。始终使用 asyncio.to_thread() 封装数据库调用——SQLAlchemy 查询会阻塞。

from airflow.models import DagBag, DagModel
from airflow.utils.db import provide_session

@app.get("/api/dags/status")
async def dag_status():
    def _fetch():
        @provide_session
        def _query(session=None):
            dagbag = DagBag()
            paused = sum(
                1 for dag_id in dagbag.dags
                if (m := session.query(DagModel).filter(DagModel.dag_id == dag_id).first())
                and m.is_paused
            )
            return {"total": len(dagbag.dags), "paused": paused}
        return _query()
    return await asyncio.to_thread(_fetch)

---

第 5 步：常见 API 端点模式

如果你需要的 SDK 方法或字段未在以下示例中展示，请先验证，再生成代码——不要猜测。可以在安装了 SDK 的任意环境中运行 python3 -c "from airflow_client.client.api import <Class>; print([m for m in dir(<Class>) if not m.startswith('_')])"，也可以在 apache/airflow-client-python 仓库中搜索类定义。

from airflow_client.client.api import DAGApi, DagRunApi
from airflow_client.client.models import TriggerDAGRunPostBody, DAGPatchBody


@app.post("/api/dags/{dag_id}/trigger")
async def trigger_dag(dag_id: str):
    def _run():
        with airflow_sdk.ApiClient(_make_config()) as client:
            return DagRunApi(client).trigger_dag_run(dag_id, TriggerDAGRunPostBody())
    result = await asyncio.to_thread(_run)
    return {"run_id": result.dag_run_id, "state": normalize_state(result.state)}


@app.patch("/api/dags/{dag_id}/pause")
async def toggle_pause(dag_id: str, is_paused: bool):
    def _run():
        with airflow_sdk.ApiClient(_make_config()) as client:
            DAGApi(client).patch_dag(dag_id, DAGPatchBody(is_paused=is_paused))
    await asyncio.to_thread(_run)
    return {"dag_id": dag_id, "is_paused": is_paused}


@app.delete("/api/dags/{dag_id}")
async def delete_dag(dag_id: str):
    def _run():
        with airflow_sdk.ApiClient(_make_config()) as client:
            DAGApi(client).delete_dag(dag_id)
    await asyncio.to_thread(_run)
    return {"deleted": dag_id}


def normalize_state(raw) -> str:
    """Convert SDK enum objects to plain strings before sending to the frontend."""
    if raw is None:
        return "never_run"
    return str(raw).lower()

DAG 运行、任务实例和日志

除了基本的 DAG CRUD 操作外，这些是最常见的调用。如需了解此处未展示的内容，请查阅 REST API 参考，了解可用端点以及对应的 Python SDK 类名和方法名。

from airflow_client.client.api import DagRunApi, TaskInstanceApi

# Latest run for a DAG
@app.get("/api/dags/{dag_id}/runs/latest")
async def latest_run(dag_id: str):
    def _fetch():
        with airflow_sdk.ApiClient(_make_config()) as client:
            runs = DagRunApi(client).get_dag_runs(dag_id, limit=1, order_by="-start_date").dag_runs
            return runs[0] if runs else None
    run = await asyncio.to_thread(_fetch)
    if not run:
        return {"state": "never_run"}
    return {"run_id": run.dag_run_id, "state": normalize_state(run.state)}


# Task instances for a specific run
@app.get("/api/dags/{dag_id}/runs/{run_id}/tasks")
async def task_instances(dag_id: str, run_id: str):
    def _fetch():
        with airflow_sdk.ApiClient(_make_config()) as client:
            return TaskInstanceApi(client).get_task_instances(dag_id, run_id).task_instances
    tasks = await asyncio.to_thread(_fetch)
    return [{"task_id": t.task_id, "state": normalize_state(t.state)} for t in tasks]


# Task log (try_number starts at 1)
@app.get("/api/dags/{dag_id}/runs/{run_id}/tasks/{task_id}/logs/{try_number}")
async def task_log(dag_id: str, run_id: str, task_id: str, try_number: int):
    def _fetch():
        with airflow_sdk.ApiClient(_make_config()) as client:
            return TaskInstanceApi(client).get_log(
                dag_id, run_id, task_id, try_number, map_index=-1
            )
    result = await asyncio.to_thread(_fetch)
    return {"log": result.content if hasattr(result, "content") else str(result)}

流式代理

使用 StreamingResponse 通过插件代理来自外部 URL 的二进制内容——适用于浏览器无法直接获取资源的情况（如 CORS、身份验证等）：

import requests
from starlette.responses import StreamingResponse

@app.get("/api/files/{filename}")
async def proxy_file(filename: str):
    def _stream():
        r = requests.get(f"https://files.example.com/{filename}", stream=True)
        r.raise_for_status()
        return r
    response = await asyncio.to_thread(_stream)
    return StreamingResponse(
        response.iter_content(chunk_size=8192),
        media_type="application/octet-stream",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )

请注意，requests.get() 是阻塞操作——应在 asyncio.to_thread 中获取内容，以免等待远程服务时阻塞事件循环。

---

第 6 步：其他插件组件类型

宏

宏由调度器（和 DAG 处理器）加载，而非 API 服务。修改后请重启调度器。

from airflow.plugins_manager import AirflowPlugin

def format_confidence(confidence: float) -> str:
    return f"{confidence * 100:.2f}%"

class MyPlugin(AirflowPlugin):
    name = "my_plugin"
    macros = [format_confidence]

可在任何模板化字段中使用，包括与 XCom 配合使用：

{{ macros.my_plugin.format_confidence(0.95) }}

{{ macros.my_plugin.format_confidence(ti.xcom_pull(task_ids='score_task')['confidence']) }}

命名模式始终为 macros.{plugin_name}.{function_name}。

中间件

中间件会应用于所有 Airflow API 请求，包括内置 REST API 和任何 FastAPI 插件。请谨慎使用，并根据需要明确筛选请求：

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request, Response

class AuditMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        # runs before every request to the Airflow API server
        response = await call_next(request)
        return response

class MyPlugin(AirflowPlugin):
    name = "my_plugin"
    fastapi_root_middlewares = [
        {"middleware": AuditMiddleware, "args": [], "kwargs": {}, "name": "Audit"}
    ]

Operator 额外链接

from airflow.sdk.bases.operatorlink import BaseOperatorLink

class MyDashboardLink(BaseOperatorLink):
    name = "Open in Dashboard"

    def get_link(self, operator, *, ti_key, **context) -> str:
        return f"https://my-dashboard.example.com/tasks/{ti_key.task_id}"

class MyPlugin(AirflowPlugin):
    name = "my_plugin"
    global_operator_extra_links = [MyDashboardLink()]  # appears on every task
    # operator_extra_links = [MyDashboardLink()]       # attach to specific operator instead

React 应用

React 应用会以 JavaScript bundle 的形式嵌入，并通过 FastAPI 提供服务。bundle 必须将自身导出为与插件名称匹配的全局变量：

// In your bundle (e.g. my-app.js)
globalThis['My Plugin'] = MyComponent;   // matches plugin name
globalThis.AirflowPlugin = MyComponent;  // fallback Airflow looks for

class MyPlugin(AirflowPlugin):
    name = "my_plugin"
    fastapi_apps = [{"app": app, "url_prefix": "/my-plugin", "name": "My Plugin"}]
    react_apps = [
        {
            "name": "My Plugin",
            "bundle_url": "/my-plugin/my-app.js",
            "destination": "nav",
            "category": "browse",
            "url_route": "my-plugin",
        }
    ]

通过添加多个条目，同一个 bundle 可以注册到多个目标位置——每个条目都需要唯一的 url_route：

react_apps = [
    {"name": "My Widget", "bundle_url": "/my-plugin/widget.js", "destination": "nav",  "url_route": "my-widget-nav"},
    {"name": "My Widget", "bundle_url": "/my-plugin/widget.js", "destination": "dag",  "url_route": "my-widget-dag"},
]

React 应用集成在 Airflow 3.1 中仍处于实验阶段。后续版本中的接口可能会发生变化。

---

第 7 步：环境变量和部署

切勿将凭据硬编码：

AIRFLOW_HOST = os.environ.get("MYPLUGIN_HOST",     "http://localhost:8080")
AIRFLOW_USER = os.environ.get("MYPLUGIN_USERNAME", "admin")
AIRFLOW_PASS = os.environ.get("MYPLUGIN_PASSWORD", "admin")

本地 Astro CLI：

# .env
MYPLUGIN_HOST=http://localhost:8080
MYPLUGIN_USERNAME=admin
MYPLUGIN_PASSWORD=admin

astro dev restart              # required after any Python plugin change

# Check logs by component (Astro CLI):
astro dev logs --api-server    # FastAPI apps, external_views — plugin import errors show here
astro dev logs --scheduler     # macros, timetables, listeners, operator links
astro dev logs --dag-processor # DAG parsing errors

# Non-Astro:
airflow plugins                # CLI — lists all loaded plugins

Astronomer 生产环境：

astro deployment variable create --deployment-id <id> MYPLUGIN_HOST=https://airflow.example.com

开发期间自动重新加载（跳过延迟加载）：

AIRFLOW__CORE__LAZY_LOAD_PLUGINS=False

部署后为静态文件清除缓存：

<script src="static/app.js?v=20240315-1"></script>

验证插件是否已加载：在 Airflow UI 中打开管理 > 插件。

FastAPI 插件会自动生成 OpenAPI 文档：

• Swagger UI：{AIRFLOW_HOST}/{url_prefix}/docs
• OpenAPI JSON：{AIRFLOW_HOST}/{url_prefix}/openapi.json

---

常见问题

问题	原因	修复方法
导航链接转到 404	href 中带有前导 /	使用 "my-plugin/ui"，而不是 "/my-plugin/ui"
导航图标未显示	icon 中缺少 /	icon 接受绝对路径："/my-plugin/static/icon.svg"
负载较高时事件循环冻结	在 async def 中直接调用同步 SDK	使用 asyncio.to_thread() 封装
1 小时后出现 401 错误	JWT 过期且没有刷新机制	使用过期前 5 分钟刷新模式
StaticFiles 在启动时抛出异常	目录不存在	启动前创建 assets/ 和 static/
插件未显示	修改 Python 文件后未重启	astro dev restart
无需登录即可访问端点	FastAPI 应用不会自动进行身份验证	如果端点必须设为私有，请添加 FastAPI 安全机制（如 OAuth2、API 密钥）
中间件影响了错误的路由	中间件会应用于所有 API 流量	在 dispatch() 中根据 request.url.path 进行筛选
在 Astro 上使用 JS fetch() 时出错	fetch() 中使用了绝对路径	始终使用相对路径：fetch('api/dags')

---

参考资料

• Airflow 插件文档
• Airflow REST API 参考 — 完整的端点列表，包含 SDK 类名和方法名
• Astronomer：使用 Airflow 插件