nat.front_ends.fastapi.routes.websocket#

WebSocket route registration.

Attributes#

`logger`
`_SAFE_SESSION_ID_RE`

Functions#

`_is_origin_allowed`(→ bool)	Return True if origin should be treated as an allowed CORS origin.
`_get_jwt_validators`(→ dict[str, ...)
`websocket_endpoint`(*, worker, session_manager, ...)	Build websocket endpoint handler with auth-flow integration.
`add_websocket_routes`(worker, app, endpoint, ...)	Add websocket route for an endpoint.

Module Contents#

logger#

_SAFE_SESSION_ID_RE#

_is_origin_allowed( origin: str | None, allowed_origins: list[str], allow_origin_regex: str | None, ) → bool#

Return True if origin should be treated as an allowed CORS origin.

Mirrors the three-tier check used by Starlette’s CORSMiddleware: 1. Wildcard "*" in allowed_origins accepts any non-empty origin. 2. Exact membership in allowed_origins. 3. Full-string match against allow_origin_regex (when set).

async _get_jwt_validators( worker: Any, session_manager: nat.runtime.session.SessionManager, ) → dict[str, nat.authentication.credential_validator.bearer_token_validator.BearerTokenValidator]#

websocket_endpoint( *, worker: Any, session_manager: nat.runtime.session.SessionManager, jwt_validators: dict[str, nat.authentication.credential_validator.bearer_token_validator.BearerTokenValidator], )#: Build websocket endpoint handler with auth-flow integration.

async add_websocket_routes( worker: Any, app: fastapi.FastAPI, endpoint: Any, session_manager: nat.runtime.session.SessionManager, )#: Add websocket route for an endpoint.