ルーティングマネージャー

ErisPulse ルーティングマネージャーは、HTTP および WebSocket ルーティングを統一的に管理し、複数のアダプターによるルーティング登録とライフサイクル管理をサポートします。基盤は抽象化レイヤーを介してカプセル化されており（現在は FastAPI + Uvicorn が使用されています）。

概要

ルーティングマネージャーの主な機能：

デコレータによるルーティング：@http / @get / @post / @put / @delete / @ws デコレータによる簡易なルート登録をサポート
自動注入：ルートハンドラは FastAPI の型を明示的にインポートする必要がなく、フレームワークが抽象オブジェクトを自動的に注入
ルートグループ化：プレフィックスとバージョン番号を伴う RouteGroup をサポート
ルートミドルウェア：glob パターンマッチングによるリクエストのインターセプトをサポート
リクエスト制限：スライディングウィンドウ方式のリクエスト制限を内蔵
CORS 対応：1 つのコマンドでクロスオリジンリソース共有を有効化
セキュリティヘッダー：レスポンスヘッダーに自動的にセキュリティ関連のヘッダーを追加
自動ドキュメント生成：OpenAPI に基づくインタラクティブなドキュメントを提供
WebSocket 対応：WebSocket 接続の完全な管理、カスタム認証、ライフサイクルフックをサポート
ライフサイクル統合：ErisPulse のライフサイクルシステムと深く統合
SSL/TLS 対応：HTTPS および WSS のセキュア接続をサポート
ホームエントリ：モジュールがルート / に登録可能なクイックエントリボタンをサポート、多言語対応も可能

抽象型

ErisPulse は、モジュールが FastAPI に直接依存しないようにするためのサーバーサイドの抽象型を提供しています。

抽象型	FastAPI 対応	説明
`HttpRequest`	`fastapi.Request`	HTTP リクエストをラップした型で、完全に互換性があります
`WebSocketConnection`	`fastapi.WebSocket`	WebSocket 接続をラップした型で、ライフサイクルフックを追加で提供します
`WebSocketDisconnect`	`fastapi.WebSocketDisconnect`	WebSocket 接続切断時の例外型

WebSocketConnection は WebSocketConnectionBase を継承しており、クライアント側の WebSocket (ClientWebSocket) と同じ send/receive/iter/close インターフェースを共有しています。クライアントとサーバーの WebSocket は、同じビジネスロジックコードを使用できます。

.raw 属性を使用することで、下層の FastAPI のネイティブオブジェクトにアクセスできます。FastAPI の型を使用したコードも完全に互換性があります。

装饰器ルーティング（推奨）

HTTP 装飾器

from ErisPulse.Core import router
@router.get("my_module", "/info")
async def get_info(request):
    return {"method": request.method, "path": str(request.url)}

# 抽象型を明示的に指定することも可能
from ErisPulse.Core import HttpRequest

@router.post("my_module", "/data")
async def post_data(request: HttpRequest):
    data = await request.json()
    return {"received": data}

@router.put("my_module", "/data/{item_id}")
async def update_data(request):
    return {"updated": True}

@router.delete("my_module", "/data/{item_id}")
async def delete_data(request):
    return {"deleted": True}

自動注入ルール：ハンドラの最初の引数の名前が request または req であり、FastAPI の型注釈がない場合、フレームワークは自動的に HttpRequest を注入します。引数が存在しない、またはリクエスト引数名でないハンドラには影響しません。

WebSocket 装飾器

from ErisPulse.Core import WebSocketConnection, WebSocketDisconnect

# 基本的な WebSocket
@router.ws("my_module", "/ws")
async def websocket_handler(ws):
    async for msg in ws.iter_text():
        await ws.send_text(f"Echo: {msg}")

