airflow-plugins

Airflow 3 プラグイン

Airflow 3 プラグインを使うと、FastAPI アプリ、React UI、ミドルウェア、マクロ、オペレーターボタン、カスタムタイムテーブルを Airflow プロセスに直接組み込めます。サイドカーも追加サーバーも不要です。

重大：プラグインコンポーネント（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. appbuilder_views / Flask ではなく、fastapi_apps / FastAPI を使用していますか？
2. すべての HTML/JS アセットパスと fetch() 呼び出しは相対パス（先頭にスラッシュなし）ですか？
3. 同期 SDK または SQLAlchemy の呼び出しはすべて asyncio.to_thread() でラップされていますか？
4. FastAPI アプリをマウントする前に、static/ と assets/ ディレクトリが存在していますか？
5. エンドポイントを非公開にする必要がある場合、FastAPI の認証を明示的に追加しましたか？

---

ステップ 1：プラグインコンポーネントを選択する

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 の登録、または DB への直接アクセスのみを行うプラグインでは、この手順は不要です。ステップ 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 を追加してください」と伝えます

Airflow 独自の REST API との通信には apache-airflow-client を使用してください。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。

常に、SDK のロジックをすべて含む通常の内部 def _fetch() を定義し、その後 await asyncio.to_thread(_fetch) を呼び出すパターンを使用してください。

代替案：データベースへの直接アクセス

警告 — 注意して使用し、ユーザーにも伝えてください。 Airflow のメタデータデータベース（metadb）は公開インターフェースではありません。直接書き込んだり、不適切なクエリを実行したりすると、スケジューラーの状態が破損するおそれがあります。このパターンを使用する場合は必ず、ユーザーに次のように明示してください。「これは Airflow の内部データベースに直接アクセスします。内部モデルは公開 API の一部ではなく、Airflow のバージョン間で変更される可能性があります。また、不適切なクエリにより metadb で問題が発生することがあります。REST API で公開されていない操作を行う場合を除き、apache-airflow-client を使用してください。」

FastAPI プラグインのエンドポイントはタスクワーカーではなく、API サーバープロセス内で実行されるため、HTTP の往復や JWT なしで Airflow の内部 SQLAlchemy モデルに直接アクセスできます。REST API で公開されていない読み取り操作に限って使用するか、HTTP による追加のオーバーヘッドが実際に問題となる場合に使用してください。SQLAlchemy のクエリはブロッキングするため、DB 呼び出しは必ず asyncio.to_thread() でラップしてください。

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 以外で最もよく使われる呼び出しです。ここに記載されていないものについては、利用可能なエンドポイントと対応する Python SDK のクラス名／メソッド名をREST API リファレンスで確認してください。

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)}

ストリーミングプロキシ

外部 URL のバイナリコンテンツをプラグイン経由でプロキシするには、StreamingResponse を使用します。ブラウザーからリソースを直接取得できない場合（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：その他のプラグインコンポーネントの種類

マクロ

マクロは API サーバーではなく、スケジューラー（および DAG プロセッサー）によって読み込まれます。変更後はスケジューラーを再起動してください。

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} です。

ミドルウェア

ミドルウェアは、組み込みの REST API や FastAPI プラグインを含む、すべての Airflow API リクエストに適用されます。必要に応じてリクエストを明示的にフィルタリングし、使用は最小限にとどめてください。

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"}
    ]

オペレーターの追加リンク

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 アプリは、FastAPI 経由で配信される JavaScript バンドルとして埋め込まれます。バンドルは、プラグイン名に一致するグローバル変数として自身を公開する必要があります。

// 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",
        }
    ]

複数のエントリーを追加することで、同じバンドルを複数の宛先に登録できます。それぞれに一意の 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 の Admin > Plugins を開きます。

OpenAPI ドキュメントは FastAPI プラグイン用に自動生成されます。

• 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 で JavaScript の fetch() が動作しない	fetch() に絶対パスを指定している	必ず相対パスを使用します：fetch('api/dags')

---

参考資料

• Airflow プラグインのドキュメント
• Airflow REST API リファレンス — SDK のクラス名とメソッド名を含む、すべてのエンドポイントの一覧
• Astronomer：Airflow プラグインの使用