一个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
blueprints包下必须存在
__init__.py,该文件可以为空， 一个目录必须包含__init__.py文件才能被识别为包（Package），这是Python模块系统的约定
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,
package_path="blueprints", # 指定路由文件存放的包，包内必须存在`__init__.py`文件，该文件可以为空
# 指定蓝图实例变量后缀，不写默认为'_bp'
# 如：'chat_bp = Blueprint(prefix='/chat')'
bp_suffix="_bp"
)
bp_suffix参数的存在是为了平衡灵活性与规范性，它通过命名约定帮助开发者更精确地控制哪些蓝图实例应被自动注册
目录结构示例：
project/
├── blueprints/
__init__.py # 必须存在，可以为空
│ └── 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-2.0.0.tar.gz
(11.0 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-2.0.0.tar.gz.
File metadata
- Download URL: nowebsocket-2.0.0.tar.gz
- Upload date:
- Size: 11.0 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.1.0 CPython/3.13.1
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
0736d6b08bf010015aaeceb5d76d9db4d9af035a56a909fb999dc460c91a273d
|
|
| MD5 |
d3e807838b3be93e462dde5f0b7f4c5a
|
|
| BLAKE2b-256 |
d2e3a2595a374f1603e509001720ddb92c769f9cf3eab81d6eec5394108c6d3d
|
File details
Details for the file nowebsocket-2.0.0-py3-none-any.whl.
File metadata
- Download URL: nowebsocket-2.0.0-py3-none-any.whl
- Upload date:
- Size: 13.2 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 |
3aca7faf26acff311f1f011af3b2fcf54f48186d114c19914a30acfa3d0aedee
|
|
| MD5 |
af60ea687917f87dc3581566a250bf4e
|
|
| BLAKE2b-256 |
6c2e517da1170e6c190e60c6e8e7f250ae90dfb3f705a4f1739c1a5b2cadd24f
|