# ライフサイクルフック付きの WebSocket
@router.ws("my_module", "/ws/chat")
async def chat(ws: WebSocketConnection):
    @ws.on_disconnect
    async def on_disconnect(ws, reason="unknown"):
        print(f"ユーザーが切断: {reason}")

    @ws.on_error
    async def on_error(ws, error=""):
        print(f"接続エラー: {error}")

    async for msg in ws.iter_text():
        await ws.send_text(f"Echo: {msg}")

# 認証付きの WebSocket
async def ws_auth(ws: WebSocketConnection) -> bool:
    token = ws.query_params.get("token")
    return token == "secret"

@router.ws("my_module", "/secure_ws", auth_handler=ws_auth)
async def secure_ws_handler(ws):
    while True:
        data = await ws.receive_text()
        await ws.send_text(f"Echo: {data}")

注意：WebSocket ハンドラと認証ハンドラも自動注入をサポートしています。WebSocketConnection を取得するために引数の型注釈は不要です。fastapi.WebSocket を型注釈に指定することで、元のオブジェクトを渡すこともできますが、抽象型を使用することを推奨します。

伝統的な登録方法

async def hello_handler(request):
    return {"message": "Hello World"}

# 基本的な登録
router.register_http_route(
    module_name="my_module",
    path="/hello",
    handler=hello_handler,
    methods=["GET"],
)

# 限界値制限とドキュメント情報付き
router.register_http_route(
    module_name="my_module",
    path="/api/data",
    handler=data_handler,
    methods=["POST"],
    rate_limit="10/minute",
    summary="データインターフェース",
    tags=["API"],
)

WebSocket 登録

from ErisPulse.Core import WebSocketConnection

async def websocket_handler(ws: WebSocketConnection):
    async for msg in ws.iter_text():
        await ws.send_text(f"Echo: {msg}")

# 基本的な登録
router.register_websocket(
    module_name="my_module",
    path="/ws",
    handler=websocket_handler,
)

# 認証付きの登録（推奨）
async def auth_handler(ws: WebSocketConnection) -> bool:
    token = ws.query_params.get("token")
    return token == "secret"

router.register_websocket(
    module_name="my_module",
    path="/secure_ws",
    handler=websocket_handler,
    auth_handler=auth_handler,
)

パラメータの説明：

パラメータ	説明	デフォルト値
`module_name`	モジュール名（必須）	-
`path`	WebSocket パス	-
`handler`	処理関数	-
`auth_handler`	認証関数。`False` を返すと接続が自動的に切断されます	`None`
`auto_accept`	自動的に `accept()` を呼び出すかどうか	`True`

推奨：接続の確認には auth_handler を使用してください。auto_accept を False に設定するのは、接続の流れを完全に制御する必要がある場合に限ってください。

WebSocket ライフサイクルフック

WebSocketConnection は、手動での try/catch なしに、切断とエラーのコールバックを登録することができます。

from ErisPulse.Core import WebSocketConnection

@router.ws("my_module", "/ws")
async def my_ws(ws: WebSocketConnection):
    # デコレータ方式で登録
    @ws.on_disconnect
    async def on_close(ws, reason="unknown"):
        print(f"切断原因: {reason}")

    # 直接呼び出すこともできます
    async def on_err(ws, error=""):
        print(f"エラー: {error}")
    ws.on_error(on_err)

    # 通常の業務ロジック
    async for msg in ws.iter_text():
        await ws.send_text(f"Echo: {msg}")

ルートのグループ化

# プレフィックスを付けてルートグループを作成
group = router.group("my_module", prefix="/v1")

@group.get("/users")
async def list_users(request):
    return {"users": []}

@group.post("/users")
async def create_user(request):
    return {"created": True}

# 実際のパス: /my_module/v1/users

ルーティングミドルウェア

ミドルウェアは、パスに対して glob パターンによるマッチングをサポートしています：

@router.middleware("/my_module/*")
async def auth_middleware(request, call_next):
    token = request.headers.get("Authorization")
    if not token:
        return {"error": "Unauthorized"}
    return await call_next(request)

@router.middleware("/my_module/admin/*")
async def admin_middleware(request, call_next):
    return await call_next(request)

