Router Manager

The ErisPulse Router Manager provides unified HTTP and WebSocket routing management, supporting multi-adapter route registration and lifecycle management. The underlying layer is encapsulated through an abstraction layer (currently FastAPI + Uvicorn).

Overview

The main features of the routing manager:

• Decorator Routes: Support for quick registration with @http / @get / @post / @put / @delete / @ws decorators
• Automatic Injection: Route handlers do not require importing FastAPI types; the framework automatically injects abstract objects
• Route Grouping: Support for RouteGroup with prefix and version number
• Route Middleware: Support for request interception with glob pattern matching
• Rate Limiting: Built-in sliding window rate limiting
• CORS Support: One-click enablement of cross-origin resource sharing
• Security Headers: Automatic addition of security response headers
• Automatic Documentation: Interactive documentation based on OpenAPI
• WebSocket Support: Complete WebSocket connection management, custom authentication, and lifecycle hooks
• Lifecycle Integration: Deep integration with the ErisPulse lifecycle system
• SSL/TLS Support: Support for HTTPS and WSS secure connections
• Home Entry Point: Support for registering quick entry buttons for modules at the root route /, with internationalization support

Abstract Types

ErisPulse provides server-side abstract types, allowing modules to avoid direct dependencies on FastAPI:

WebSocketConnection inherits from WebSocketConnectionBase and shares the same send/receive/iter/close interface with the client-side WebSocket (ClientWebSocket). Business logic code can be reused between client and server WebSocket connections.

The underlying native FastAPI object can be accessed via the .raw attribute. Code that directly uses FastAPI types is fully compatible as well.

Decorator-based Routing (Recommended)

HTTP Decorators

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

# You can also explicitly annotate abstract types
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}

Automatic Injection Rule: When the first parameter of a handler is named request or req and has no FastAPI type annotation, the framework automatically injects HttpRequest. Handlers with no parameters or non-request parameter names are unaffected.

WebSocket Decorators

from ErisPulse.Core import WebSocketConnection, WebSocketDisconnect

# Basic 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 with lifecycle hooks
@router.ws("my_module", "/ws/chat")
async def chat(ws: WebSocketConnection):
    @ws.on_disconnect
    async def on_disconnect(ws, reason="unknown"):
        print(f"User disconnected: {reason}")

    @ws.on_error
    async def on_error(ws, error=""):
        print(f"Connection error: {error}")

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

# WebSocket with authentication
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}")

Note: WebSocket handlers and authentication handlers also support automatic injection. You can obtain WebSocketConnection without parameter annotations. You can also pass in the native object by annotating with fastapi.WebSocket, but abstract types are recommended.

Traditional Registration Methods

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

# Basic registration
router.register_http_route(
    module_name="my_module",
    path="/hello",
    handler=hello_handler,
    methods=["GET"],
)

# With rate limiting and documentation
router.register_http_route(
    module_name="my_module",
    path="/api/data",
    handler=data_handler,
    methods=["POST"],
    rate_limit="10/minute",
    summary="Data API",
    tags=["API"],
)

WebSocket Registration

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

# Basic registration
router.register_websocket(
    module_name="my_module",
    path="/ws",
    handler=websocket_handler,
)

# With authentication (recommended)
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,
)

Parameter Description:

Recommendation: Use auth_handler for connection confirmation instead of setting auto_accept=False. Only set auto_accept=False if you need full control over the connection process.

WebSocket Lifecycle Hooks

WebSocketConnection provides callback registration for disconnection and errors, eliminating the need for manual try/catch:

from ErisPulse.Core import WebSocketConnection

@router.ws("my_module", "/ws")
async def my_ws(ws: WebSocketConnection):
    # Register using decorator
    @ws.on_disconnect
    async def on_close(ws, reason="unknown"):
        print(f"Disconnect reason: {reason}")

    # Alternatively, register directly
    async def on_err(ws, error=""):
        print(f"Error: {error}")
    ws.on_error(on_err)

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

