chongming/chongming-utils-worker

Fork 0

Files

T

chakcy a58e92c21e modify: 插件风格统一

2026-08-14 14:33:19 +08:00

8.5 KiB

Raw Blame History

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"   # 自动在 data_dir 下建 SQLite 库；不需要则置 None
    auto_bind = True           # 首次初始化时自动 auto_bind_models(self.type)

    async def on_init(self):
        # 数据库已由基类自动初始化（self.db）
        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):
        # 关闭自定义客户端等（后台任务与数据库由基类统一处理）
        ...


# 暴露给 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` 下）；`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()
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 响应模型类

装饰器行为：

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

请求信封协议

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），缺省 GET
meta.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 配置解析

8.5 KiB Raw Blame History Unescape Escape