chongming-worker

通用 Worker 框架核心，提供 Zenoh RPC 端点的装饰器定义、参数解析、生命周期管理和配置加载。

---

安装

uv pip install -e utils/worker

---

核心功能

• @rpc_method 装饰器：将异步函数注册为 Zenoh Queryable 端点
• 自动参数解析：根据 HTTP 方法从 query 参数或 JSON body 解析请求
• Pydantic 模型绑定：auto_bind_models() 自动绑定生成的模型
• Worker 运行时：create_worker / create_workers 管理生命周期
• 配置加载：load_worker_config 读取 TOML 配置

---

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 响应模型类

装饰器行为：

method	参数来源	请求绑定
GET	URL query string	TOML params 定义
POST	JSON body	自动绑定到 Pydantic 模型
MQTT	MQTT payload JSON	同 POST

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")

---

插件集成模式

# 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")

async def create_workers(session, config, count, worker_manager):
    """插件入口：初始化资源，启动 Worker"""
    from chongming_database import Database
    from .global_var import set_db

    db = Database(...)
    await db.initialize()
    set_db(db)

    return _create_workers(session, config, count, worker_manager)

async def cleanup(worker_manager):
    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,  # 运行时
)

---

依赖

• zenoh — Zenoh Python 绑定
• pydantic — 数据验证与序列化
• tomli — TOML 配置解析