From 1ec0b2329cf3f7b67481dff6530ede898d831ddc Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Wed, 12 Aug 2026 10:19:05 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20=E6=8F=92=E4=BB=B6=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 77 +++++++++- src/chongming_worker/__init__.py | 4 + src/chongming_worker/plugin.py | 244 +++++++++++++++++++++++++++++++ 3 files changed, 324 insertions(+), 1 deletion(-) create mode 100644 src/chongming_worker/plugin.py diff --git a/README.md b/README.md index 36e390d..a8ab2f1 100644 --- a/README.md +++ b/README.md @@ -18,10 +18,77 @@ uv pip install -e utils/worker - **自动参数解析**：根据 HTTP 方法从 query 参数或 JSON body 解析请求 - **Pydantic 模型绑定**：`auto_bind_models()` 自动绑定生成的模型 - **Worker 运行时**：`create_worker` / `create_workers` 管理生命周期 +- **`Plugin` 基类**：统一插件初始化 / 清理生命周期（幂等、日志、数据库、后台任务、清理钩子） - **配置加载**：`load_worker_config` 读取 TOML 配置 --- +## Plugin 基类（推荐） + +`chongming_worker.plugin.Plugin` 统一了各插件 `__init__.py` 的初始化/清理风格， +消除重复样板（数据库初始化、global_var 写入、幂等保护、后台任务、日志、清理钩子）。 + +### 最小示例 + +```python +# 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 注册 +# from .routes import router # 可选：自定义 FastAPI 路由（网关自动发现） + +logger = logging.getLogger(__name__) + +auto_bind_models("my_plugin") + + +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/`（必填） | +| `db_name` | 默认 SQLite 文件名（放 `data_dir` 下）；`None` 表示不自动建库 | +| `auto_bind` | 首次初始化时是否自动 `auto_bind_models(self.type)` | +| `on_init()` | 首次 `create_workers` 时调用一次，初始化资源 | +| `on_cleanup()` | `cleanup` 时调用，清理自定义资源 | + +### 框架自动处理 + +- **幂等**：重复 `create_workers` 只初始化一次（旧写法需手写 `_initialized` 标志） +- **统一日志**：`logger` 自动带 `[type]` 前缀 +- **数据库**：`init_database()` 统一建路径、建表、记录，`cleanup` 自动 `dispose()` +- **global_var**：`set_var()` / `get_var()` 统一读写插件 `global_var` 模块 +- **后台任务**：`spawn_task()` 跟踪，`cleanup` 时自动取消并 `gather` +- **清理钩子**：`register_cleanup()` 注册，`cleanup` 时逆序执行 +- **委托**：自动调用 `runtime.create_workers` 创建 Worker 副本 + +--- + ## API ### `@rpc_method` 装饰器 @@ -95,6 +162,8 @@ cfg = load_worker_config("example") ## 插件集成模式 +> 推荐使用上文的 **Plugin 基类**；以下为不依赖基类的传统写法（仍受支持）。 + ```python # src/plugins/my_plugin/__init__.py from chongming_worker import auto_bind_models @@ -103,15 +172,20 @@ 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_manager): @@ -132,6 +206,7 @@ from chongming_worker import ( load_worker_config, # TOML 配置加载 WorkerConfig, OperationConfig, # 配置类型 create_worker, create_workers, # 运行时 + Plugin, # 统一生命周期基类 ) ``` diff --git a/src/chongming_worker/__init__.py b/src/chongming_worker/__init__.py index 3f2aca2..2ab7cc7 100644 --- a/src/chongming_worker/__init__.py +++ b/src/chongming_worker/__init__.py @@ -22,6 +22,9 @@ from chongming_worker.runtime import ( create_worker, create_workers, ) +from chongming_worker.plugin import ( + Plugin, +) __all__ = [ "RPCMethod", @@ -37,4 +40,5 @@ __all__ = [ "OperationConfig", "create_worker", "create_workers", + "Plugin", ] diff --git a/src/chongming_worker/plugin.py b/src/chongming_worker/plugin.py new file mode 100644 index 0000000..84661c0 --- /dev/null +++ b/src/chongming_worker/plugin.py @@ -0,0 +1,244 @@ +""" +Plugin 基类 — 统一 Worker 插件初始化与清理生命周期 + +目标：消除各插件 ``__init__.py`` 中重复且风格不一的样板代码 +（数据库初始化、global_var 写入、幂等保护、后台任务、日志、清理钩子）， +让新插件只需覆写 ``on_init`` / ``on_cleanup`` 两个钩子即可。 + +典型用法（src/plugins//__init__.py）:: + + import logging + + from chongming_worker.plugin import Plugin + + from . import router # 触发 @rpc_method 注册（等价 from .router import *） + # from .routes import router # 可选：自定义 FastAPI 路由（网关自动发现） + + logger = logging.getLogger(__name__) + + class MyPlugin(Plugin): + type = "myname" + db_name = "myname.db" # 自动在 data_dir 下建库；不需要则置 None + auto_bind = True # 首次初始化时自动 auto_bind_models("myname") + + async def on_init(self): + # 数据库已由基类 _bootstrap 初始化（self.db） + self.set_var("db", self.db) + # 需要后台任务时: + # self.spawn_task(self._background_loop()) + + async def on_cleanup(self): + # 关闭自定义客户端等（后台任务与数据库由基类统一处理） + ... + + # 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法一致） + plugin = MyPlugin() + create_workers = plugin.create_workers + cleanup = plugin.cleanup + +框架自动处理: + - 幂等：重复 ``create_workers`` 只会初始化一次 + - 统一日志：``logger`` 自动带 ``[type]`` 前缀 + - 数据库：``_bootstrap()`` / ``init_database()`` 统一建路径、建表、记录，``cleanup`` 自动 dispose + - global_var：``set_var()`` 写入插件自己的 ``global_var`` 模块（白名单由插件决定） + - 后台任务：``spawn_task()`` 跟踪，``cleanup`` 时自动取消 + - 清理钩子：``register_cleanup()`` 注册，``cleanup`` 时逆序执行 + - 委托 ``chongming_worker.runtime.create_workers`` 创建 Worker 副本 +""" +from __future__ import annotations + +import asyncio +import importlib +import logging +from pathlib import Path +from typing import Any, Awaitable, Callable + +from chongming_worker.decorators import auto_bind_models +from chongming_worker.runtime import create_workers as _create_workers + + +class Plugin: + """ + 插件统一生命周期基类。 + + 子类覆写点: + - ``type``: 插件类型（worker type，必填，匹配 subject 前缀与 config/workers/） + - ``db_name``: 默认 SQLite 文件名（放 data_dir 下），None 表示不自动建库 + - ``auto_bind``: 首次初始化时是否自动 ``auto_bind_models(self.type)`` + - ``on_init()`` / ``on_cleanup()``: 初始化 / 清理钩子 + """ + + #: 插件类型（worker type），必须被子类覆盖 + type: str = "default" + #: 默认数据库文件名（data_dir 下）；None 表示不自动建库 + db_name: str | None = None + #: 首次初始化时自动调用 auto_bind_models + auto_bind: bool = True + + def __init__(self) -> None: + self.logger = logging.getLogger(f"plugins.{self.type}") + self.config: dict = {} + #: Database 实例（由 init_database 填充；类型标注为 Any 以避免强依赖 chongming_database） + self.db: Any = None + self._initialized = False + self._tasks: list[asyncio.Task] = [] + self._cleanup_hooks: list[Callable[[], Awaitable[None]]] = [] + self._vars: dict[str, Any] = {} + self._gv_module: Any = None + self._gv_probed = False + + # ------------------------------------------------------------------ + # 子类可覆写钩子 + # ------------------------------------------------------------------ + async def _bootstrap(self) -> None: + """ + 默认资源引导：按 db_name 初始化数据库。 + + 由基类在 on_init 之前自动调用；子类需要完全自定义建库逻辑时覆写此方法。 + """ + if self.db_name: + await self.init_database(self.db_name) + + async def on_init(self) -> None: + """首次 create_workers 时调用一次，初始化业务资源（数据库已由 _bootstrap 引导）。""" + + async def on_cleanup(self) -> None: + """cleanup 时调用，清理自定义资源。""" + + # ------------------------------------------------------------------ + # 对外统一接口（app lifespan 直接绑定这两个 bound method） + # ------------------------------------------------------------------ + async def create_workers(self, session, config, count=1, worker_manager=None): + """ + 插件入口 — 由 app lifespan 调用，返回 Worker 任务列表。 + + Args: + session: Zenoh 会话 + config: Worker 配置字典（含 [workers.config] 展平后的运行时配置） + 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 + ) + return _create_workers(session, config, count, worker_manager) + + async def cleanup(self, worker_item: str) -> None: + """ + 插件清理 — 由 app shutdown 调用。 + + 依次: 取消后台任务 → 逆序执行注册的清理钩子 → on_cleanup() → 关闭数据库。 + """ + self.logger.info("[%s] 清理开始: %s", self.type, worker_item) + try: + # 1. 取消后台任务 + if self._tasks: + for t in self._tasks: + if not t.done(): + t.cancel() + await asyncio.gather(*self._tasks, return_exceptions=True) + self._tasks.clear() + # 2. 逆序执行清理钩子（后注册的先清理，符合资源释放直觉） + for hook in reversed(self._cleanup_hooks): + try: + await hook() + except Exception as e: # noqa: BLE001 + self.logger.error("[%s] 清理钩子异常: %s", self.type, e) + self._cleanup_hooks.clear() + # 3. 子类自定义清理 + await self.on_cleanup() + # 4. 关闭数据库 + if self.db is not None: + await self.db.dispose() + self.db = None + finally: + self._initialized = False + self.logger.info("[%s] %s 清理完毕", self.type, worker_item) + + # ------------------------------------------------------------------ + # 辅助方法 + # ------------------------------------------------------------------ + async def init_database(self, filename: str | None = None) -> Any: + """ + 在 data_dir 下初始化 SQLite 数据库，返回 Database 实例并存入 self.db。 + + Args: + filename: 数据库文件名（默认取 self.db_name） + """ + from chongming_database import Database + + filename = filename or self.db_name + if not filename: + raise ValueError("未指定数据库文件名（请设置 db_name 或传入 filename）") + db_path = Path(self.config.get("data_dir", "./data")) / filename + db = Database(db_path) + await db.initialize() + self.db = db + self.logger.info("[%s] 数据库已初始化: %s", self.type, db_path) + return db + + def set_var(self, name: str, value: Any) -> None: + """ + 写入运行时变量，并同步到插件 global_var 模块。 + + - 若模块存在 ``set_global_var`` 函数则调用之（白名单由插件自己决定） + - 否则直接写模块属性（兼容无 setter 的模块） + """ + self._vars[name] = value + gv = self._load_global_var() + if gv is not None: + setter = getattr(gv, "set_global_var", None) + if setter is not None: + setter(name, value) + else: + setattr(gv, name, value) + + def get_var(self, name: str, default: Any = None) -> Any: + """读取运行时变量：优先 global_var 模块属性，其次内部字典。""" + gv = self._load_global_var() + if gv is not None and hasattr(gv, name): + return getattr(gv, name) + return self._vars.get(name, default) + + def register_cleanup(self, hook: Callable[[], Awaitable[None]]) -> None: + """ + 注册异步清理钩子，cleanup 时逆序执行。 + + 适合在 on_init 中注册"关闭客户端"这类逻辑，避免在 on_cleanup 里堆叠 try/except。 + """ + self._cleanup_hooks.append(hook) + + def spawn_task(self, coro: Awaitable[Any]) -> asyncio.Task: + """创建后台任务并跟踪，cleanup 时自动取消。""" + task = asyncio.create_task(coro) # type: ignore[arg-type] + self._tasks.append(task) + return task + + def _load_global_var(self) -> Any: + """懒加载插件 global_var 模块（不存在则缓存 None）。""" + if self._gv_probed: + return self._gv_module + self._gv_probed = True + for mod in ( + f"plugins.{self.type}.global_var", + f"src.plugins.{self.type}.global_var", + ): + try: + self._gv_module = importlib.import_module(mod) + return self._gv_module + except ImportError: + continue + return None + + +__all__ = ["Plugin"]