A simple but yet elegant publish/subscribe socket implementation using FastAPI Websocket.

pip install fastapi-pubsub

FastAPI provides an awesome implementation of Websockets. But it can be painful to configure it to work as a publish/subscribe mechanism.

FastAPI Pub/Sub implemented a BasePubSubWebSocket that handles all client's connections under the hood. All you have to do is extend this class, overriding the method generate_object_to_send(self) with your desired logic.

from fastapi_pubsub import BasePubSubWebSocket

class SimplePubSubWebsocket(BasePubSubWebSocket):
    async def generate_object_to_send(self):
        # any logic that you want to implement to send to connected users
        connections = self.connection_manager.get(self.websocket_id)
        return {"num_connected_clients": len(connections)}

app = FastAPI()

@app.websocket("/ws/connected_clients")
async def online_players_endpoint(websocket: WebSocket):
    ssw = SimplePubSubWebsocket(websocket, 'connected_clients')
    await ssw.run()

A minimum example of a server/client communication can be seen here. Follow the instructions to see it working.

Despite being full functional, this project is in a very early stage. Pull requests are very welcome!