377 lines
15 KiB
Markdown
377 lines
15 KiB
Markdown
# Worker 插件开发指南
|
||
|
||
每个 `src/plugins/<name>/` 子目录是一个独立的 Worker 插件，封装特定领域的业务逻辑。插件通过 Zenoh RPC 对外提供服务，HTTP 路由由框架根据 TOML 配置自动生成。
|
||
|
||
---
|
||
|
||
## 目录结构约定
|
||
|
||
```text
|
||
<plugin_name>/
|
||
├── __init__.py # 入口：Plugin 子类 + create_workers/cleanup 绑定
|
||
├── global_var.py # 模块级全局状态（Database、API client 等）
|
||
├── router/ # RPC 端点实现
|
||
│ ├── __init__.py # 导入所有端点模块（`from . import *` 或显式导入）
|
||
│ └── <操作名>.py # 每个端点一个文件，使用 @rpc_method 装饰器
|
||
├── models/ # 数据模型
|
||
│ ├── __init__.py # 模型导出（由 gen_models 自动生成，勿手动修改）
|
||
│ ├── pydantic.py # Pydantic 请求/响应模型（由 gen_models 自动生成）
|
||
│ └── <表名>.py # SQLModel ORM 模型（如需要数据库）
|
||
├── routes.py # 自定义 FastAPI 路由（可选，用于特殊端点）
|
||
└── README.md # 插件说明文档（可选）
|
||
```
|
||
|
||
---
|
||
|
||
## 已有插件概览
|
||
|
||
| 插件 | 特点 | 适合参考 |
|
||
| ------------------------ | -------------------------------------------- | ---------- |
|
||
| **example** | 最简示例，Plugin 基类 + Database + 生命周期 | 入门学习 |
|
||
| **auth** | SQLModel ORM + 管理员初始化 + 多进程幂等 | 数据库操作 |
|
||
| **doubao** | 外部 API 客户端（Ark/OpenAI）+ AICC 安全通道 | 外部集成 |
|
||
| **aibox_sophon_adapter** | 自定义 `APIRouter` + MQTT 客户端组件（`mqtt_clients.py`）+ 后台任务 | 高级模式 |
|
||
|
||
---
|
||
|
||
## 开发新插件：完整流程
|
||
|
||
### 第一步：创建目录结构
|
||
|
||
> **推荐**：直接用脚手架生成统一风格的插件骨架，无需手动创建目录：
|
||
>
|
||
> ```bash
|
||
> uv run gen_plugin <name> [operation ...]
|
||
> ```
|
||
>
|
||
> 以下为手动创建方式（了解结构时参考）。
|
||
|
||
```bash
|
||
mkdir -p src/plugins/<name>/router
|
||
mkdir -p src/plugins/<name>/models
|
||
touch src/plugins/<name>/__init__.py
|
||
touch src/plugins/<name>/global_var.py
|
||
touch src/plugins/<name>/router/__init__.py
|
||
```
|
||
|
||
### 第二步：注册插件
|
||
|
||
在 `config/workers/<name>/config.toml` 中通过 `[workers]` 节注册（网关启动时自动扫描发现）：
|
||
|
||
```toml
|
||
# config/workers/<name>/config.toml
|
||
[workers]
|
||
type = "<name>" # 插件类型（worker type）
|
||
module = "plugins.<name>" # Python 模块路径
|
||
factory = "create_workers" # 工厂函数名（Plugin 基类绑定的方法）
|
||
cleanup = "cleanup" # 清理函数名
|
||
count = 3 # Worker 副本数
|
||
|
||
# 运行时配置（展平后传给 create_workers 的 config 参数）
|
||
[workers.config]
|
||
data_dir = "./data"
|
||
```
|
||
|
||
> 旧式全局注册文件 `config/workers/config.toml`（`[[workers]]` 数组）仅作兜底兼容，新插件请使用各插件自己的 `config.toml`。
|
||
|
||
### 第三步：编写端点 TOML 配置
|
||
|
||
创建 `config/workers/<name>/config.toml`（分组配置）：
|
||
|
||
```toml
|
||
[worker]
|
||
defaults = { timeout = 3.0, auth_required = true, cacheable = false, internal = false }
|
||
|
||
[worker.registration]
|
||
router_prefix = "/<prefix>"
|
||
tags = ["<name>"]
|
||
```
|
||
|
||
创建每个操作的 `.toml` 文件，如 `config/workers/<name>/my_action.toml`：
|
||
|
||
```toml
|
||
[POST]
|
||
summary = "我的操作"
|
||
auth_required = true
|
||
|
||
[POST.response_model]
|
||
result = { type = "str", required = true }
|
||
message = { type = "str", default = "" }
|
||
```
|
||
|
||
### 第四步：生成 Pydantic 模型
|
||
|
||
```bash
|
||
uv run python -m chongming.scripts.gen_models
|
||
```
|
||
|
||
自动在 `src/plugins/<name>/models/` 下生成 `pydantic.py` 和 `__init__.py`。
|
||
|
||
### 第五步：实现 RPC 端点
|
||
|
||
`src/plugins/<name>/router/my_action.py`：
|
||
|
||
```python
|
||
from chongming_worker.decorators import rpc_method
|
||
from ..models.pydantic import MyActionPostRequest, MyActionPostResponse
|
||
|
||
@rpc_method(
|
||
subject="<name>.my_action",
|
||
method="POST",
|
||
response_model=MyActionPostResponse,
|
||
)
|
||
async def my_action(request: MyActionPostRequest) -> MyActionPostResponse:
|
||
"""业务逻辑"""
|
||
result = await do_something(request)
|
||
return MyActionPostResponse(result="success", message="done")
|
||
```
|
||
|
||
### 第六步：更新 router 导出
|
||
|
||
`src/plugins/<name>/router/__init__.py`：
|
||
|
||
```python
|
||
from . import my_action
|
||
__all__ = ["my_action"]
|
||
```
|
||
|
||
### 第七步：实现插件入口（推荐 Plugin 基类）
|
||
|
||
`src/plugins/<name>/__init__.py`（由 `gen_plugin` 脚手架生成）：
|
||
|
||
```python
|
||
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 = "<name>"
|
||
db_name = "<name>.db" # 自动在 data_dir 下建 SQLite 库（不需要则置 None）
|
||
auto_bind = True # 首次初始化时自动 auto_bind_models("<name>")
|
||
|
||
async def on_init(self):
|
||
"""首次 create_workers 时调用一次：初始化资源、启动后台任务。"""
|
||
self.set_var("db", self.db) # 发布到 global_var 供业务代码读取
|
||
# self.spawn_task(self._background_loop())
|
||
|
||
async def on_cleanup(self):
|
||
"""cleanup 时调用：关闭自定义客户端等（后台任务与数据库由基类处理）。"""
|
||
|
||
|
||
# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致）
|
||
_plugin = MyPlugin()
|
||
create_workers = _plugin.create_workers
|
||
cleanup = _plugin.cleanup
|
||
```
|
||
|
||
基类自动处理：幂等（初始化锁）、统一日志、数据库建库/释放、后台任务取消、清理钩子、模型自动绑定、cleanup 时重置已发布的 global_var。详见 `utils/worker/README.md` 的「Plugin 基类」章节。
|
||
|
||
> **旧写法已废弃**：不依赖基类的传统 `create_workers` / `cleanup` 模块级函数写法仅框架层面保留兼容，现有插件（auth / doubao / aibox_sophon_adapter / example）均已迁移到 Plugin 基类，新插件请勿采用旧写法。
|
||
|
||
---
|
||
|
||
## 自定义 FastAPI 路由（高级）
|
||
|
||
当端点无法用 TOML 描述时（如文件上传），插件可暴露自定义 `APIRouter`。
|
||
|
||
`src/plugins/<name>/routes.py`：
|
||
|
||
```python
|
||
from fastapi import APIRouter, UploadFile, File, Form
|
||
|
||
router = APIRouter(prefix="/<prefix>", tags=["<name>"])
|
||
|
||
@router.post("/upload")
|
||
async def upload(
|
||
file: UploadFile = File(...),
|
||
description: str = Form(...),
|
||
):
|
||
# 处理文件上传
|
||
return {"status": "ok"}
|
||
```
|
||
|
||
在 `__init__.py` 中暴露 `router`：
|
||
|
||
```python
|
||
from .routes import router # 网关通过 getattr 自动发现
|
||
```
|
||
|
||
网关在 `load_workers()` 阶段自动检测并注册该 router，无需修改网关代码。
|
||
|
||
---
|
||
|
||
## 为现有插件添加新端点
|
||
|
||
1. 创建 `config/workers/<name>/<new_op>.toml`
|
||
2. 运行 `uv run python -m chongming.scripts.gen_models`
|
||
3. 创建 `router/<new_op>.py`，引用生成的模型
|
||
4. 更新 `router/__init__.py`
|
||
5. 重启服务
|
||
|
||
---
|
||
|
||
## 共享对象（global_var）在两种路由中的使用
|
||
|
||
### 数据流
|
||
|
||
共享对象（Database、API client 等）在 `Plugin.on_init` 中初始化，通过 `set_var` 发布到 `global_var` 模块；两种路由的业务代码（模块级函数）在函数体内 `from .global_var import ...` 读取。
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
P["Plugin.on_init<br/>self.db = ..."] -->|"self.set_var('db', self.db)"| GV["global_var 模块<br/>db / require()"]
|
||
R1["router/*.py<br/>@rpc_method 函数"] -->|"from ..global_var import require"| GV
|
||
R2["routes.py<br/>APIRouter 处理器"] -->|"from .global_var import db"| GV
|
||
```
|
||
|
||
> **为什么需要 `global_var.py`**：`@rpc_method` 函数和 APIRouter 处理器都是模块级函数，拿不到 Plugin 实例，只能通过模块级容器读取；`global_var.py` 独立成模块，可避免与 `__init__.py` 循环导入。`set_var()` 内部会调用 `global_var.set_global_var()`（白名单校验）完成发布。
|
||
|
||
### 注入：on_init → global_var
|
||
|
||
`src/plugins/<name>/__init__.py`：
|
||
|
||
```python
|
||
class MyPlugin(Plugin):
|
||
type = "<name>"
|
||
db_name = "<name>.db"
|
||
|
||
async def on_init(self):
|
||
self.set_var("db", self.db) # 发布到 global_var 模块
|
||
```
|
||
|
||
### 路由一：TOML 驱动标准路由（@rpc_method）
|
||
|
||
`router/<op>.py`（请求经 Zenoh → Worker Queryable → 调用函数）：
|
||
|
||
```python
|
||
from chongming_worker.decorators import rpc_method
|
||
from ..models.pydantic import OpGetRequest, OpGetResponse
|
||
|
||
@rpc_method(subject="<name>.op", method="GET", response_model=OpGetResponse)
|
||
async def op(request: OpGetRequest) -> OpGetResponse:
|
||
from ..global_var import require
|
||
db = require("db") # 未初始化立即抛清晰错误
|
||
async with db.get_session() as session:
|
||
...
|
||
return OpGetResponse(...)
|
||
```
|
||
|
||
### 路由二：插件自定义路由（APIRouter）
|
||
|
||
`routes.py`（FastAPI 直接调用，适合 multipart 上传等 TOML 无法描述的端点）：
|
||
|
||
```python
|
||
from fastapi import APIRouter, UploadFile, File
|
||
|
||
router = APIRouter(prefix="/<name>", tags=["<name>"])
|
||
|
||
@router.post("/upload")
|
||
async def upload(file: UploadFile = File(...)):
|
||
from .global_var import db
|
||
if db is None: # 容忍未初始化，优雅降级
|
||
return {"error": "db 未初始化"}
|
||
async with db.get_session() as session:
|
||
...
|
||
return {"ok": True}
|
||
```
|
||
|
||
在 `__init__.py` 顶部暴露给网关：`from .routes import router`（网关通过 `getattr(module, "router")` 自动发现）。
|
||
|
||
### 两种路由对比
|
||
|
||
| | 路由一 `@rpc_method` | 路由二 `APIRouter` |
|
||
| ---------- | ----------------------------------------- | --------------------------------------------- |
|
||
| 文件位置 | `router/<op>.py` | `routes.py` |
|
||
| 入口发现 | 装饰器注册 → Worker Queryable | `__init__.py` 暴露 `router` 属性 |
|
||
| 函数签名 | `(request: Pydantic模型) -> Pydantic模型` | FastAPI 风格（`Request`/`UploadFile`/`Form`） |
|
||
| 适用场景 | 标准 JSON / query 端点 | multipart 上传、手动解析报文 |
|
||
| 读共享对象 | `from ..global_var import require` | `from .global_var import db` |
|
||
| 请求路径 | HTTP → Zenoh → Worker 回调 | FastAPI 直接调用 |
|
||
|
||
### 经验
|
||
|
||
- **在函数体内 import**（不要放文件顶部）：模块 import 时 `create_workers` 尚未执行、`global_var` 还是空；请求到达时变量才注入完成。
|
||
- **两种读取风格**：`require("db")`（必须有，未初始化立即报错）vs `from .global_var import db` + `if db is None`（允许优雅降级）。
|
||
- **参考实现**：`aibox_sophon_adapter/routes.py` 的 `_save_alarm_record`（自定义路由中 `from .global_var import db` + `async with db.get_session()`）。
|
||
|
||
---
|
||
|
||
## @rpc_method 装饰器
|
||
|
||
```python
|
||
@rpc_method(
|
||
subject="<plugin>.<operation>", # Zenoh 主题（必需）
|
||
method="GET", # HTTP 方法：GET / POST / MQTT
|
||
response_model=ResponseCls, # Pydantic 响应模型类
|
||
)
|
||
async def handler(request: RequestCls) -> ResponseCls:
|
||
...
|
||
```
|
||
|
||
| method | 参数来源 | 请求绑定 |
|
||
| ------ | ------------------ | ---------------------------- |
|
||
| `GET` | URL query string | TOML `params` 定义 |
|
||
| `POST` | JSON body | 自动绑定到 Pydantic 请求模型 |
|
||
| `MQTT` | MQTT payload JSON | 同 POST |
|
||
|
||
**请求信封协议**：网关统一把请求组装为 `{"meta": {"method", "user_info"}, "data": {...}}`
|
||
发给 Worker；运行时按 `meta.method` 分派到对应 `@rpc_method`，`data` 经请求模型校验后
|
||
传给处理器。MQTT 客户端发布消息时需按信封格式（见主 README「请求负载格式」），
|
||
MQTT 专用端点要显式 `"meta": {"method": "MQTT"}`。
|
||
|
||
- 设置 `internal = true` 后，端点不注册 HTTP 路由，仅可通过 Zenoh RPC 或 MQTT 内部调用
|
||
- MQTT 方法返回 `None` 时，Bridge 自动跳过响应发布
|
||
|
||
---
|
||
|
||
## 启动契约校验
|
||
|
||
服务启动时（`load_workers` 之后、注册路由之前），网关自动核对三方一致性：
|
||
|
||
| 检查项 | 策略 |
|
||
| ---------------------------- | --------------------------- |
|
||
| TOML HTTP 路由缺 `@rpc_method` | **硬失败**，启动直接终止 |
|
||
| `[MQTT]` / `internal` 路由缺注册 | 警告日志 |
|
||
| 已注册方法无 TOML 路由 | 警告日志（反向漂移检测） |
|
||
| params/response_model 缺生成模型 | 警告日志（gen_models 过期） |
|
||
| 生成模型未被任何路由引用 | 警告日志（孤儿模型） |
|
||
|
||
新增端点后如果启动报错「缺少对应的 @rpc_method 注册」，检查 TOML 的 `subject`/方法节
|
||
与 `router/<op>.py` 中的 `@rpc_method(subject=..., method=...)` 是否一致。
|
||
|
||
---
|
||
|
||
## 命名约定
|
||
|
||
| 概念 | 约定 | 示例 |
|
||
| ------------- | ------------------------------ | ---------------------------- |
|
||
| 插件名 | 小写下划线 | `aibox_sophon_adapter` |
|
||
| 操作名 | 小写下划线 | `board_ping`, `list_devices` |
|
||
| Zenoh subject | `<插件>.<操作>` | `example.add` |
|
||
| TOML 文件名 | 与操作名一致 | `add.toml` |
|
||
| Python 文件名 | 与操作名一致 | `add.py` |
|
||
| 模型类名 | `<操作><方法>Request/Response` | `AddGetRequest` |
|
||
|
||
| 概念 | 约定 | 示例 |
|
||
| ----------------- | --------------- | ----------------------------------- |
|
||
| 插件名 | 小写下划线 | `doubao`, `auth`, `example` |
|
||
| 操作名 | 小写下划线 | `analyze`, `add`, `change_password` |
|
||
| Zenoh subject | `<插件>.<操作>` | `example.add`, `doubao.analyze` |
|
||
| TOML 文件名 | 与操作名一致 | `analyze.toml`, `add.toml` |
|
||
| Python 路由文件名 | 与操作名一致 | `analyze.py`, `add.py` |
|
||
|
||
---
|
||
|
||
## 现有插件
|
||
|
||
| 插件 | 说明 | 文档 |
|
||
| ---------------------------- | -------------------------------------- | ---------------- |
|
||
| [auth](auth/README.md) | 用户注册、登录、JWT 鉴权、用户信息管理 | 含 SQLite 数据库 |
|
||
| [doubao](doubao/README.md) | 火山引擎豆包视觉理解模型集成 | 第三方 API 调用 |
|
||
| [example](example/README.md) | 四则运算示例 + 访问计数器 | 入门参考 |
|