chongming-worker (0.4.5)
Published 2026-09-18 15:32:03 +08:00 by chakcy
Installation
pip install --index-url http://124.71.68.6:3000/api/packages/chongming/pypi/simple/ --extra-index-url https://pypi.org/simple chongming-workerAbout this package
chongming worker
chongming-worker
通用 Worker 框架核心，提供 Zenoh RPC 端点的装饰器定义、参数解析、生命周期管理和配置加载。
安装
uv pip install -e utils/worker
核心功能
@rpc_method装饰器：将异步函数注册为 Zenoh Queryable 端点（同subject#METHOD冲突注册即报错）- 自动参数解析：根据 HTTP 方法从 query 参数或 JSON body 解析请求
- Pydantic 模型绑定：
auto_bind_models()自动绑定生成的模型 - 信封协议：请求负载统一为
{"meta": {...}, "data": {...}}，运行时按meta.method分派 - Worker 运行时：
create_worker/create_workers管理生命周期（结构化日志） Plugin基类：统一插件初始化 / 清理生命周期（幂等、日志、数据库、后台任务、清理钩子）- 配置加载：
load_worker_config读取 TOML 配置
Plugin 基类（推荐）
chongming_worker.plugin.Plugin 统一了各插件 __init__.py 的初始化/清理风格，
消除重复样板（数据库初始化、global_var 写入、幂等保护、后台任务、日志、清理钩子）。
最小示例
# src/plugins/my_plugin/__init__.py
import logging
from chongming_worker.plugin import Plugin
from . import router # noqa: F401 触发 @rpc_method 注册
# from .routes import router # 可选：自定义 FastAPI 路由（网关自动发现）
logger = logging.getLogger(__name__)
class MyPlugin(Plugin):
type = "my_plugin"
db_name = "my_plugin.db" # SQLite：自动在 data_dir 下建库；config 含 database.url 时优先走 PostgreSQL；不需要则置 None
auto_bind = True # 首次初始化时自动 auto_bind_models(self.type)
async def on_init(self):
# 数据库已由基类自动初始化（self.db，SQLite 或 PostgreSQL）
self.set_var("db", self.db) # 写入插件 global_var 模块
# self.spawn_task(self._background_loop()) # 后台任务（cleanup 自动取消）
# self.register_cleanup(self._close_client) # 注册清理钩子（逆序执行）
async def on_cleanup(self):
# 关闭自定义客户端等（后台任务与数据库由基类统一处理）
...
# PostgreSQL 接入（可选）：在插件 config.toml 配置 database 段后，
# 基类 init_database() 检测到 url 即走 Database.from_config 接入 PG::
#
# # config/workers/my_plugin/config.toml
# [workers.config.database]
# type = "postgresql"
# url = "postgresql+asyncpg://user:pwd@127.0.0.1:5432/mydb"
# generate_schemas = true
# 对象存储接入（可选）：在插件 config.toml 配置 storage 段后，
# 基类 init_storage() 自动初始化 self.storage（S3/Minio，直连不经网关）::
#
# # config/workers/my_plugin/config.toml
# [workers.config.storage]
# endpoint = "192.168.1.10:9000" # Minio 地址 host:port（不带协议前缀）
# access_key = "minioadmin"
# secret_key = "minioadmin"
# bucket = "chongming-files"
# secure = false
# # region = "" # 可选
# # auto_create_bucket = true # 默认 true：启动时确保 bucket 存在
#
# 语义：未配置 / enabled=false → self.storage 为 None（插件自行降级）；
# 缺必填键（endpoint/access_key/secret_key/bucket）→ 启动硬失败（ValueError）；
# 连接失败 → 告警降级为 None。API 见 chongming_worker.storage.StorageClient
# （put_file/put_bytes/get_file/get_bytes/delete/presigned_put_url/presigned_get_url
# 及对应 async a* 版本，handler 中直接 await）。
# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法一致）
plugin = MyPlugin()
create_workers = plugin.create_workers
cleanup = plugin.cleanup
子类可覆写点
| 属性 / 方法 | 说明 |
|---|---|
type |
插件类型（worker type），匹配 subject 前缀与 config/workers/<type>（必填） |
db_name |
默认 SQLite 文件名（放 data_dir 下）；config 含 database.url 时优先走 PostgreSQL；None 表示不自动建库 |
auto_bind |
首次初始化时是否自动 auto_bind_models(self.type) |
on_init() |
首次 create_workers 时调用一次，初始化资源 |
on_cleanup() |
cleanup 时调用，清理自定义资源 |
框架自动处理
- 幂等：重复
create_workers只初始化一次（asyncio.Lock守卫，并发调用安全） - 统一日志：
logger自动带[type]前缀 - 数据库：
init_database()统一建库建表、记录，cleanup自动dispose()（SQLite 按db_name；config 含database.url时走 PostgreSQL） - 对象存储：
init_storage()按 config 的storage段初始化self.storage（S3/Minio，可选），cleanup自动关闭 - global_var：
set_var()/get_var()统一读写插件global_var模块；cleanup时自动将已发布的变量重置为None，避免业务代码拿到已释放的资源 - 后台任务：
spawn_task()跟踪，cleanup时自动取消并gather - 清理钩子：
register_cleanup()注册，cleanup时逆序执行 - 委托：自动调用
runtime.create_workers创建 Worker 副本（经_create_workers_impl，测试可替换桩）
API
@rpc_method 装饰器
from chongming_worker.decorators import rpc_method
from ..models.pydantic import AddGetRequest, AddGetResponse
@rpc_method(
subject="example.add",
method="GET",
response_model=AddGetResponse,
)
async def add(request: AddGetRequest) -> AddGetResponse:
return AddGetResponse(result=request.a + request.b)
| 参数 | 类型 | 说明 |
|---|---|---|
subject |
str |
Zenoh 主题（必需），如 "example.add" |
method |
str |
HTTP 方法："GET" / "POST" / "MQTT" |
response_model |
type[BaseModel] |
Pydantic 响应模型类 |
stream |
bool |
流式端点：函数返回生成器，框架逐块多回复推送（默认 false） |
装饰器行为：
| method | 参数来源 | 请求绑定 |
|---|---|---|
GET |
URL query string | TOML params 定义 |
POST |
JSON body | 自动绑定到 Pydantic 模型 |
MQTT |
MQTT payload JSON | 同 POST |
流式端点（SSE 多回复）
@rpc_method(..., stream=True) 的函数返回生成器（async / sync 均可），框架自动将每个
yield 的 dict 作为独立 Zenoh 回复块推送（Query/Reply 原生多回复），网关以 SSE
（text/event-stream）逐块转发给 HTTP 客户端。无需端口、进程或新依赖：
@rpc_method(subject="example.stream_chat", method="POST", stream=True)
async def stream_chat(prompt: str = "你好", tokens: int = 10):
for i in range(tokens):
await asyncio.sleep(0.05) # 模拟逐 token 生成
yield {"token": f"{prompt}·{i + 1}", "index": i}
yield {"done": True}
协议约定（框架自动处理，插件无需关心底层实现）：
- 生成器正常结束 = 流结束。没有协议层结束哨兵；
{"done": True}这类字段只是业务标记 - 生成器中途异常 → 框架发送
{"error": ...}收尾块后停流 - 客户端断开 / 查询被网关放弃 → 逐块 reply 失败时框架立即停流，避免 Worker 空转
- 每个块必须是可 JSON 序列化的对象，单块建议小于 64KB（zenoh 消息上限内）
网关侧配合：对应 TOML 方法节声明 stream = true 并配大 timeout（流总寿命上限），
详见网关仓库 README「流式端点（SSE）」。非流式端点（不声明 stream）行为完全不变。
请求信封协议
Worker 运行时收到的请求负载统一为信封结构（网关 HTTP 路由 / MQTT Bridge / chongming-request Client 自动组装）：
{
"meta": { "method": "GET", "user_info": {...}, "timeout": 30, "cacheable": false },
"data": { "a": 1, "b": 2 }
}
meta.method决定分派到哪个@rpc_method（GET/POST/.../MQTT），缺省GETmeta.user_info注入到签名为user_info: dict | None = None的处理器参数data为业务参数，经请求模型校验后传给处理器- 非信封载荷（无
meta/data）按旧行为兜底：整包视为data、method 取 GET，并输出一次告警
注册表冲突检测
@rpc_method 注册时以 subject#METHOD 为唯一键；同一键重复注册且函数不同时直接抛
ValueError（含双方函数位置），不再静默覆盖——静默覆盖会悄悄改变运行时分派行为。
auto_bind_models(worker_type)
自动将 gen_models 生成的 Pydantic 模型绑定到对应端点，省去手动导入和绑定。
from chongming_worker import auto_bind_models
auto_bind_models("example")
create_workers
from chongming_worker.runtime import create_workers
tasks = create_workers(
session=zenoh_session,
config=plugin_config,
count=3,
worker_manager=worker_manager,
)
| 参数 | 类型 | 说明 |
|---|---|---|
session |
zenoh.Session |
Zenoh 会话 |
config |
dict |
插件配置 |
count |
int |
Worker 副本数 |
worker_manager |
WorkerManager |
Worker 管理器实例 |
load_worker_config
from chongming_worker.config import load_worker_config
cfg = load_worker_config("example")
插件集成模式
推荐使用上文的 Plugin 基类（
src/plugins/下所有插件均已迁移到该风格）； 以下为不依赖基类的传统写法，仅框架层面保留兼容，新插件请勿采用。
# src/plugins/my_plugin/__init__.py（旧式写法，已不推荐）
from chongming_worker import auto_bind_models
from chongming_worker.runtime import create_workers as _create_workers
from .router import * # 触发 @rpc_method 注册
auto_bind_models("my_plugin")
_initialized = False
async def create_workers(session, config, count, worker_manager):
"""插件入口：初始化资源，启动 Worker"""
global _initialized
if _initialized:
return _create_workers(session, config, count, worker_manager)
from chongming_database import Database
from .global_var import set_db
db = Database(...)
await db.initialize()
set_db(db)
_initialized = True
return _create_workers(session, config, count, worker_manager)
async def cleanup(worker_item: str):
from .global_var import db
await db.dispose()
导出清单
from chongming_worker import (
RPCMethod, rpc_method, # 装饰器
bind_model, auto_bind_models, # 模型绑定
get_rpc_methods, get_all_rpc_methods, clear_registry, # 注册表
generate_route_configs, # 路由配置生成
load_worker_config, # TOML 配置加载
WorkerConfig, OperationConfig, # 配置类型
create_worker, create_workers, # 运行时
Plugin, # 统一生命周期基类
)
依赖
zenoh— Zenoh Python 绑定pydantic— 数据验证与序列化tomli— TOML 配置解析
Requirements
Requires Python: >=3.12
Details
2026-09-18 15:32:03 +08:00
Assets (2)
Versions (8)
View all
PyPI
17
147 KiB