リクエスト関連 ID（X-Request-ID）

2.7.0 以降、すべての HTTP リクエストには、ログ / リクエストの連携を可能にする X-Request-ID 関連 ID が含まれます。

生成ルール：クライアントが X-Request-ID リクエストヘッダーを送信している場合、それを優先して使用します（分散トレーシングの場面）。それ以外の場合は UUID を自動生成します。
レスポンスヘッダー：レスポンスには X-Request-ID が返信され、クライアントがリクエストとログを対応付けることができます。
ライフサイクルイベント：server.request および server.response イベントのデータに request_id フィールドが追加されました。

# モジュール内でリクエストイベントを監視し、request_id でリクエストとレスポンスを連携します
@sdk.lifecycle.on("server.request")
async def on_request(data):
    print(f"[{data['request_id']}] {data['method']} {data['path']}")

@sdk.lifecycle.on("server.response")
async def on_response(data):
    print(f"[{data['request_id']}] -> {data['status_code']}")

クライアントは、サービス間のトレースを可能にするために独自の ID を設定できます。

curl -H "X-Request-ID: my-trace-id" http://localhost:8080/my_module/health

速率制限

ルーティングに対してスライディングウィンドウアルゴリズムを使用したリクエスト制限を実装します。

@router.get("my_module", "/limited", rate_limit="10/minute")
async def limited_endpoint(request):
    return {"ok": True}

@router.post("my_module", "/submit", rate_limit="5/minute")
async def submit_data(request):
    return {"submitted": True}

リクエスト制限の形式：{回数}/{時間単位}、例：10/minute、100/hour。

CORS 設定

router.setup_cors(
    allow_origins=["https://example.com"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

config.toml でも設定可能です：

[router.cors]
allow_origins = ["https://example.com"]
allow_methods = ["GET", "POST"]
allow_headers = ["*"]

セキュリティヘッダー

router.setup_security_headers()

自動的に X-Content-Type-Options、X-Frame-Options、X-XSS-Protection などのセキュリティヘッダーを追加します。

また、config.toml で設定することもできます：

[router.security]
enabled = true

自動ドキュメント

Router はデフォルトで OpenAPI のインタラクティブなドキュメントを有効にしています：

# ドキュメントの無効化
router.disable_docs()

# ドキュメント情報のカスタマイズ
router.set_docs_info(
    title="My API",
    description="API ドキュメント",
    version="1.0.0"
)

パス処理

ルートパスには、モジュール名が自動的にプレフィックスとして追加され、競合を回避します：

# モジュール "my_module" にパス "/api" を登録
# 実際のアクセスパスは "/my_module/api" になります
router.register_http_route("my_module", "/api", handler)

システムルーティング

ルーティングマネージャーは、以下のシステムルーティングを自動的に提供します。

ヘルスチェック

GET /health
# 戻り値:
{"status": "ok", "service": "ErisPulse Router"}

ルートページ

GET /
# ErisPulse ブランドページを返す

ルートルーティング / は、ErisPulse ブランドページを表示し、ダッシュボードの利用可能性を自動的に検出し、エントリーボタンを追加します。

ホームページのエントリ

ルーティングマネージャーは、外部モジュールがルートルート / にクイックエントリボタンを登録することを許可し、ユーザーが各モジュールの管理ページに迅速にアクセスできるようにします。

エントリの登録

# 簡単な登録
router.register_home_entry(
    name="マイダッシュボード",
    url="/mymodule/admin",
)

# イコン付きの登録（SVG）
router.register_home_entry(
    name="コンソール",
    url="/console",
    icon_svg='<svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8"><path d="M4 17l6-6-6-6"/><path d="M12 19h8"/></svg>',
)

# 国際化をサポートする登録（i18n ディクショナリ形式）
router.register_home_entry(
    name={"i18n": "mymodule.home.entry", "default": "マイダッシュボード"},
    url="/mymodule/admin",
)