WebSockets and Application Lifecycle

WebSocket routes

WebSocket handlers use the same path converters as HTTP routes:

from flasgo import Flasgo, WebSocket

app = Flasgo()


@app.websocket("/rooms/<int:room_id>")
async def room(websocket: WebSocket, room_id: int) -> None:
    await websocket.accept()
    async for message in websocket.iter_json():
        await websocket.send_json(
            {"room_id": room_id, "echo": message}
        )

The API includes:

• accept(subprotocol=..., headers=...)
• receive_text(), receive_bytes(), and receive_json()
• send_text(), send_bytes(), and send_json()
• iter_text(), iter_bytes(), and iter_json()
• close(code=1000, reason="") and deny(...)

WebSocketDisconnect identifies peer disconnects. Receiving after the client has left raises WebSocketDisconnect with the client’s close code; sending or receiving after the application closed the socket raises RuntimeError. Flasgo validates message types and JSON, bounds message size and rate, checks close codes, and rejects invalid or oversized close reasons. websocket.query_params enforces MAX_FORM_FIELDS like HTTP requests.

Flasgo inspects the handler once at registration. It passes the connection to a parameter named websocket or to any parameter annotated WebSocket, including Annotated[WebSocket, ...]; annotations imported only under TYPE_CHECKING do not fail connections. A handler that returns without calling accept() is denied with 403 and logged with the outcome not_accepted.

Specific routes take precedence over catch-all routes. Registration rejects equivalent path shapes even when parameter names or converters differ, intersecting patterns with equal specificity, and routes containing multiple path converters. See Route Precedence and Conflicts.

Handshake security

Before acceptance, Flasgo:

• enforces MAX_REQUEST_HEAD_BYTES before tracing, routing, or session loading;
• applies the HTTP host allowlist;
• requires an exact same-origin or allowlisted Origin;
• rejects a missing Origin by default;
• runs default client-IP route limits before session loading and authentication;
• loads the session using the configured backend (signed cookies by default), without persisting handler changes;
• runs route authorization and then custom identity-based rate limits;
• creates a local request ID.

Oversized handshakes receive HTTP 431 when the ASGI server supports the WebSocket HTTP-response extension; otherwise Flasgo closes with code 1008. Head-limit denials bypass OpenTelemetry and never reuse an incoming request ID. Rejected Host, Origin, and default client-IP quotas do not invoke session storage.

For a separate browser frontend, configure exact origins:

app = Flasgo(
    settings={
        "ALLOWED_HOSTS": {"api.example.com"},
        "WEBSOCKET_ALLOWED_ORIGINS": {"https://app.example.com"},
    }
)

Set WEBSOCKET_ALLOW_MISSING_ORIGIN=True only for a deliberate non-browser client model that does not rely on ambient cookies. An accepted subprotocol must be one the client offered:

@app.websocket("/events")
async def events(websocket: WebSocket):
    await websocket.accept(subprotocol="events.v1")
    await websocket.send_json({"ready": True})

Default limits are 65,536 bytes per message and 120 messages per minute per connection. Change them with WEBSOCKET_MAX_MESSAGE_BYTES and WEBSOCKET_MAX_MESSAGES_PER_MINUTE.

Authorization

Authorization decorators work before the WebSocket is accepted:

from flasgo import IsAuthenticated


@app.websocket("/account-events")
@app.authorize(IsAuthenticated(), backend="bearer")
async def account_events(websocket: WebSocket):
    await websocket.accept()
    user_id = websocket.user.id if websocket.user else None
    await websocket.send_json({"user_id": user_id})

Register the backend before serving. See Security and authentication.

WebSocket routes also accept public=True for policy inspection; blueprints group HTTP routes only. Server-side sessions are loaded at the handshake but changes are never saved by the WebSocket handler. Revocation affects future loads; an accepted connection does not continuously revalidate its acquired identity.

Lifespan and process state

Register one async generator that yields exactly once:

from collections.abc import AsyncGenerator

from flasgo import Flasgo

app = Flasgo()


@app.lifespan
async def lifespan(app: Flasgo) -> AsyncGenerator[None, None]:
    app.state.http_client = await create_http_client()
    try:
        yield
    finally:
        await app.state.http_client.aclose()

Code before yield runs at startup; cleanup after yield runs at shutdown. A startup failure is reported to the ASGI server and the application does not enter a partially ready state, so a later startup can run again. A handler that yields more than once fails shutdown with RuntimeError, and Flasgo closes the generator immediately so its finally cleanup still runs.

app.state is process-global. Store connection pools or immutable service objects there, never request data. Each worker has its own state.

Background tasks

Attach best-effort work to a response:

from flasgo import Response


def write_audit_event(action: str) -> None:
    ...


@app.post("/jobs")
async def create_job() -> Response:
    response = Response.json({"accepted": True}, status_code=202)
    response.add_task(write_audit_event, "job-created")
    return response

You can also construct BackgroundTasks and pass it as background= to a response. Tasks:

• run sequentially only after the complete response was sent successfully, including the final chunk of a stream;
• run after request dependency cleanup, with request/user/session context cleared;
• await async functions and run synchronous functions in a worker thread;
• inherit the request ID in logs;
• continue after a preceding task fails.

Tasks must acquire their own resources; yielded request dependencies have already been cleaned up. They are in-process and best effort. Use a durable queue for payments, emails, data pipelines, or anything that must survive a worker restart.

BackgroundTasks.bind_observer() is deprecated in 0.10.0 and emits a DeprecationWarning; Flasgo no longer uses it. Use the background task metrics instead. An observer that raises no longer marks a successful task as failed.