Framework API

framework

Application framework for WebTransport services.

ServerApp

Implement a high-level WebTransport application with routing and middleware.

server property

server: WebTransportServer

Return the underlying WebTransportServer instance.

add_middleware

add_middleware(*, middleware: MiddlewareProtocol) -> None

Add a middleware to the processing chain.

middleware

middleware(middleware_func: MiddlewareProtocol) -> MiddlewareProtocol

Register a middleware function.

on_shutdown

on_shutdown(handler: F) -> F

Register a handler to run on application shutdown.

on_startup

on_startup(handler: F) -> F

Register a handler to run on application startup.

pattern_route

pattern_route(*, pattern: str) -> Callable[[SessionHandler], SessionHandler]

Register a session handler for a URL pattern.

route

route(*, path: str) -> Callable[[SessionHandler], SessionHandler]

Register a session handler for a specific path.

run

run(*, host: str | None = None, port: int | None = None, **kwargs: Any) -> None

Execute the server application in a new asyncio event loop.

serve async

serve(*, host: str | None = None, port: int | None = None, **kwargs: Any) -> None

Start the server and listen for connections indefinitely.

shutdown async

shutdown() -> None

Run shutdown handlers and exit stateful middleware.

startup async

startup() -> None

Run startup handlers and enter stateful middleware.

AuthHandlerProtocol

Define the authentication handler interface.

MiddlewareManager

Manage a chain of server middleware.

add_middleware

add_middleware(*, middleware: MiddlewareProtocol) -> None

Add a middleware to the chain.

get_middleware_count

get_middleware_count() -> int

Return the number of registered middleware.

process_request async

process_request(*, session: SessionProtocol) -> None

Process a request through the middleware chain.

remove_middleware

remove_middleware(*, middleware: MiddlewareProtocol) -> None

Remove a middleware from the chain.

MiddlewareProtocol

Define the middleware interface.

MiddlewareRejected

Indicate a session request rejection by middleware.

StatefulMiddlewareProtocol

Define the stateful middleware interface.

RequestRouter

Route session requests to handlers based on path matching.

add_pattern_route

add_pattern_route(*, pattern: str, handler: SessionHandler) -> None

Register a route for a regular expression pattern.

add_route

add_route(*, path: str, handler: SessionHandler, override: bool = False) -> None

Register a route for an exact path match.

get_all_routes

get_all_routes() -> dict[str, SessionHandler]

Return a copy of all registered exact-match routes.

get_route_handler

get_route_handler(*, path: str) -> SessionHandler | None

Return the handler for a specific path (exact match only).

get_route_stats

get_route_stats() -> dict[str, Any]

Return statistics about the configured routes.

remove_pattern_route

remove_pattern_route(*, pattern: str) -> None

Unregister a route for a regular expression pattern.

remove_route

remove_route(*, path: str) -> None

Unregister a route for an exact path match.

route_request

route_request(*, session: WebTransportSession) -> tuple[SessionHandler, dict[str, Any]] | None

Dispatch a request to the appropriate handler based on the session path.

set_default_handler

set_default_handler(*, handler: SessionHandler) -> None

Configure a default handler for unmatched routes.

create_auth_middleware

create_auth_middleware(*, auth_handler: AuthHandlerProtocol) -> MiddlewareProtocol

Instantiate authentication middleware with a custom handler.

create_cors_middleware

create_cors_middleware(*, allowed_origins: list[str]) -> MiddlewareProtocol

Instantiate CORS middleware to validate the Origin header.

create_rate_limit_middleware

create_rate_limit_middleware(*, max_requests: int = _WINDOW_REQUEST_LIMIT, window_seconds: int = _WINDOW_TTL, cleanup_interval: int = _CLEANUP_INTERVAL, max_tracked_ips: int = _TRACKED_IP_CAPACITY) -> RateLimiter

Instantiate a stateful rate-limiting middleware.