diff --git a/README.md b/README.md index 54687c1..4b23a2a 100644 --- a/README.md +++ b/README.md @@ -1,38 +1,49 @@ # Worker 插件开发指南 -每个子目录代表一个独立的 Worker 插件，封装特定领域的业务逻辑。插件通过 Zenoh RPC 对外提供服务， -HTTP 路由由框架根据 TOML 配置自动生成。 +每个 `src/plugins//` 子目录是一个独立的 Worker 插件，封装特定领域的业务逻辑。插件通过 Zenoh RPC 对外提供服务，HTTP 路由由框架根据 TOML 配置自动生成。 --- ## 目录结构约定 ```text -plugin_name/ -├── __init__.py # 入口：create_workers() 工厂 + cleanup() -├── global_var.py # 模块级全局状态（数据库连接、API 客户端等） +/ +├── __init__.py # 入口：create_workers() 工厂 + cleanup() + auto_bind_models() +├── global_var.py # 模块级全局状态（Database、API client 等） ├── router/ # RPC 端点实现 -│ ├── __init__.py # 导入所有端点模块，__all__ 导出 +│ ├── __init__.py # 导入所有端点模块（`from . import *` 或显式导入） │ └── <操作名>.py # 每个端点一个文件，使用 @rpc_method 装饰器 ├── models/ # 数据模型 -│ ├── __init__.py # 模型导出（自动生成，勿手动修改） -│ ├── pydantic.py # Pydantic 请求/响应模型（由 gen_models.py 自动生成） +│ ├── __init__.py # 模型导出（由 gen_models 自动生成，勿手动修改） +│ ├── pydantic.py # Pydantic 请求/响应模型（由 gen_models 自动生成） │ └── <表名>.py # SQLModel ORM 模型（如需要数据库） -└── README.md # 插件说明文档 +├── routes.py # 自定义 FastAPI 路由（可选，用于特殊端点） +└── README.md # 插件说明文档（可选） ``` --- +## 已有插件概览 + +| 插件 | 特点 | 适合参考 | +| ------------------------ | ------------------------------------------------- | ---------- | +| **example** | 最简示例，`auto_bind_models` + Database + cleanup | 入门学习 | +| **auth** | SQLModel ORM + 管理员初始化 + 多进程幂等 | 数据库操作 | +| **doubao** | 外部 API 客户端（Ark/OpenAI）+ AICC 安全通道 | 外部集成 | +| **aibox_sophon_adapter** | 自定义 `APIRouter` + MQTT 转发 + 后台任务 | 高级模式 | + +--- + ## 开发新插件：完整流程 ### 第一步：创建目录结构 ```bash -mkdir -p src/plugins//router -mkdir -p src/plugins//models -touch src/plugins//__init__.py -touch src/plugins//global_var.py -touch src/plugins//router/__init__.py +mkdir -p src/plugins//router +mkdir -p src/plugins//models +touch src/plugins//__init__.py +touch src/plugins//global_var.py +touch src/plugins//router/__init__.py ``` ### 第二步：注册插件 @@ -41,20 +52,20 @@ touch src/plugins//router/__init__.py ```toml [[workers]] -type = "" -module = "plugins." +type = "" +module = "plugins." factory = "create_workers" cleanup = "cleanup" count = 3 [workers.config] -type = "" -# 自定义配置项... +type = "" +data_dir = "./data" ``` ### 第三步：编写端点 TOML 配置 -创建 `config/workers//config.toml`（分组配置）： +创建 `config/workers//config.toml`（分组配置）： ```toml [worker] @@ -62,9 +73,10 @@ defaults = { timeout = 3.0, auth_required = true, cacheable = false, internal = [worker.registration] router_prefix = "/" +tags = [""] ``` -创建每个操作的 `.toml` 文件，如 `config/workers//my_action.toml`： +创建每个操作的 `.toml` 文件，如 `config/workers//my_action.toml`： ```toml [POST] @@ -82,30 +94,30 @@ message = { type = "str", default = "" } uv run python -m chongming.scripts.gen_models ``` -这会自动在 `src/plugins//models/` 下生成 `pydantic.py` 和更新 `__init__.py`。 +自动在 `src/plugins//models/` 下生成 `pydantic.py` 和 `__init__.py`。 ### 第五步：实现 RPC 端点 -`src/plugins//router/my_action.py`： +`src/plugins//router/my_action.py`： ```python from chongming_worker.decorators import rpc_method from ..models.pydantic import MyActionPostRequest, MyActionPostResponse @rpc_method( - subject=".my_action", + subject=".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//router/__init__.py`： +`src/plugins//router/__init__.py`： ```python from . import my_action @@ -114,23 +126,27 @@ __all__ = ["my_action"] ### 第七步：实现工厂函数 -`src/plugins//__init__.py`： +`src/plugins//__init__.py`： ```python +from pathlib import Path +from chongming_worker import auto_bind_models from chongming_worker.runtime import create_workers as _create_workers +from .router import * # 触发 @rpc_method 注册 -def create_workers(session, config, count, worker_manager): - """插件入口：初始化全局资源，启动 Worker""" - from . import router # 触发 @rpc_method 注册 +auto_bind_models("") # 自动绑定生成的 Pydantic 模型 + +async def create_workers(session, config, count, worker_manager): + """插件入口：初始化资源，启动 Worker""" + from chongming_database import Database from .global_var import set_db - # 初始化数据库、客户端等 - from chongming_database import Database - db = Database(config["data_dir"] / "mydb.db") - # await db.initialize() -- 在异步环境中 - + db_path = Path(config.get("data_dir", "./data")) / ".db" + db = Database(db_path) + await db.initialize() set_db(db) - _create_workers(session, config, count, worker_manager) + + return _create_workers(session, config, count, worker_manager) async def cleanup(worker_manager): """清理资源""" @@ -139,13 +155,45 @@ async def cleanup(worker_manager): await db.dispose() ``` +> **注意**：`create_workers` 必须是 `async` 函数，因为 `Database.initialize()` 是异步的。框架在 `asyncio.gather` 中并发调用所有插件的工厂函数。 + +--- + +## 自定义 FastAPI 路由（高级） + +当端点无法用 TOML 描述时（如文件上传），插件可暴露自定义 `APIRouter`。 + +`src/plugins//routes.py`： + +```python +from fastapi import APIRouter, UploadFile, File, Form + +router = APIRouter(prefix="/", tags=[""]) + +@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//` 下创建 `<新操作>.toml` +1. 创建 `config/workers//.toml` 2. 运行 `uv run python -m chongming.scripts.gen_models` -3. 在 `router/` 下创建 `<新操作>.py`，引用生成的模型 +3. 创建 `router/.py`，引用生成的模型 4. 更新 `router/__init__.py` 5. 重启服务 @@ -153,13 +201,15 @@ async def cleanup(worker_manager): ## 全局变量模式 -每个插件通过 `global_var.py` 管理模块级共享资源，避免跨请求的状态冲突： +每个插件通过 `global_var.py` 管理模块级共享资源： ```python # global_var.py -db: "Database | None" = None +from chongming_database import Database -def set_db(database: "Database"): +db: Database | None = None + +def set_db(database: Database): global db db = database ``` @@ -169,35 +219,48 @@ def set_db(database: "Database"): ```python from ..global_var import db -@rpc_method(subject="plugin.action", method="GET", ...) -async def handler(request: RequestModel): +@rpc_method(subject=".action", method="GET", ...) +async def handler(request): async with db.get_session() as session: - # 执行数据库操作 ... ``` --- -## 装饰器说明 +## @rpc_method 装饰器 ```python @rpc_method( subject=".", # Zenoh 主题（必需） - method="GET", # HTTP 方法：GET 或 POST + method="GET", # HTTP 方法：GET / POST / MQTT response_model=ResponseCls, # Pydantic 响应模型类 ) async def handler(request: RequestCls) -> ResponseCls: ... ``` -- **GET 端点**：参数通过 URL query string 传递，在 TOML 中用 `params` 定义 -- **POST 端点**：参数通过 JSON body 传递，自动绑定到 Pydantic 请求模型 -- **internal 端点**：设置 `internal = true` 后，不注册 HTTP 路由，仅可通过 Zenoh RPC 内部调用 +| method | 参数来源 | 请求绑定 | +| ------ | ----------------- | ---------------------------- | +| `GET` | URL query string | TOML `params` 定义 | +| `POST` | JSON body | 自动绑定到 Pydantic 请求模型 | +| `MQTT` | MQTT payload JSON | 同 POST | + +- 设置 `internal = true` 后，端点不注册 HTTP 路由，仅可通过 Zenoh RPC 或 MQTT 内部调用 +- MQTT 方法返回 `None` 时，Bridge 自动跳过响应发布 --- ## 命名约定 +| 概念 | 约定 | 示例 | +| ------------- | ------------------------------ | ---------------------------- | +| 插件名 | 小写下划线 | `aibox_sophon_adapter` | +| 操作名 | 小写下划线 | `board_ping`, `list_devices` | +| Zenoh subject | `<插件>.<操作>` | `example.add` | +| TOML 文件名 | 与操作名一致 | `add.toml` | +| Python 文件名 | 与操作名一致 | `add.py` | +| 模型类名 | `<操作><方法>Request/Response` | `AddGetRequest` | + | 概念 | 约定 | 示例 | | ----------------- | --------------- | ----------------------------------- | | 插件名 | 小写下划线 | `doubao`, `auth`, `example` |