ChannelBox it is a package for Starlette framework that allows you to make named webscoket channels.
Project description
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 for user groups
https://github.com/Sobolev5/channel-box
Install
To install run:
pip install channel-box
Full working example
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
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,
payload_type="json",
) # Create new user channel
channel_add_status = await ChannelBox.add_channel_to_group(
channel=channel,
group_name=group_name,
) # Add channel to named group
sprint(channel_add_status)
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=group_name,
payload=payload,
save_history=True,
) # Send to all user channels
## Send messages
Send message to any channel from any part of your code:
```python
from channel_box import ChannelBox
await ChannelBox.channel_send(
group_name="MyChat",
payload={
"username": "Message from any part of your code",
"message": "hello world"
},
save_history=True,
)
Show & flush groups:
from channel_box import ChannelBox
await ChannelBox.show_groups()
await ChannelBox.flush_groups()
Show & flush history:
from channel_box import ChannelBox
await ChannelBox.show_history()
await ChannelBox.flush_history()
Tests
tox
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
No source distribution files available for this release.See tutorial on generating distribution archives.
Built Distribution
File details
Details for the file channel_box-1.0.0-py3-none-any.whl
.
File metadata
- Download URL: channel_box-1.0.0-py3-none-any.whl
- Upload date:
- Size: 8.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/5.1.0 CPython/3.11.6
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 51c60cd2ae50f815c4b3931f7b46ce8e3a0ea788d7a904e4c66667004dc8a5dd |
|
MD5 | 05de0ac9c07bfd852be20c3d388f70d8 |
|
BLAKE2b-256 | 1241f0d705360a79d9f97173549efc1c8e06102a6b73ec7c5d6d0f4800ad9aa9 |