ewox-service-fastapi

WebSockets

ServiceApp can expose a typed WebSocket route backed by a pluggable session manager. The default manager stores live sessions in the current process.

from ewoxcore.constants.server_env import ServerEnv
from ewoxservicefastapi.service_app import ServiceApp
from ewoxservicefastapi.websocket.websocket_settings import WebSocketSettings


class APIServer(ServiceApp):
    def __init__(self) -> None:
        super().__init__(service_name="api")

        settings = WebSocketSettings(
            environment=ServerEnv.LOCALHOST,
            public_host="localhost",
            public_port=8804,
            allowed_origins=["http://localhost:3000"],
        )
        self.addWebSocket(settings)

The public URL is ws://localhost:8804/ws for the localhost environment. All other environments return a client-facing wss:// URL. Cloudflare terminates TLS and proxies the WebSocket upgrade to Uvicorn over the private origin connection, so the service does not load or configure SSL certificate files. WebSocket transport settings are code-owned and are not read from .env.

An external package can replace the in-memory manager by implementing IWebSocketSessionManager and passing a factory:

self.addWebSocket(
    settings,
    session_manager_factory=lambda websocket_settings: ExternalSessionManager(
        websocket_settings
    ),
)

The package has no Redis dependency. A Redis-backed implementation can live in another package and use this factory boundary for distributed registry and message coordination while keeping live socket objects in each API process.

Container scaling

ServiceApp starts one Uvicorn process per container and does not configure Uvicorn's multi-worker mode. Production capacity is increased by running more containers behind Cloudflare and the platform load balancer.

The default in-memory manager is intended for local development or a single-container deployment. A multi-container deployment must provide an external IWebSocketSessionManager implementation. The external manager must keep each live WebSocket object inside its owning container and use its distributed transport to route direct messages and broadcasts to the correct container. A shared registry alone is not sufficient.

Transport models are serialized with JsonUtil. Application event models should use @Serializable and be registered once through ClassRegistry.

Durable delivery protocol

Distributed adapters can use WebSocketDeliveryModel[TData] to mark a server message as requiring client acknowledgement:

delivery = WebSocketDeliveryModel[WebSocketEventModel](
    delivery_id=delivery_id,
    data=event,
)
await session.send_model(delivery)

The serialized message has type="delivery", a stable deliveryId, and typed data. After processing it, the client returns a WebSocketAcknowledgementModel with type="ack" and the same deliveryId. The external session manager decides when that acknowledgement removes its durable entry. The framework package does not assume Redis or acknowledge a delivery merely because send_text(...) completed.