loopws 0.2.0

WebSocket event bus — real-time event push over persistent connections

loopws

A standalone WebSocket event bus for Python — real-time event push over persistent connections. Pure Python + asyncio, zero required dependencies, pluggable Redis backend for multi-server deployments.

What it is

loopws is a library, not a service. You bring your own WebSocket framework (FastAPI, Starlette, aiohttp, anything that satisfies its structural protocol) and your own presence store (in-memory for dev, Redis for prod). loopws provides:

• A per-connection handler with built-in dispatch, ping/pong keepalive, subscribe/unsubscribe, and lifecycle hooks.
• A local connection registry that tracks which devices are on this server.
• A subscription matcher with service:event patterns (loopbooks:*, *:invoice.paid, etc.).
• An in-process push_event helper for same-server delivery.
• A cross-server pub/sub manager with exponential-backoff-with-jitter reconnect.
• A frozen-dataclass LoopWSConfig that bundles every timing tunable.

Install

pip install loopws                 # core only, zero dependencies
pip install loopws[redis]          # with Redis adapter
pip install loopws[dev]            # with test suite extras

Requires Python 3.11+.

Minimal example (FastAPI)

from uuid import UUID
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from loopws import BaseWebSocketHandler, LocalConnectionManager

class InMemoryStore:
    def __init__(self): self._d = {}
    async def register(self, a, d, s): self._d[(a, d)] = s
    async def unregister(self, a, d):  self._d.pop((a, d), None)
    async def get_server(self, a, d):  return self._d.get((a, d))
    async def refresh(self, a, d):     pass

store = InMemoryStore()
connection_mgr = LocalConnectionManager(store, server_id="pod-1")
app = FastAPI()

@app.websocket("/ws")
async def endpoint(ws: WebSocket, account_id: UUID, device_id: int):
    await ws.accept()
    handler = BaseWebSocketHandler(
        ws=ws, account_id=account_id, device_id=device_id,
        connection_mgr=connection_mgr,
        services={"myapp"},
        default_subscriptions={"myapp:*"},
    )
    await connection_mgr.register(account_id, device_id, ws, handler=handler)
    await handler.on_connect()
    handler.start_keepalive()
    try:
        while True:
            await handler.handle_message(await ws.receive_text())
    except WebSocketDisconnect:
        pass
    finally:
        handler.stop_keepalive()
        await handler.on_disconnect()
        await connection_mgr.unregister(account_id, device_id)

Push events from anywhere in the process:

from loopws import push_event

result = await push_event(
    connection_mgr,
    account_id=user_id,
    device_id=42,
    service="myapp",
    event="invoice.paid",
    data={"invoice_id": 9001},
)
# result: "delivered" | "filtered" | "not_connected" | "send_failed"

Multi-server with Redis

import redis.asyncio as aioredis
from loopws import PubSubManager
from loopws.redis import RedisConnectionManager, RedisPubSubBackend

redis_client = aioredis.from_url("redis://localhost:6379")
store   = RedisConnectionManager(redis_client)          # presence with TTL
backend = RedisPubSubBackend(redis_client)              # publish + subscribe

pubsub = PubSubManager(backend, server_id="pod-1")
pubsub.set_message_handler(on_cross_server_message)
await pubsub.start()

PubSubManager runs a real subscriber loop that reconnects on Redis failures with exponential backoff + full jitter.

Documentation

• INTEGRATION_GUIDE.md — step-by-step integration for new consumers
• SPECIFICATION.md — full technical & functional spec
• CHANGELOG.md — release history
• UPGRADING.md — migration notes per release
• OPERATIONS_RUNBOOK.md — deploy and ops guide

License

MIT — see LICENSE.