channel-box 0.5.1.7

ChannelBox it is a package for Starlette framework that allows you to make named webscoket channels.

channel-box

channel-box it is a package for Starlette & FastAPI framework that allows you send messages to named websocket channels from any part of your code.

Example of use:

• group chats
• notifications from backend
• alerts

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

Install

To install run:

pip install channel-box

Full working example [1] example/app.py

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

Full working example [2]

http://89.108.77.63:1025/
https://github.com/Sobolev5/LordaeronChat  

---

NGINX websocket setup

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

Check uvicorn installation

pip install uvicorn[standard]

Setup channel

from starlette.endpoints import WebSocketEndpoint
from channel_box import Channel, ChannelBox

class WsChatEndpoint(WebSocketEndpoint):

    async def on_connect(self, websocket):
        group_name = websocket.query_params.get("group_name")  # group name */ws?group_name=MyChat
        if group_name:
            channel = Channel(websocket, expires=60*60, encoding="json") # define user channel
            channel = await ChannelBox.channel_add(group_name, channel) # add user channel to named group
        await websocket.accept()

    async def on_receive(self, websocket, data):
        data = json.loads(data)
        message = data["message"]
        username = data["username"]     

        if message.strip():
            payload = {
                "username": username,
                "message": message,
            }
            group_name = websocket.query_params.get("group_name")
            if group_name:
                await ChannelBox.group_send(group_name, payload) # send to all users channels

Send messages

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

from channel_box import ChannelBox

await ChannelBox.channel_send(channel_name="MyChat", payload={"username": "Message from any part of your code", "message": "hello world"}, history=True) 

Get & flush channels:

from channel_box import ChannelBox

await ChannelBox.channels() 
await ChannelBox.channels_flush()  

Get & flush history:

from channel_box import ChannelBox

await ChannelBox.history() 
await ChannelBox.history_flush()

Tests

tox