avalon

Real-time web framework. HTTP routes, WebSockets and broadcast channels in one small, dependency-free package built on asyncio.

Avalon is for apps where the server pushes data to clients as it happens, such as chat, live dashboards, multiplayer cursors and notifications. You don't have to wire an HTTP framework, a WebSocket library and a pub/sub layer together yourself. Everything is plain standard-library Python you can read in one sitting.

Features

• Zero dependencies: pure standard library (asyncio, hashlib, json, ...), Python 3.10+.
• Decorator routing with typed path parameters: {id:int}, {x:float}, {rest:path}.
• Sync or async handlers: return a Response, a dict/list (JSON), str, bytes or None.
• WebSockets built in (RFC 6455): text/binary messages, fragmentation, automatic ping/pong, close handshake, size limits and async for msg in ws iteration.
• Broadcast channels: app.hub.join(room, ws) and await app.hub.broadcast(room, msg). Disconnected sockets are cleaned up automatically.
• HTTP/1.1 keep-alive, HEAD support and JSON error responses (HTTPError).
• WebSocket client (avalon.connect) for tests, scripts and service-to-service links.
• Fully type-hinted (py.typed).

Install

pip install avalon
# or, from a checkout:
pip install .

Quickstart

from avalon import Avalon

app = Avalon()

@app.get("/hello/{name}")
def hello(request):
    return {"hello": request.params["name"]}

@app.websocket("/ws/{room}")
async def room(ws):
    name = ws.params["room"]
    app.hub.join(name, ws)
    async for message in ws:
        await app.hub.broadcast(name, {"room": name, "text": message})

if __name__ == "__main__":
    app.run("127.0.0.1", 8000)

Talk to it from Python:

import asyncio
from avalon import connect

async def main():
    ws = await connect("ws://127.0.0.1:8000/ws/lobby")
    await ws.send("hi!")
    print(await ws.receive_json())   # {'room': 'lobby', 'text': 'hi!'}
    await ws.close()

asyncio.run(main())

A complete multi-room chat with a browser UI lives in examples/chat.py:

PYTHONPATH=src python3 examples/chat.py          # http://127.0.0.1:8000
PYTHONPATH=src python3 examples/chat.py --demo   # scripted two-client demo, then exit

API overview

Everything below is importable from the top-level avalon package.

Avalon(*, keep_alive_timeout=15.0)

GET routes also answer HEAD. Unknown paths return 404, a known path with the wrong method returns 405, and an unhandled exception returns 500 (logged on the avalon logger). All error bodies are JSON: {"error": "..."}.

Handler return values: to_response(value)

Anything else raises TypeError, which the app turns into a 500.

HTTP

• Request: method, path (URL-decoded), query: dict[str, str], headers: dict[str, str] (lower-case names), body: bytes, params: dict (converted path params), version; methods text() and json() (raises HTTPError(400) on invalid JSON); properties keep_alive and is_websocket.
• Response(body=b"", status=200, headers=None, media_type=None): the default media type is text/plain; charset=utf-8. encode(keep_alive=False) returns the wire bytes.
• JSONResponse(body, status=200, headers=None): serializes any JSON-compatible body with json.dumps.
• HTMLResponse(body, status=200, headers=None): text/html; charset=utf-8.
• HTTPError(status, detail=None): raise it in a handler to return {"error": detail} with that status. detail defaults to the standard reason phrase.

WebSockets

• WebSocket: path, params, headers, state (a free-form dict), closed, close_code, max_size
  • await ws.send(str | bytes) sends a text or binary message. await ws.send_json(obj) sends JSON as text.
  • await ws.receive() -> str | bytes and await ws.receive_json().
  • async for msg in ws: yields messages until the peer closes.
  • await ws.ping(data=b"") and await ws.close(code=1000, reason="").
• WebSocketClosed(code, reason): raised by receive/send once the connection is closed. Inside a handler you can let it propagate, because the app treats it as a normal disconnect. An unhandled exception in a handler closes the socket with code 1011.
• Opcode: frame opcode enum (TEXT, BINARY, CLOSE, PING, PONG, CONTINUATION).
• await connect(url, *, headers=None) -> WebSocket: client for ws:// URLs. It raises HandshakeError (with .status) if the server rejects the upgrade.

Channels: Hub

Routing

• Router(): add(path, handler, methods=("GET",), kind="http" | "websocket") -> Route and resolve(path, method="GET", kind="http") -> (Route, params), which raises HTTPError(404/405).
• Route: path, handler, methods, kind, match(path) -> dict | None.
• Path converters: {name} / {name:str} (one segment), {name:int}, {name:float}, {name:path} (may contain /). The first registered match wins.

Limitations

Avalon is deliberately small. It has no TLS (put it behind a reverse proxy), no WebSocket extensions or subprotocol negotiation, no chunked request bodies (they are rejected with 501), and hubs are in-process only, so they don't fan out across multiple workers.

Development

python3 -m unittest discover -s tests -v
# or, with pytest installed:
python3 -m pytest

License

MIT