路由管理器

ErisPulse 路由管理器提供統一的 HTTP 和 WebSocket 路由管理，支援多適配器路由註冊和生命週期管理。底層透過抽象層封裝（目前為 FastAPI + Uvicorn）

概述

路由管理器的主要功能：

• 裝飾器路由：支援 @http / @get / @post / @put / @delete / @ws 裝飾器快速註冊
• 自動注入：路由處理器無需導入 FastAPI 類型，框架自動注入抽象物件
• 路由分組：支援帶前綴和版本號的 RouteGroup
• 路由中間件：支援 glob 模式匹配的請求攔截
• 速率限制：內建滑動視窗限流
• CORS 支援：一鍵開啟跨域資源共享
• 安全頭：自動添加安全回應頭
• 自動文件：基於 OpenAPI 的互動式文件
• WebSocket 支援：完整的 WebSocket 連線管理、自訂認證和生命週期鉤子
• 生命週期整合：與 ErisPulse 生命週期系統深度整合
• SSL/TLS 支援：支援 HTTPS 和 WSS 安全連線
• 主頁入口：支援模組在根路由 / 註冊快速入口按鈕，支援國際化

抽象類型

ErisPulse 提供了服務端抽象類型，使模組無需直接依賴 FastAPI：

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

參數說明：

建議：使用 auth_handler 進行連接確認，而非關閉 auto_accept。僅在你需要完全控制連接流程時才設定 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"
)

路徑處理

路由路徑會自動加上模組名稱作為前綴，以避免衝突：

# 將路徑 "/api" 註冊到模組 "my_module"
# 實際可存取的路徑為 "/my_module/api"
router.register_http_route("my_module", "/api", handler)

系統路由

路由管理器自動提供以下系統路由：

健康檢查

GET /health
# 回傳:
{"status": "ok", "service": "ErisPulse Router"}

根頁面

GET /
# 回傳 ErisPulse 品牌頁

根路由 / 顯示 ErisPulse 品牌頁面，自動偵測 Dashboard 的可用性並添加入口按鈕。

主頁入口

路由管理器允許外部模組在根路由 / 上註冊快捷入口按鈕，方便使用者快速訪問各模組的管理頁面。

註冊入口

# 簡單註冊
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",
)

參數說明：

Dashboard 自動註冊

當偵測到 sdk.Dashboard 可用時，路由管理器會自動在入口列表首位添加 Dashboard 按鈕，無需手動註冊。

生命週期集成

from ErisPulse.Core import lifecycle

@lifecycle.on("server.start")
async def on_server_start(event):
    print(f"伺服器已啟動: {event['data']['base_url']}")

@lifecycle.on("server.stop")
async def on_server_stop(event):
    print("伺服器正在停止...")

最佳實踐

1. 優先使用抽象類型：使用 HttpRequest / WebSocketConnection 替代 fastapi.Request / fastapi.WebSocket，避免硬性依賴
2. 利用自動注入：處理器第一個參數命名為 request 或 req，無需任何類型註解即可獲得 HttpRequest
3. 明確傳入 module_name：裝飾器第一個參數必須為模組名，不可省略
4. 使用路由分組：對同一模組的多個路由使用 group() 進行組織
5. 安全性考量：為敏感操作實現認證機制和安全頭
6. 合理限流：對高頻接口設定速率限制
7. 使用生命週期鉤子：透過 @ws.on_disconnect / @ws.on_error 處理 WebSocket 異常，避免手動 try/catch

相關文件

• HTTP 客戶端 - 使用內建 HTTP 客戶端發送請求
• 模組開發指南 - 了解模組路由註冊
• 最佳實踐 - 路由使用建議