DOCSEARCHbyAlgolia
Submit Docs

fastapi_websocket_pubsub

repository·master·Indexed 20 days ago

https://github.com/permitio/fastapi_websocket_pubsub

A high-performance, durable Pub/Sub implementation for FastAPI and WebSockets. It enables real-time, multi-cast communication between servers and clients, with support for scaling across multiple server instances using Redis, Kafka, or Postgres backends via Broadcaster.

Tokens
1.3K
Snippets
6
Records
6
Agent score
20%

What's inside fastapi_websocket_pubsub

  1. Scale Pub/Sub across multiple server instances using Broadcaster

    master

    To ensure clients receive messages regardless of which server instance they are connected to, initialize PubSubEndpoint with a broadcaster connection string. This requires installing the appropriate backend dependencies (e.g., Redis, Postgres, or Kafka).

    Required dependencies:

    • Redis: pip install fastapi_websocket_pubsub[redis]
    • Postgres: pip install fastapi_websocket_pubsub[postgres]
    • Kafka: pip install fastapi_websocket_pubsub[kafka]
    • All: pip install fastapi_websocket_pubsub[all]

    Implementation pattern: When using a broadcaster, you must manually handle the websocket route by calling endpoint.main_loop(websocket) within your websocket endpoint.

    app = FastAPI()
# Use a broadcaster backend (e.g., Postgres) to sync instances
endpoint = PubSubEndpoint(broadcaster="postgres://localhost:5432/")

@app.websocket("/pubsub")
async def websocket_rpc_endpoint(websocket: WebSocket):
    # Manually route the websocket to the pubsub main loop
    await endpoint.main_loop(websocket)

  3. Full usage example: Triggering Pub/Sub via HTTP

    master

    This example demonstrates a common pattern: a client subscribes to a topic via WebSockets, and a standard HTTP GET request on the server triggers a publication to that topic.

    # --- SERVER SIDE ---
import asyncio
import uvicorn
from fastapi import FastAPI
from fastapi_websocket_pubsub import PubSubEndpoint

app = FastAPI()
endpoint = PubSubEndpoint()
endpoint.register_route(app, "/pubsub")

@app.get("/trigger")
async def trigger_events():
    # Trigger the event via HTTP
    endpoint.publish(["triggered"])

# --- CLIENT SIDE ---
from fastapi_websocket_pubsub import PubSubClient

async def on_trigger(data):
    print("Trigger URL was accessed")

async def run_client():
    async with PubSubClient(server_uri="ws://localhost/pubsub") as client:
        client.subscribe("triggered", on_trigger)
        # Keep client running to listen
        await asyncio.sleep(60)

  4. Subscribe to events with PubSubClient

    master

    To receive events on the client side, use PubSubClient.subscribe(topic, callback). The callback is an async function that receives the event data. The client is typically used as an async context manager.

    # Callback to be called upon event being published on server
async def on_event(data):
    print("We got an event! with data- ", data)

# Subscribe for the event 
async with PubSubClient(server_uri="ws://localhost/pubsub") as client:
    client.subscribe("my event", on_event)

  5. Publish events from the server with PubSubEndpoint

    master

    To broadcast events from your FastAPI server to connected clients, use PubSubEndpoint.publish(topics, data). You must first register the endpoint route using register_route(app, path).

    from fastapi import FastAPI
from fastapi_websocket_pubsub import PubSubEndpoint

app = FastAPI() 
endpoint = PubSubEndpoint()
endpoint.register_route(app, path="/pubsub")

# Publish to specific topics with data
endpoint.publish(["my_event_topic"], data=["my", "data", 1])

  6. Configure logging for fastapi_websocket_pubsub

    master

    The library uses fastapi-websocket-rpc for logging. You can control the logging output using the logging_config.set_mode helper or the WS_RPC_LOGGING environment variable. All loggers are prefixed with fastapi.ws_rpc.pubsub.

    # Set RPC to log in the same style as Uvicorn
from fastapi_websocket_rpc.logger import logging_config, LoggingModes
logging_config.set_mode(LoggingModes.UVICORN)