diff --git a/backends/mizan-fastapi/src/mizan_fastapi/__init__.py b/backends/mizan-fastapi/src/mizan_fastapi/__init__.py index 13c2bc4..ef98217 100644 --- a/backends/mizan-fastapi/src/mizan_fastapi/__init__.py +++ b/backends/mizan-fastapi/src/mizan_fastapi/__init__.py @@ -1,10 +1,9 @@ """ mizan-fastapi — FastAPI backend adapter for the Mizan protocol. -HTTP RPC dispatch and context bundling on top of mizan-core's function -registry. Channels, Forms, Shapes, SSR are out of scope — FastAPI -projects use native equivalents (WebSocket, Pydantic, ORM-of-choice, -SSR frameworks). +HTTP RPC dispatch, context bundling, and a WebSocket carrying channel +subscriptions and RPC on mizan-django's envelope. Forms, Shapes and SSR are +out of scope — FastAPI projects use native equivalents. Usage: from fastapi import FastAPI @@ -12,6 +11,7 @@ Usage: app = FastAPI() app.include_router(router, prefix="/api/mizan") + app.include_router(ws_router, prefix="/api/mizan") app.add_exception_handler(MizanError, mizan_exception_handler) # Register your @client-decorated functions @@ -35,9 +35,16 @@ from .executor import ( execute_function, ) from .router import router, mizan_exception_handler, mizan_validation_handler +from .websocket import ws_router +from .channels import ReactChannel, broadcast, get_channel, register as register_channel __all__ = [ "router", + "ws_router", + "ReactChannel", + "register_channel", + "get_channel", + "broadcast", "mizan_exception_handler", "mizan_validation_handler", "execute_function", diff --git a/backends/mizan-fastapi/src/mizan_fastapi/channels.py b/backends/mizan-fastapi/src/mizan_fastapi/channels.py new file mode 100644 index 0000000..797b5dd --- /dev/null +++ b/backends/mizan-fastapi/src/mizan_fastapi/channels.py @@ -0,0 +1,124 @@ +""" +Channels for FastAPI — multiplexed pub/sub over the one WebSocket connection. + +A channel names a group of subscribers and decides who may join it. `group(params)` +is the fan-out key, so two subscribers with the same params share a group and a push +addressed to those params reaches both. + +Membership is held in this process. Django's channel layer carries groups across +workers through Redis; nothing here does, so a push reaches only the subscribers +connected to the process that sent it. One process — a desktop shell, a single +uvicorn worker — is the shape this serves. +""" + +from __future__ import annotations + +import asyncio +from collections import defaultdict +from typing import Any, ClassVar + +from fastapi.encoders import jsonable_encoder +from pydantic import BaseModel + +# group name -> the live sockets subscribed to it +_groups: dict[str, set[Any]] = defaultdict(set) +_registry: dict[str, type["ReactChannel"]] = {} +_lock = asyncio.Lock() + + +class ReactChannel: + """A named fan-out. Subclass, override what the channel decides, and register it. + + A subclass may define `on_connect(params)` / `on_disconnect()`; the socket handler + calls them when they exist. + """ + + name: ClassVar[str] = "" + + def authorize(self, params: BaseModel | None = None) -> bool: + """Whether this subscriber may join. Default: anyone may.""" + return True + + def group(self, params: BaseModel | None = None) -> str: + """The fan-out key. Subscribers sharing it share every message sent to it.""" + if params is None: + return self.name + parts = sorted(f"{k}={v}" for k, v in params.model_dump().items()) + return f"{self.name}:{':'.join(parts)}" if parts else self.name + + def receive(self, params: BaseModel | None, msg: BaseModel) -> BaseModel | None: + """What a client-sent message becomes for the group. None drops it.""" + return msg + + @classmethod + async def push(cls, message: BaseModel | dict, **params: Any) -> None: + """Send to every subscriber whose params key this group, from anywhere in the app.""" + channel = cls() + key = channel.group(_Params(params) if params else None) + await broadcast(key, cls.__name__, message, params) + + +class _Params: + """Params given as keywords rather than a model, so `group` can read them uniformly.""" + + def __init__(self, values: dict[str, Any]) -> None: + self._values = values + + def model_dump(self) -> dict[str, Any]: + return self._values + + +def register(channel_class: type[ReactChannel], name: str) -> None: + channel_class.name = name + _registry[name] = channel_class + + +def get_channel(name: str) -> type[ReactChannel] | None: + return _registry.get(name) + + +def registered() -> dict[str, type[ReactChannel]]: + return dict(_registry) + + +async def join(group: str, socket: Any) -> None: + async with _lock: + _groups[group].add(socket) + + +async def leave(group: str, socket: Any) -> None: + async with _lock: + _groups[group].discard(socket) + if not _groups[group]: + del _groups[group] + + +async def leave_all(socket: Any) -> None: + async with _lock: + for group in [g for g, sockets in _groups.items() if socket in sockets]: + _groups[group].discard(socket) + if not _groups[group]: + del _groups[group] + + +async def members(group: str) -> set[Any]: + async with _lock: + return set(_groups.get(group, ())) + + +async def broadcast( + group: str, type_name: str, message: BaseModel | dict, params: dict[str, Any] | None = None +) -> None: + """Deliver to the group. A socket that fails to take it has departed, and is dropped.""" + payload = { + "channel": group.split(":", 1)[0], + "params": params or {}, + "type": type_name, + "data": jsonable_encoder(message), + } + for socket in await members(group): + try: + await socket.send_json(payload) + except Exception as e: + print(f"mizan.channels: dropping subscriber from {group}: {type(e).__name__}: {e}") + await leave(group, socket) diff --git a/backends/mizan-fastapi/src/mizan_fastapi/websocket.py b/backends/mizan-fastapi/src/mizan_fastapi/websocket.py new file mode 100644 index 0000000..412e860 --- /dev/null +++ b/backends/mizan-fastapi/src/mizan_fastapi/websocket.py @@ -0,0 +1,192 @@ +""" +The WebSocket endpoint — channel subscriptions and RPC over one connection. + + Client sends: + {"action": "subscribe", "channel": "chat", "params": {...}} + {"action": "unsubscribe", "channel": "chat", "params": {...}} + {"action": "message", "channel": "chat", "params": {...}, "data": {...}} + {"action": "rpc", "id": "request-id", "fn": "function_name", "args": {...}} + + Server sends: + {"channel": "chat", "params": {...}, "type": "...", "data": {...}} + {"id": "request-id", "ok": true, "data": {...}} + {"id": "request-id", "ok": false, "error": {"code": "...", "message": "..."}} + {"error": "..."} + +The envelope is mizan-django's, so a client speaks to either backend unchanged. RPC +dispatches through the same `execute_function` the HTTP route calls — one dispatch path, +so the two transports cannot disagree about what a function does. + +`data` is the `{result, invalidate, merge}` envelope every non-HTTP transport hands the +kernel, so `mizanCall` applies a socket mutation's invalidation exactly as it applies an +HTTP one. mizan-django's socket sends a bare result and drops it. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from fastapi import APIRouter, WebSocket, WebSocketDisconnect + +from mizan_core.registry import get_function +from mizan_fastapi import channels +from mizan_fastapi.executor import ( + MizanError, + compute_invalidation, + compute_merges, + execute_function, +) + +logger = logging.getLogger(__name__) + +ws_router = APIRouter() + + +class _SocketRequest: + """What a server function receives when the call arrived over the socket. + + The HTTP path hands functions a Starlette `Request`; there is none here, so this + carries the surface a function actually reads — `state`, headers, and a method, since + a socket RPC sends data and expects an answer. + """ + + method = "POST" + + def __init__(self, socket: WebSocket) -> None: + self.state = socket.state + self.scope = socket.scope + self.headers = socket.headers + self.query_params = socket.query_params + self.socket = socket + + +def _params_model(channel_cls: Any, raw: dict[str, Any] | None) -> Any: + """Params as the channel's declared model, or a bare holder when it declares none.""" + model = getattr(channel_cls, "Params", None) + if model is not None and raw: + return model(**raw) + return channels._Params(raw) if raw else None + + +def _resolve(body: dict[str, Any]) -> tuple[Any, Any, str] | None: + """The channel class, its params, and the name — or None when the name is unknown.""" + name = body.get("channel") or "" + channel_cls = channels.get_channel(name) + if channel_cls is None: + return None + return channel_cls, _params_model(channel_cls, body.get("params")), name + + +async def _subscribe(socket: WebSocket, body: dict[str, Any]) -> None: + found = _resolve(body) + if found is None: + await socket.send_json({"error": f"unknown channel {body.get('channel')!r}"}) + return + channel_cls, params, name = found + + channel = channel_cls() + if not channel.authorize(params): + await socket.send_json({"error": f"not authorized for channel {name!r}"}) + return + + await channels.join(channel.group(params), socket) + hook = getattr(channel, "on_connect", None) + if hook is not None: + await hook(params) + + +async def _unsubscribe(socket: WebSocket, body: dict[str, Any]) -> None: + found = _resolve(body) + if found is None: + return + channel_cls, params, _ = found + channel = channel_cls() + await channels.leave(channel.group(params), socket) + hook = getattr(channel, "on_disconnect", None) + if hook is not None: + await hook() + + +async def _message(socket: WebSocket, body: dict[str, Any]) -> None: + found = _resolve(body) + if found is None: + await socket.send_json({"error": f"unknown channel {body.get('channel')!r}"}) + return + channel_cls, params, name = found + + channel = channel_cls() + if not channel.authorize(params): + await socket.send_json({"error": f"not authorized for channel {name!r}"}) + return + + outgoing = channel.receive(params, body.get("data") or {}) + if outgoing is None: + return # the channel dropped it + await channels.broadcast( + channel.group(params), type(outgoing).__name__, outgoing, body.get("params") + ) + + +async def _rpc(socket: WebSocket, body: dict[str, Any]) -> None: + request_id = body.get("id") + fn_name = body.get("fn") + if not fn_name: + await socket.send_json( + { + "id": request_id, + "ok": False, + "error": {"code": "BAD_REQUEST", "message": "rpc requires 'fn'"}, + } + ) + return + + args = body.get("args") or {} + try: + fn_class = get_function(fn_name) + result = await execute_function(_SocketRequest(socket), fn_name, args) + except MizanError as e: + payload: dict[str, Any] = {"code": e.code.value, "message": e.message} + if e.details: + payload["details"] = e.details + await socket.send_json({"id": request_id, "ok": False, "error": payload}) + return + + data: dict[str, Any] = { + "result": result, + "invalidate": compute_invalidation(fn_class, args), + } + merges = compute_merges(fn_class, args, result) + if merges: + data["merge"] = merges + await socket.send_json({"id": request_id, "ok": True, "data": data}) + + +_ACTIONS = { + "subscribe": _subscribe, + "unsubscribe": _unsubscribe, + "message": _message, + "rpc": _rpc, +} + + +@ws_router.websocket("/ws/") +async def mizan_socket(socket: WebSocket) -> None: + """One connection, every action. A disconnect ends the loop and clears the membership. + + A close is how a socket ends, so the disconnect is reported as the departure it is + rather than raised as a fault; `finally` clears the membership either way. + """ + await socket.accept() + try: + while True: + body = await socket.receive_json() + handler = _ACTIONS.get(body.get("action")) + if handler is None: + await socket.send_json({"error": f"unknown action {body.get('action')!r}"}) + continue + await handler(socket, body) + except WebSocketDisconnect as e: + logger.debug("mizan socket closed: code=%s reason=%s", e.code, e.reason) + finally: + await channels.leave_all(socket)