大家好,我想分享一个库,以帮助**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更容易。