一个Python WebSocket服务端框架，支持零配置多文件路由管理，实现高效、模块化的实时通信开发。
Project description
NoWebsocket 库 使用文档
一个Python WebSocket服务端框架，支持零配置多文件路由管理，实现高效、模块化的实时通信开发。
安装指南
环境要求
- Python 3.7+
- 依赖库：无额外依赖
安装步骤
- 将库代码克隆到项目目录：
git clone git@github.com:dzy-china/NoWebsocket.git
- pip安装：
pip install NoWebsocket
快速开始
创建一个简单的WebSocket服务器
from NoWebsocket import WebSocketServer, WebSocketRouter, WebSocketApplication
# 定义应用类
class EchoApp(WebSocketApplication):
def on_message(self, message):
self.connection.send_text(f"Echo: {message}")
# 配置路由
router = WebSocketRouter()
router.add_route("/echo", EchoApp)
# 启动服务器
server = WebSocketServer(("0.0.0.0", 8765), router)
print("Server running on ws://localhost:8765/echo")
server.serve_forever()
测试连接
使用WebSocket客户端工具（如websocat）连接：
websocat ws://localhost:8765/echo
优雅开始
入口文件：main.py
from NoWebsocket import WebSocketServer, WebSocketRouter, Blueprint
def create_app():
router = WebSocketRouter()
# 自动注册所有蓝图
Blueprint.auto_register(
router,
package_path='blueprints',
bp_suffix='_bp'
)
return WebSocketServer(("0.0.0.0", 9000), router)
if __name__ == "__main__":
server = create_app()
print("WebSocket server running on ws://localhost:9000")
try:
server.serve_forever() # 启动服务
except KeyboardInterrupt:
server.shutdown() # 优雅关闭
print("\n🛑 Server stopped")
路由控制器处理文件：blueprints/chat/Uer.py
from NoWebsocket import WebSocketApplication, Blueprint
chat_bp = Blueprint(prefix='/chat')
@chat_bp.route("/youmi")
class User(WebSocketApplication):
def on_open(self):
"""连接建立时触发"""
pass
def on_message(self, message):
print(message)
self.connection.send_text(f"你好，我是悠米系统服务，收到了你的信息：\'{message}\'")
def on_binary(self, data):
"""收到二进制消息时触发"""
print(data)
def on_close(self):
"""连接关闭时触发"""
print("on_close")
测试连接 index.html
<!DOCTYPE html>
<html lang="zh-cn">
<head>
<title>WebSocket 示例</title>
</head>
<body>
<div>
<p>连接状态: <span id="status">未连接</span></p>
<input type="text" id="messageInput" placeholder="输入消息">
<button onclick="sendMessage()">发送</button>
</div>
<div id="messages"></div>
<script>
// 创建 WebSocket 连接
const socket = new WebSocket('ws://localhost:9000/chat/youmi'); // 使用公开测试服务器
// 连接打开时触发
socket.addEventListener('open', (event) => {
updateStatus('已连接');
logMessage('系统: 连接已建立');
// 发送初始测试消息
socket.send('你好，服务器!');
});
// 接收消息时触发
socket.addEventListener('message', (event) => {
logMessage(`服务器: ${event.data}`);
});
// 错误处理
socket.addEventListener('error', (event) => {
updateStatus('连接错误');
console.error('WebSocket 错误:', event);
});
// 连接关闭时触发
socket.addEventListener('close', (event) => {
updateStatus('连接已关闭');
logMessage(`系统: 连接关闭 (代码 ${event.code})`);
});
// 发送消息
function sendMessage() {
const input = document.getElementById('messageInput');
const message = input.value;
if (message) {
socket.send(message);
logMessage(`你: ${message}`);
input.value = '';
}
}
// 更新状态显示
function updateStatus(status) {
document.getElementById('status').textContent = status;
}
// 记录消息到页面
function logMessage(message) {
const messagesDiv = document.getElementById('messages');
const p = document.createElement('p');
p.textContent = message;
messagesDiv.appendChild(p);
}
</script>
</body>
</html>
配置选项
在服务器初始化时设置：
server = WebSocketServer(
("0.0.0.0", 8765),
router,
max_header_size=8192, # 最大请求头大小
max_message_size=4*1024*1024, # 最大消息大小（4MB）
read_timeout=60 # 等待读取信息超时（秒）
)
路由管理
装饰器路由
blueprints/chat/Uer.py
from NoWebsocket import WebSocketApplication, Blueprint
chat_bp = Blueprint(prefix='/chat') # 蓝图实例
@chat_bp.route("/youmi")
class User(WebSocketApplication):
def on_open(self):
"""连接建立时触发"""
pass
def on_message(self, message):
print(message)
self.connection.send_text(f"你好，我是悠米服务，收到了你的信息：\'{message}\'")
def on_binary(self, data):
"""收到二进制消息时触发"""
print(data)
def on_close(self):
"""连接关闭时触发"""
print("on_close")
约定配置
main.py
from NoWebsocket import Blueprint
Blueprint.auto_register(
router,
# __init__ _bp
# 自动扫描`blueprints`目录下所有以`.py`结尾的文件(__init__.py除外)
package_path="blueprints",
# `blueprints`目录下以`.py`结尾的文件内必须存在蓝图实例变量后缀`_bp`
# 如：'chat_bp = Blueprint(prefix='/chat')'
bp_suffix="_bp"
)
目录结构示例：
project/
├── blueprints/
│ └── chat/
│ └── User.py # 包含 chat_bp = Blueprint()
└── main.py
高级配置
自定义路由前缀
api_bp = Blueprint(prefix="/api/v2")
@api_bp.route("/status")
class StatusHandler(WebSocketApplication):
def on_open(self):
self.connection.send_text("API v2 Ready")
参数化路由
定义动态路径：
router.add_route("/user/{user_id:int}", UserHandler)
在应用中获取参数：
class UserHandler(WebSocketApplication):
def on_open(self):
user_id = self.path_params["user_id"]
self.connection.send_text(f"User ID: {user_id}")
示例应用
聊天应用
# blueprints/chat/handlers.py
from NoWebsocket import WebSocketApplication, Blueprint
chat_bp = Blueprint(prefix="/chat")
@chat_bp.route("/public")
class PublicChat(WebSocketApplication):
def on_message(self, message):
self.connection.send_text(f"[Public] {message}")
@chat_bp.route("/private/{room}")
class PrivateChat(WebSocketApplication):
def on_message(self, message):
room = self.path_params["room"]
self.connection.send_text(f"[{room}] {message}")
文件传输
# blueprints/file/handlers.py
from NoWebsocket import WebSocketApplication, Blueprint
file_bp = Blueprint(prefix="/file")
@file_bp.route("/upload")
class FileUpload(WebSocketApplication):
def on_binary(self, data):
with open("received_file.bin", "ab") as f:
f.write(data)
常见问题
1. 如何处理连接中断？
- 现象：客户端意外断开。
- 解决：在
on_close中清理资源：class MyApp(WebSocketApplication): def on_close(self): print("Connection closed")
2. 如何调试协议错误？
- 步骤：
- 启用详细日志：
import logging logging.basicConfig(level=logging.DEBUG)
- 检查握手头和帧格式是否符合RFC6455。
- 启用详细日志：
3. 如何扩展应用基类？
-
示例：添加身份验证：
class AuthApp(WebSocketApplication): def on_open(self): token = self.connection.request.headers.get("token") if not validate_token(token): self.connection.close(1008, "Invalid token")
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
nowebsocket-1.0.3.tar.gz
(10.7 kB
view details)
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file nowebsocket-1.0.3.tar.gz.
File metadata
- Download URL: nowebsocket-1.0.3.tar.gz
- Upload date:
- Size: 10.7 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.1.0 CPython/3.13.1
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
2f459aa68f8939615f882bf33d4844f31f36e954b3b059b3e8095cfe48512ade
|
|
| MD5 |
57522085550d6176fde9b6c497e8e5de
|
|
| BLAKE2b-256 |
6ab227560f51eb878c9e37356d3e4096c48198ddf66b851bc9c9b21d8ccf85df
|
File details
Details for the file nowebsocket-1.0.3-py3-none-any.whl.
File metadata
- Download URL: nowebsocket-1.0.3-py3-none-any.whl
- Upload date:
- Size: 13.0 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.1.0 CPython/3.13.1
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
4e4225145163450727dc9e33d3fb24491dda9297ab6ddf66e950ee7e07f4a93a
|
|
| MD5 |
2feb666c59b4f777ecac4c72a014968e
|
|
| BLAKE2b-256 |
26761cc56fbbfa41ea11b0e8bc3dfd43fd02b342b9accf12e29c6da33bc5854e
|
