modify: 插件初始化
This commit is contained in:
@@ -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",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,244 @@
|
||||
"""
|
||||
Plugin 基类 — 统一 Worker 插件初始化与清理生命周期
|
||||
|
||||
目标：消除各插件 ``__init__.py`` 中重复且风格不一的样板代码
|
||||
（数据库初始化、global_var 写入、幂等保护、后台任务、日志、清理钩子），
|
||||
让新插件只需覆写 ``on_init`` / ``on_cleanup`` 两个钩子即可。
|
||||
|
||||
典型用法（src/plugins/<name>/__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/<type>）
|
||||
- ``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"]
|
||||
Reference in New Issue
Block a user