channel-box 0.4.0

channel-box it is a simple tool for Starlette framework that allows you to make named webscoket channels.

channel-box

channel-box it is a simple tool for Starlette framework that allows you send messages to named websocket channels.

Example of use:

• group chats
• notifications from backend
• alerts

https://github.com/Sobolev5/channel-box

Install

To install run:

pip install channel-box

[important!] see full working example example/app.py

https://channel-box.andrey-sobolev.ru/
https://github.com/Sobolev5/channel-box/tree/master/example

NGINX websocket setup

http://nginx.org/en/docs/http/websocket.html

Check uvicorn installation

pip install uvicorn[standard]

Setup channel

class Channel(ChannelBoxEndpoint):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.expires = 1600
        self.encoding = "json"

    async def on_connect(self, websocket):
        channel_name = websocket.query_params.get("channel_name", "MySimpleChat")  # channel name */ws?channel_name=MySimpleChat
        await self.channel_get_or_create(channel_name, websocket) 
        await websocket.accept()

    async def on_receive(self, websocket, data):
        message = data["message"]
        username = data["username"]     
        if message.strip():
            payload = {
                "username": username,
                "message": message,
            }
            await self.channel_send(payload, history=True)

Send messages

Send message to any channel from any part of your code:

from channel_box import channel_box
await channel_box.channel_send(channel_name="MySimpleChat", payload={"username": "Message HTTPEndpoint", "message": "hello from Message"}, history=True) 

Get & flush channels:

await channel_box.channels() 
await channel_box.channels_flush()  

Get & flush history:

await channel_box.history() 
await channel_box.history_flush()

TODO

typing
__doc__ strings

Try my free time tracker

My free time tracker for developers Workhours.space.