Skip to main content

Python library for simple websocket managment

Project description

ci workflow PyPI Version GitHub

Websocket Rooms: websocket_rooms

A python library for creating WebSocket rooms, for message sharing or data distribution to multiple connections.

This library was created after building several real-time web apps and implementing the same mechanism to broadcast real-time messages between clients listening for the same real-time telemetries. The library simplifies the solution for this issue and proposes a simpler way to handle multiple WebSocket clients that act the same way.

Basic use:

Let's create a chatroom where everyone can post their messages:

from websocket_rooms import Room

chat_room = Room()

@chat_room.on_receive("json")
async def on_receive(room: Room, websocket: WebSocket, message: Any) -> None:
    await room.push_json(message)

@chat_room.on_connection
async def on_chatroom_connection(room: Room, websocket: WebSocket) -> None:
    logging.info("{} joined the chat room".format(websocket.client.host))

@chat_app.websocket("/chat")
async def connect_websocket(websocket: WebSocket):
    await chat_room.connect(websocket)

More advanced usage

Example of a more advanced use case, with modification to the Room base class: Suppose a class RoomWithClientId, where each WebSocket has a client_id associated with it, which it receives on connection:

class RoomWithClientId(Room):
    def __init__(self, base_room: Optional[BaseRoom] = None) -> None:
        super().__init__(base_room)
        self._id_to_ws = {}

    async def connect(self, websocket: WebSocket, client_id: int) -> None:
        self._id_to_ws[websocket] = client_id
        await super().connect(websocket)

    def get_client_id(self, websocket: WebSocket) -> int:
        return self._id_to_ws.get(websocket)


chat_room = RoomWithClientId()

@chat_room.on_receive("json")
async def on_chatroom_receive(room: RoomWithClientId, websocket: WebSocket, message: Any) -> None:
    await room.push_json(message)

@chat_room.on_connection
async def on_chatroom_connection(room: RoomWithClientId, websocket: WebSocket, client_id: int) -> None:
    logging.info("{} joined the chat room".format(client_id))

@app.websocket("/chat/{client_id}")
async def connect_websocket(websocket: WebSocket, client_id: int):
    await chat_room.connect(websocket, client_id)

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

websocket-rooms-0.1.1.tar.gz (8.2 kB view details)

Uploaded Source

File details

Details for the file websocket-rooms-0.1.1.tar.gz.

File metadata

  • Download URL: websocket-rooms-0.1.1.tar.gz
  • Upload date:
  • Size: 8.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/4.0.1 CPython/3.10.4

File hashes

Hashes for websocket-rooms-0.1.1.tar.gz
Algorithm Hash digest
SHA256 ed029bc6bcb84efd7b3eb41dcbc466427cc2bbf44e29a18dfaca9ce7262882a9
MD5 41a7aa5926730368f80e7060398d3ccf
BLAKE2b-256 6d0f9eb5bc3ea2c7e9004c18c95684e0bbe7c64a1d570cce531ddee191d29657

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page