trading_agent.gateway.routes.event_routes

The WebSocket that streams every event to a client as it happens.

A client can ask for one desk's events only with ?desk=<name>; events that belong to no desk, such as account-wide ones, are always sent.

The handler waits for two things at once: the next event, and the client hanging up. Without the second, a closed connection would only be noticed at the next send, and on a quiet bus its subscription would linger.

Typical usage example:

application.include_router(EventRoutes(event_bus, guard).router())

EventRoutes

EventRoutes(event_bus, guard, redactor=None)

Builds the /ws/events WebSocket route.

Creates the route for one event bus.

Parameters:

Name	Type	Description	Default
event_bus	EventBus	The bus whose events are streamed.	required
guard	AccessGuard	The guard that checks the access token.	required
redactor	SecretRedactor | None	Masks secrets before an event is sent, or None to send it unchanged.	None

Source code in src/trading_agent/gateway/routes/event_routes.py

def __init__(
    self,
    event_bus: event_bus_module.EventBus,
    guard: access_guard.AccessGuard,
    redactor: secret_redactor.SecretRedactor | None = None,
):
    """Creates the route for one event bus.

    Args:
        event_bus (event_bus_module.EventBus): The bus whose events are streamed.
        guard (access_guard.AccessGuard): The guard that checks the access token.
        redactor (secret_redactor.SecretRedactor | None): Masks secrets before an event is sent, or None to send it unchanged.
    """
    self._event_bus = event_bus
    self._guard = guard
    self._redactor = redactor

router

router()

Builds the router holding the WebSocket route.

Returns:

Type	Description
APIRouter	The router.

Source code in src/trading_agent/gateway/routes/event_routes.py

def router(self) -> fastapi.APIRouter:
    """Builds the router holding the WebSocket route.

    Returns:
        fastapi.APIRouter: The router.
    """
    router = fastapi.APIRouter()
    router.add_api_websocket_route('/ws/events', self.stream_events)
    return router

stream_events async

stream_events(websocket)

Sends each event to the client as JSON until it disconnects.

Parameters:

Name	Type	Description	Default
websocket	WebSocket	The client's connection, with the token in its token query parameter and an optional desk in desk.	required

Source code in src/trading_agent/gateway/routes/event_routes.py

async def stream_events(self, websocket: fastapi.WebSocket) -> None:
    """Sends each event to the client as JSON until it disconnects.

    Args:
        websocket (fastapi.WebSocket): The client's connection, with the token in its `token` query parameter and an optional desk in `desk`.
    """
    if not self._guard.is_valid(websocket.query_params.get('token')):
        await websocket.close(code=_POLICY_VIOLATION)
        return
    await websocket.accept()
    disconnect_task = asyncio.create_task(
        self._wait_for_disconnect(websocket)
    )
    with self._event_bus.subscribe() as subscription:
        try:
            await self._forward_events(
                websocket,
                subscription,
                disconnect_task,
                websocket.query_params.get('desk'),
            )
        except fastapi.WebSocketDisconnect:
            return
        finally:
            disconnect_task.cancel()