大家好,我想分享一个库,以帮助**FastAPI更容易更结构化地处理WebSockets**,作为[https://github.com/encode/broadcaster]的替代品,后者已经被存档。
基本上,它允许从**任何地方广播WebSocket消息:** 背景作业,HTTP端点或脚本。它也使得处理**群组消息**更加容易(集成广播支持)。此库**完全支持类型提示**并支持**自动AsyncAPI文档生成**。
以下是一个使用**chanx**定义WebSocket 的 FastAPI 方式 :
@channel(name="chat", description="实时聊天 API")
class ChatConsumer(AsyncJsonWebsocketConsumer):
groups = ["chat_room"] # 连接时自动加入此组
"""
@ws_handler(
summary = "处理聊天消息",
output_type=ChatNotificationMessage
)
同步函数handle_chat( self, message: ChatMessage ) -> None:
#广播到所有组客户端
self.broadcast_message(
ChatNotificationMessage (
payload=ChatPayload (message= f"用户: {message.payload.message} ")
)
)
"""
此包已发布在 PyPI 上 [https://pypi.org/project/chanx/ ]
存储库已发布在 [https://github.com/huynguyengl99/chanx]
FastAPI(包括房间聊天和后台作业)使用教程已被包含在以下文档中:[https://chanx.readthedocs.io/en/latest/tutorial-fastapi/prerequisites.html]
希望大家喜欢并使处理 WebSocket更容易。
评论 (0)