chongming/chongming-utils-worker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modify: 插件风格统一

Browse Source
This commit is contained in:
chakcy
2026-08-14 14:33:19 +08:00
parent 1ec0b2329c
commit a58e92c21e
4 changed files with 142 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+36 -16
View File
@@ -14,10 +14,11 @@ uv pip install -e utils/worker
## 核心功能
- **`@rpc_method` 装饰器**：将异步函数注册为 Zenoh Queryable 端点
- **`@rpc_method` 装饰器**：将异步函数注册为 Zenoh Queryable 端点（同 `subject#METHOD` 冲突注册即报错）
- **自动参数解析**：根据 HTTP 方法从 query 参数或 JSON body 解析请求
- **Pydantic 模型绑定**：`auto_bind_models()` 自动绑定生成的模型
- **Worker 运行时**：`create_worker` / `create_workers` 管理生命周期
- **信封协议**：请求负载统一为 `{"meta": {...}, "data": {...}}`，运行时按 `meta.method` 分派
- **Worker 运行时**：`create_worker` / `create_workers` 管理生命周期（结构化日志）
- **`Plugin` 基类**：统一插件初始化 / 清理生命周期（幂等、日志、数据库、后台任务、清理钩子）
- **配置加载**：`load_worker_config` 读取 TOML 配置
@@ -34,7 +35,6 @@ uv pip install -e utils/worker
# src/plugins/my_plugin/__init__.py
import logging
from chongming_worker import auto_bind_models
from chongming_worker.plugin import Plugin
from . import router # noqa: F401 触发 @rpc_method 注册
@@ -42,8 +42,6 @@ from . import router # noqa: F401 触发 @rpc_method 注册
logger = logging.getLogger(__name__)
auto_bind_models("my_plugin")
class MyPlugin(Plugin):
type = "my_plugin"
@@ -79,13 +77,13 @@ cleanup = plugin.cleanup
### 框架自动处理
- **幂等**：重复 `create_workers` 只初始化一次（旧写法需手写 `_initialized` 标志）
- **幂等**：重复 `create_workers` 只初始化一次（`asyncio.Lock` 守卫，并发调用安全）
- **统一日志**：`logger` 自动带 `[type]` 前缀
- **数据库**：`init_database()` 统一建路径、建表、记录，`cleanup` 自动 `dispose()`
- **global_var**：`set_var()` / `get_var()` 统一读写插件 `global_var` 模块
- **global_var**：`set_var()` / `get_var()` 统一读写插件 `global_var` 模块；`cleanup` 时自动将已发布的变量重置为 `None`，避免业务代码拿到已释放的资源
- **后台任务**：`spawn_task()` 跟踪，`cleanup` 时自动取消并 `gather`
- **清理钩子**：`register_cleanup()` 注册，`cleanup` 时逆序执行
- **委托**：自动调用 `runtime.create_workers` 创建 Worker 副本
- **委托**：自动调用 `runtime.create_workers` 创建 Worker 副本（经 `_create_workers_impl`，测试可替换桩）
---
@@ -114,11 +112,32 @@ async def add(request: AddGetRequest) -> AddGetResponse:
**装饰器行为**：
| method | 参数来源 | 请求绑定 |
| ------ | ----------------- | ------------------------ |
| `GET` | URL query string | TOML `params` 定义 |
| `POST` | JSON body | 自动绑定到 Pydantic 模型 |
| `MQTT` | MQTT payload JSON | 同 POST |
| method | 参数来源 | 请求绑定 |
| ------ | ------------------ | ------------------------ |
| `GET` | URL query string | TOML `params` 定义 |
| `POST` | JSON body | 自动绑定到 Pydantic 模型 |
| `MQTT` | MQTT payload JSON | 同 POST |
### 请求信封协议
Worker 运行时收到的请求负载统一为信封结构（网关 HTTP 路由 / MQTT Bridge / `chongming-request` Client 自动组装）：
```json
{
"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)`
@@ -162,10 +181,11 @@ cfg = load_worker_config("example")
## 插件集成模式
> 推荐使用上文的 **Plugin 基类**；以下为不依赖基类的传统写法（仍受支持）。
> 推荐使用上文的 **Plugin 基类**（`src/plugins/` 下所有插件均已迁移到该风格）；
> 以下为不依赖基类的传统写法，仅框架层面保留兼容，新插件请勿采用。
```python
# src/plugins/my_plugin/__init__.py
# 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 注册
@@ -188,7 +208,7 @@ async def create_workers(session, config, count, worker_manager):
_initialized = True
return _create_workers(session, config, count, worker_manager)
async def cleanup(worker_manager):
async def cleanup(worker_item: str):
from .global_var import db
await db.dispose()
```
src/chongming_worker/decorators.py
+27 -7
View File
@@ -15,10 +15,13 @@
"""
import inspect
import importlib
import logging
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, get_type_hints, Union
from pydantic import BaseModel, Field, create_model, ValidationError
logger = logging.getLogger("chongming_worker.decorators")
# ============================================================
# 类型系统
@@ -204,7 +207,8 @@ class RPCMethod:
try:
validated = req_model(**payload)
result = validated.model_dump()
# 保留原始 payload 中不属于请求模型的额外字段（如 _method、user_info 等）
# 保留原始 payload 中不属于请求模型的额外字段（信封协议下通常为空，
# 框架字段 method/user_info 由 runtime 从 meta 信封提取注入）
for k, v in payload.items():
if k not in result:
result[k] = v
@@ -342,6 +346,19 @@ def rpc_method(
_RPC_REGISTRY[worker_type] = {}
# 使用 subject + "#" + method 作为唯一 key，允许多个 HTTP 方法共享同一 subject
registry_key = f"{subject}#{method.upper()}"
# 冲突检测：同 key 重复注册 fail fast，避免静默覆盖改变运行时分派行为
existing = _RPC_REGISTRY[worker_type].get(registry_key)
if existing is not None:
if existing.func is func:
logger.debug("重复注册同一函数，跳过: %s", registry_key)
return func
raise ValueError(
f"RPC 端点冲突: '{registry_key}' 已注册为 "
f"{existing.func.__module__}.{existing.func.__qualname__}，"
f"当前又注册 {func.__module__}.{func.__qualname__}。"
f"同一 subject + 同一 HTTP 方法只能注册一次。"
)
_RPC_REGISTRY[worker_type][registry_key] = rpc
# 给函数附加元数据，方便外部访问
@@ -431,7 +448,10 @@ def auto_bind_models(plugin_name: str):
try:
models_module = importlib.import_module(f"src.plugins.{plugin_name}.models.pydantic")
except ImportError:
print(f"⚠️ auto_bind_models: 未找到 models 模块 (plugins.{plugin_name}.models.pydantic)")
logger.warning(
"auto_bind_models: 未找到 models 模块 (plugins.%s.models.pydantic)",
plugin_name,
)
return
# 获取模块中所有 BaseModel 子类
@@ -442,7 +462,7 @@ def auto_bind_models(plugin_name: str):
model_classes[name] = obj
if not model_classes:
print(f"⚠️ auto_bind_models: 未找到任何 Pydantic 模型")
logger.warning("auto_bind_models: 未找到任何 Pydantic 模型")
return
registry = _RPC_REGISTRY.get(plugin_name, {})
@@ -458,19 +478,19 @@ def auto_bind_models(plugin_name: str):
if req_model_name in model_classes:
rpc.request_model = model_classes[req_model_name]
bound_count += 1
print(f" ✓ 绑定请求模型: {req_model_name} -> {rpc.subject} [{rpc.method}]")
logger.debug("绑定请求模型: %s -> %s [%s]", req_model_name, rpc.subject, rpc.method)
# 尝试匹配响应模型: <Op><Method>Response
resp_model_name = f"{op_camel}{method_tag}Response"
if resp_model_name in model_classes:
rpc.response_model = model_classes[resp_model_name]
bound_count += 1
print(f" ✓ 绑定响应模型: {resp_model_name} -> {rpc.subject} [{rpc.method}]")
logger.debug("绑定响应模型: %s -> %s [%s]", resp_model_name, rpc.subject, rpc.method)
if bound_count > 0:
print(f"✅ auto_bind_models: 已绑定 {bound_count} 个模型到 {plugin_name} 的 RPC 端点")
logger.info("auto_bind_models: 已绑定 %d 个模型到 %s 的 RPC 端点", bound_count, plugin_name)
else:
print(f"⚠️ auto_bind_models: 未找到匹配的模型绑定")
logger.warning("auto_bind_models: 未找到匹配的模型绑定 (%s)", plugin_name)
def get_rpc_methods(worker_type: str) -> Dict[str, RPCMethod]:
src/chongming_worker/plugin.py
+40 -13
View File
@@ -81,6 +81,8 @@ class Plugin:
#: Database 实例（由 init_database 填充；类型标注为 Any 以避免强依赖 chongming_database）
self.db: Any = None
self._initialized = False
#: 初始化互斥锁——同一进程内并发 create_workers（如测试模拟多 worker 启动）串行化
self._init_lock = asyncio.Lock()
self._tasks: list[asyncio.Task] = []
self._cleanup_hooks: list[Callable[[], Awaitable[None]]] = []
self._vars: dict[str, Any] = {}
@@ -118,19 +120,29 @@ class Plugin:
count: Worker 副本数
worker_manager: WorkerManager 实例（可选）
"""
if not self._initialized:
self.config = config or {}
self.logger.info("[%s] 初始化开始 (count=%d)", self.type, count)
if self.auto_bind:
auto_bind_models(self.type)
await self._bootstrap()
await self.on_init()
self._initialized = True
self.logger.info("[%s] 初始化完成", self.type)
else:
self.logger.warning(
"[%s] create_workers 重复调用，跳过重复初始化", self.type
)
async with self._init_lock:
if not self._initialized:
self.config = config or {}
self.logger.info("[%s] 初始化开始 (count=%d)", self.type, count)
if self.auto_bind:
auto_bind_models(self.type)
await self._bootstrap()
await self.on_init()
self._initialized = True
self.logger.info("[%s] 初始化完成", self.type)
else:
self.logger.warning(
"[%s] create_workers 重复调用，跳过重复初始化", self.type
)
return self._create_workers_impl(session, config, count, worker_manager)
def _create_workers_impl(self, session, config, count=1, worker_manager=None):
"""
创建 Worker 副本任务（委托通用运行时）。
独立方法便于测试用桩替换（如 test_auth_race / verify_example_plugin），
避免在测试中依赖真实 zenoh 会话。
"""
return _create_workers(session, config, count, worker_manager)
async def cleanup(self, worker_item: str) -> None:
@@ -161,6 +173,21 @@ class Plugin:
if self.db is not None:
await self.db.dispose()
self.db = None
# 5. 清空通过 set_var 发布的全局变量，避免 global_var 继续持有已释放资源
for name in list(self._vars):
try:
gv = self._load_global_var()
if gv is not None:
setter = getattr(gv, "set_global_var", None)
if setter is not None:
setter(name, None)
else:
setattr(gv, name, None)
except Exception as e: # noqa: BLE001
self.logger.warning(
"[%s] 清理全局变量 %s 失败: %s", self.type, name, e
)
self._vars.clear()
finally:
self._initialized = False
self.logger.info("[%s] %s 清理完毕", self.type, worker_item)
src/chongming_worker/runtime.py
+39 -17
View File
@@ -6,6 +6,7 @@ Worker 运行时 — 通用的生命周期管理
import os
import asyncio
import json
import logging
import orjson
from typing import Any, Dict, List, Optional
@@ -16,6 +17,11 @@ from chongming_worker.decorators import (
get_rpc_methods,
)
logger = logging.getLogger("chongming_worker.runtime")
# 是否已提示过"非信封载荷"警告（每个进程仅提示一次，避免刷屏）
_warned_legacy_payload = False
async def create_worker(
session: zenoh.Session,
worker_type: str,
@@ -27,7 +33,7 @@ async def create_worker(
创建一个 Worker 协程
自动发现通过 @rpc_method 注册的端点，为每个 subject 声明 Zenoh Queryable，
收到请求后按 _method 字段分派到具体方法。
收到请求后按信封 meta.method 字段分派到具体方法。
Args:
session: Zenoh 会话
@@ -36,7 +42,7 @@ async def create_worker(
config: 自定义配置参数
worker_manager: WorkerManager 实例（用于本地注册）
"""
print(f"[Worker] {worker_id} 启动 (type={worker_type})")
logger.info("[Worker] %s 启动 (type=%s)", worker_id, worker_type)
# 1. 注册存活令牌
token = session.liveliness().declare_token(f"worker/{worker_type}/{worker_id}")
@@ -48,11 +54,11 @@ async def create_worker(
# 3. 获取该 worker 类型的所有 RPC 方法
rpc_methods = get_rpc_methods(worker_type)
if not rpc_methods:
print(f"[Worker] {worker_id} 警告: 未找到任何 RPC 方法")
logger.warning("[Worker] %s 警告: 未找到任何 RPC 方法", worker_id)
else:
print(f"[Worker] {worker_id} 发现 {len(rpc_methods)} 个端点:")
logger.info("[Worker] %s 发现 %d 个端点:", worker_id, len(rpc_methods))
for _key, rpc in rpc_methods.items():
print(f" {rpc.method} {rpc.subject} -> {rpc.name}({', '.join(rpc.params.keys())})")
logger.info(" %s %s -> %s(%s)", rpc.method, rpc.subject, rpc.name, ", ".join(rpc.params.keys()))
# 4. 按 subject 分组（多个方法可共享同一 subject，如 GET + POST）
subject_methods: Dict[str, List[RPCMethod]] = {}
@@ -67,11 +73,25 @@ async def create_worker(
loop = asyncio.get_running_loop()
async def _handle_query(query: zenoh.Query):
try:
# 解析请求负载
# payload_str = str(query.payload) if query.payload else "{}"
# 解析请求负载（信封协议: {"meta": {...}, "data": {...}}）
payload_bytes = bytes(query.payload) if query.payload else b""
data = orjson.loads(payload_bytes) if payload_bytes else {}
http_method = data.get("_method", "GET").upper()
raw = orjson.loads(payload_bytes) if payload_bytes else {}
if isinstance(raw, dict) and ("meta" in raw or "data" in raw):
meta = raw.get("meta") or {}
data = raw.get("data") or {}
else:
# 非信封载荷: 整包视为业务数据，method 默认 GET（无控制字段语义）
global _warned_legacy_payload
if not _warned_legacy_payload:
logger.warning(
"收到非信封格式载荷（缺少 meta/data），按 method=GET 处理: %s",
subject,
)
_warned_legacy_payload = True
meta = {}
data = raw if isinstance(raw, dict) else {}
http_method = str(meta.get("method", "GET")).upper()
# 匹配方法（按 HTTP 方法）
matched = next((m for m in methods if m.method.upper() == http_method), None)
@@ -85,10 +105,12 @@ async def create_worker(
return
# 调用 RPC 方法（支持同步和异步）
# user_info 从 meta 信封注入（框架参数，不进业务 payload）
result = await matched.call(
data,
session=session,
worker_id=worker_id
worker_id=worker_id,
user_info=meta.get("user_info"),
)
# 回复
@@ -96,7 +118,7 @@ async def create_worker(
key_expr=query.key_expr,
payload=orjson.dumps(result)
)
print(f"[Worker] {worker_id} 回复 {matched.method} {subject}: {result}")
logger.debug("[Worker] %s 回复 %s %s", worker_id, matched.method, subject)
except Exception as e:
error_response = {
"result": 0.0,
@@ -108,7 +130,7 @@ async def create_worker(
key_expr=query.key_expr,
payload=json.dumps(error_response).encode("utf-8")
)
print(f"[Worker] {worker_id} 异常 {subject}: {e}")
logger.error("[Worker] %s 异常 %s: %s", worker_id, subject, e)
def callback(query: zenoh.Query):
# 当前 create_worker 运行在事件循环中，可直接 create_task
@@ -123,16 +145,16 @@ async def create_worker(
# 使用回调方式声明
q = session.declare_queryable(key, make_callback(subject, methods))
queryables.append(q)
print(f"[Worker] {worker_id} Queryable: '{key}' [{', '.join(m.method for m in methods)}]")
logger.info("[Worker] %s Queryable: '%s' [%s]", worker_id, key, ", ".join(m.method for m in methods))
# 6. 保持 worker 存活直到被取消
stop_event = asyncio.Event()
try:
await stop_event.wait()
except asyncio.CancelledError:
print(f"[Worker] {worker_id} 正在关闭...")
logger.info("[Worker] %s 正在关闭...", worker_id)
except Exception as e:
print(f"[Worker] {worker_id} 发生异常: {e}")
logger.error("[Worker] %s 发生异常: %s", worker_id, e)
finally:
# 清理
if worker_manager is not None:
@@ -140,7 +162,7 @@ async def create_worker(
token.undeclare()
for q in queryables:
q.undeclare()
print(f"[Worker] {worker_id} 已关闭")
logger.info("[Worker] %s 已关闭", worker_id)
def create_workers(
session: zenoh.Session,
@@ -177,7 +199,7 @@ def create_workers(
)
)
tasks.append(task)
print(f"[Worker] 已创建任务: {worker_id} (type={worker_type})")
logger.info("[Worker] 已创建任务: %s (type=%s)", worker_id, worker_type)
return tasks