Route Groups

# Create a route group with a prefix
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}

# Actual path: /my_module/v1/users

Route Middleware

Middleware supports glob pattern matching for paths:

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

Request Correlation ID (X-Request-ID)

Starting from version 2.7.0, each HTTP request carries a X-Request-ID correlation ID, which is used for log and distributed tracing correlation:

• Generation Rule: The client-provided X-Request-ID header is prioritized (in distributed tracing scenarios); otherwise, a UUID is generated automatically.
• Response Header: The response will include a X-Request-ID, which helps the client match requests with logs.
• Lifecycle Events: The server.request and server.response event data now include a request_id field.

# Listen for request events in the module and correlate requests and responses by 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']}")

Clients can customize the ID to facilitate cross-service tracing:

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

Rate Limiting

Rate limiting for routes using the sliding window algorithm:

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

Rate limiting format: {count}/{time window}, for example 10/minute, 100/hour.

CORS Configuration

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

CORS can also be configured via config.toml:

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

Security Headers

router.setup_security_headers()

Automatically adds security headers such as X-Content-Type-Options, X-Frame-Options, and X-XSS-Protection.

It can also be configured via config.toml:

[router.security]
enabled = true

Automatic Documentation

Router enables OpenAPI interactive documentation by default:

# Disable documentation
router.disable_docs()

# Customize documentation information
router.set_docs_info(
    title="My API",
    description="API documentation",
    version="1.0.0"
)

Path Handling

Route paths are automatically prefixed with the module name to avoid conflicts:

# Register the path "/api" to the module "my_module"
# The actual accessible path is "/my_module/api"
router.register_http_route("my_module", "/api", handler)

System Routes

The routing manager automatically provides the following system routes:

Health Check

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

Root Page

GET /
# Returns ErisPulse brand page

The root route / displays the ErisPulse brand page, automatically detects the availability of the Dashboard and adds an entry button.

Home Entry

The router manager allows external modules to register quick-access entry buttons on the root route /, making it convenient for users to quickly access the management pages of various modules.

Registering an Entry

# Simple registration
router.register_home_entry(
    name="My Dashboard",
    url="/mymodule/admin",
)

# Registration with an icon (SVG)
router.register_home_entry(
    name="Dashboard",
    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>',
)

# Registration with internationalization support (project i18n dictionary format)
router.register_home_entry(
    name={"i18n": "mymodule.home.entry", "default": "My Dashboard"},
    url="/mymodule/admin",
)

Parameter Description:

Automatic Dashboard Registration

When sdk.Dashboard is detected as available, the router manager automatically adds a Dashboard button as the first entry in the list, eliminating the need for manual registration.

Lifecycle Integration

from ErisPulse.Core import lifecycle

@lifecycle.on("server.start")
async def on_server_start(event):
    print(f"Server has started: {event['data']['base_url']}")

@lifecycle.on("server.stop")
async def on_server_stop(event):
    print("Server is stopping...")

Best Practices

1. Prefer abstract types: Use HttpRequest / WebSocketConnection instead of fastapi.Request / fastapi.WebSocket to avoid hard dependencies.
2. Leverage automatic injection: Name the first parameter of a handler request or req to automatically receive an HttpRequest without any type annotation.
3. Explicitly pass module_name: The first parameter of a decorator must be the module name and cannot be omitted.
4. Use route grouping: Organize multiple routes from the same module using group().
5. Security considerations: Implement authentication mechanisms and security headers for sensitive operations.
6. Apply rate limiting appropriately: Set rate limits for high-frequency endpoints.
7. Use lifecycle hooks: Handle WebSocket exceptions using @ws.on_disconnect / @ws.on_error to avoid manual try/catch blocks.

Related Documentation

• HTTP Client - Use the built-in HTTP client to send requests
• Module Development Guide - Learn about module route registration
• Best Practices - Routing usage recommendations