diff --git a/README.md b/README.md index 7232025..5441739 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,89 @@ summary = "Add via MQTT" params = ["a: float", "b: float"] ``` +### 两种路由注册方式 + +本项目支持两种路由注册方式，根据场景选用： + +#### 方式一：TOML 动态路由（标准 JSON 端点） + +通过 `config/workers/<插件>/*.toml` 定义 `[GET]` / `[POST]` 节，网关自动生成 FastAPI 路由。 +适用于参数和响应可被 Pydantic 模型描述的标准 JSON 接口。 + +- 路由生成流程：`TOML 配置 → 自动生成 Pydantic 模型 → 注册 FastAPI 路由` +- 底层 RPC：`FastAPI 路由 → WorkerManager.call() → Zenoh Query → Worker @rpc_method → 响应` + +```toml +# config/workers/example/add.toml +[GET] +summary = "加法" +params = ["a: float", "b: float"] +[GET.response_model] +result = { type = "float", required = true } +``` + +#### 方式二：插件自定义路由（特殊端点） + +当端点无法用 TOML 描述时（如 `multipart/form-data` 文件上传），插件可在模块顶层暴露 +一个 `APIRouter` 实例，网关在 `load_workers()` 阶段自动发现并注册（无需修改网关代码）。 + +**约定**：插件模块（`__init__.py`）需导入并暴露一个名为 `router` 的 `APIRouter` 实例： + +```python +# src/plugins//__init__.py +from .routes import router # 暴露给网关自动发现 +``` + +```python +# src/plugins//routes.py +from fastapi import APIRouter, UploadFile, File, Form + +router = APIRouter(prefix="/", tags=[""]) + +@router.post("/upload") +async def upload( + file: UploadFile = File(...), + param1: str = Form(...), +): + ... + return {"result": "ok"} +``` + +- 全局前缀（如 `/api/v1`）由网关自动注入，插件无需关心 +- tags 应与 TOML 自动生成的端点保持一致（使用 worker_type） +- TOML 配置中对应端点应设 `internal = true`，避免重复注册 + +### HTTP / MQTT 调用路径 + +同一端点通过**不同通道**到达 Worker 的路径对比： + +| 通道 | 调用路径 | 适用场景 | +| -------- | --------------------------------------------------------------------------------------- | ------------------------------ | +| **HTTP** | `客户端 → FastAPI 路由 → WorkerManager.call() → Zenoh Query → Worker` | REST API、Web 前端、文件上传 | +| **MQTT** | `MQTT 客户端 → 内嵌 Broker → MQTT Bridge → WorkerManager.call() → Zenoh Query → Worker` | IoT 设备上报、心跳、低带宽环境 | + +HTTP 和 MQTT 最终都通过 Zenoh RPC 调用 Worker 的 `@rpc_method` 处理函数，业务逻辑完全共享。 + +``` + ┌──────────────────────────────┐ + │ WorkerManager │ + │ (负载均衡 / 超时重试) │ + └──────────┬───────────────────┘ + │ Zenoh Query + ┌──────────▼───────────────────┐ + │ Worker 副本池 │ + │ ┌─────┐ ┌─────┐ ┌─────┐ │ + │ │ W1 │ │ W2 │ │ W3 │ │ + │ └──┬──┘ └──┬──┘ └──┬──┘ │ + │ │ │ │ │ + │ @rpc_method 处理函数 │ + └────────────────────────────────┘ + HTTP 请求到达 ▲ ▲ MQTT 消息到达 + ┌───────┘ └───────┐ + FastAPI 路由（TOML 自动生成） MQTT Bridge + 插件自定义路由（router 约定） (chongming/+/+) +``` + ### 添加新 Worker 插件 1. 创建 `src/plugins//` — `__init__.py`, `global_var.py`, `router/`, `models/` @@ -245,11 +328,12 @@ uv pip install -e utils/ # 安装/更新工具包子包 ## 内置插件 -| 插件 | HTTP 前缀 | MQTT 主题前缀 | 说明 | -| ----------- | --------- | ------------- | ------------------------------------ | -| **auth** | `/auth` | `auth.*` | 注册、登录、JWT 鉴权、个人信息、改密 | -| **doubao** | `/doubao` | `doubao.*` | 火山引擎豆包视觉理解模型 | -| **example** | `/calc` | `example.*` | 四则运算 + 访问计数 + MQTT 心跳示例 | +| 插件 | HTTP 前缀 | MQTT 主题前缀 | 说明 | +| ------------------------ | --------- | ------------------------ | ------------------------------------ | +| **auth** | `/auth` | `auth.*` | 注册、登录、JWT 鉴权、个人信息、改密 | +| **doubao** | `/doubao` | `doubao.*` | 火山引擎豆包视觉理解模型 | +| **example** | `/calc` | `example.*` | 四则运算 + 访问计数 + MQTT 心跳示例 | +| **aibox_sophon_adapter** | `/aibox` | `aibox_sophon_adapter.*` | AI Box 设备心跳适配 + 视频上传 | --- diff --git a/pyproject.toml b/pyproject.toml index 243961b..a9ecf9a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dependencies = [ "chongming-password", "chongming-worker", "amqtt>=0.11.3", + "aiofile>=3.11.1", ] description = "基于 modules-bank 的 fastapi 打包脚手架" name = "chongming-zenoh" diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py index 7e90e4a..839708b 100644 --- a/src/chongming/app/__init__.py +++ b/src/chongming/app/__init__.py @@ -42,11 +42,11 @@ async def lifespan(app: FastAPI): # 4. WorkerManager manager = lifespan_setup.setup_worker_manager(app, session) - # 5. 加载 workers - worker_tasks = await lifespan_setup.load_workers(app, session, manager) + # 5. 加载 workers（同时自动发现插件自定义路由） + worker_tasks, custom_routers = await lifespan_setup.load_workers(app, session, manager) - # 6. 注册路由 - await lifespan_setup.register_routes(app, manager) + # 6. 注册路由（含 TOML 动态路由 + 插件自定义路由） + await lifespan_setup.register_routes(app, manager, custom_routers) # 7. 启动 MQTT Bridge（内嵌 Broker + Worker 桥接） await lifespan_setup.setup_mqtt(app, manager) diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index fe2fdcd..f74bde1 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -218,13 +218,18 @@ def setup_worker_manager(app: FastAPI, session: zenoh.Session) -> WorkerManager: async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerManager): """ 加载配置中的 worker 插件，启动对应的 worker 任务。 - 返回任务列表。 + + 同时自动发现插件模块中暴露的 APIRouter（约定：模块顶层名为 `router` 的 APIRouter 实例）， + 收集后供 register_custom_routes() 统一注册，实现网关与插件的解耦。 + + 返回: (worker_tasks, custom_routers) """ workers_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml") workers_conf = get_workers_config(workers_config_path).get("workers", []) logger.info(f"读取到 {len(workers_conf)} 个 worker 配置") worker_tasks = [] + custom_routers = [] for wc in workers_conf: module_name = wc.get("module", "") factory_name = wc.get("factory", "") @@ -237,6 +242,15 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana try: module = __import__(module_name, fromlist=[factory_name]) + + # 自动发现插件的自定义 FastAPI 路由（约定：模块暴露 `router` 属性） + plugin_router = getattr(module, "router", None) + if plugin_router is not None: + from fastapi import APIRouter + if isinstance(plugin_router, APIRouter): + custom_routers.append(plugin_router) + logger.info(f"发现插件自定义路由: {module_name}.router") + factory = getattr(module, factory_name) if inspect.iscoroutinefunction(factory): tasks = await factory(session, worker_config, count, worker_manager=manager) @@ -248,13 +262,18 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana logger.error(f"加载 Worker 插件失败 {module_name}: {e}") app.state.worker_tasks = worker_tasks - return worker_tasks + return worker_tasks, custom_routers -async def register_routes(app: FastAPI, manager: WorkerManager): - """注册动态路由。""" +async def register_routes(app: FastAPI, manager: WorkerManager, custom_routers: list | None = None): + """注册动态路由和插件自定义路由。""" prefix = config["default"]["prefix"] register_dynamic_routes(app, manager, prefix=prefix) - # 可选：等待路由注册生效（已在 register_dynamic_routes 内部完成日志） + # 注册插件自定义路由（如 FORM-DATA 文件上传等 TOML 无法描述的端点） + # 统一加上全局 prefix（如 /api/v1），与 TOML 动态路由保持一致 + if custom_routers: + for r in custom_routers: + app.include_router(r, prefix=prefix) + logger.info(f"已注册 {len(custom_routers)} 个插件自定义路由（prefix={prefix}）") def setup_mqtt_broker(app: FastAPI, host: str, port: int): """启动 MQTT Broker 子进程（原子锁确保多 worker 仅启动一个）""" diff --git a/src/chongming/app/routes/storage.py b/src/chongming/app/routes/storage.py index 098907b..8376dee 100644 --- a/src/chongming/app/routes/storage.py +++ b/src/chongming/app/routes/storage.py @@ -1,10 +1,11 @@ # src/chongming/app/routes/storage.py import os -import shutil import uuid import urllib.parse from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Request from fastapi.responses import FileResponse +from aiofile import AIOFile + router = APIRouter(prefix="/api/v1/storage", tags=["Storage"]) @@ -21,8 +22,9 @@ async def upload_file( dest = os.path.join(upload_root, object_key) os.makedirs(os.path.dirname(dest), exist_ok=True) - with open(dest, "wb") as f: - shutil.copyfileobj(file.file, f) + async with AIOFile(dest, "wb") as afp: + while chunk := await file.read(1024 * 1024): + await afp.write(chunk) file_size = os.path.getsize(dest) return { @@ -33,7 +35,6 @@ async def upload_file( "size": file_size, } - @router.get("/download/{file_key:path}") async def download_file(request: Request, file_key: str): """从本地存储下载文件""" @@ -48,4 +49,4 @@ async def download_file(request: Request, file_key: str): path, filename=raw_filename, headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded}"}, - ) \ No newline at end of file + ) diff --git a/uv.lock b/uv.lock index e882a19..c4371b1 100644 --- a/uv.lock +++ b/uv.lock @@ -2,6 +2,18 @@ version = 1 revision = 3 requires-python = ">=3.12" +[[package]] +name = "aiofile" +version = "3.11.1" +source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" } +dependencies = [ + { name = "caio" }, +] +sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/48/41/2fea7e193e061ce54eacc3b7bc0e6a99e4fcff43c78cf0a76dd781ed8334/aiofile-3.11.1.tar.gz", hash = "sha256:1f91912c6643d2a4e49ca4ae3514f0bf3867ce948a36d99a6411b8f4755f4cf9", size = 19342, upload-time = "2026-05-16T08:18:33.538Z" } +wheels = [ + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/67/cd/0d76dfc5de72bde52f55f53e925c7d152d9c7906634ec1e0cbc7e8d4ad93/aiofile-3.11.1-py3-none-any.whl", hash = "sha256:ce77d14ac07f77bc2b757834a5c129321f3f705c474593deed5ab209079a52c9", size = 20446, upload-time = "2026-05-16T08:18:32.051Z" }, +] + [[package]] name = "aiosqlite" version = "0.22.1" @@ -81,6 +93,27 @@ wheels = [ { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/42/c9/8638db32514dbb9157b3d82680c6faea89283523edf9ed2415ea3884f2ae/apscheduler-3.11.3-py3-none-any.whl", hash = "sha256:bbeb2ec02d23d3c06a6c07ed7f0f3939ada6680eb121fae809a69bb42c537a30", size = 66024, upload-time = "2026-06-28T19:39:20.982Z" }, ] +[[package]] +name = "caio" +version = "0.9.25" +source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" } +sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/92/88/b8527e1b00c1811db339a1df8bd1ae49d146fcea9d6a5c40e3a80aaeb38d/caio-0.9.25.tar.gz", hash = "sha256:16498e7f81d1d0f5a4c0ad3f2540e65fe25691376e0a5bd367f558067113ed10", size = 26781, upload-time = "2025-12-26T15:21:36.501Z" } +wheels = [ + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/d3/25/79c98ebe12df31548ba4eaf44db11b7cad6b3e7b4203718335620939083c/caio-0.9.25-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:fb7ff95af4c31ad3f03179149aab61097a71fd85e05f89b4786de0359dffd044", size = 36983, upload-time = "2025-12-26T15:21:36.075Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/a3/2b/21288691f16d479945968a0a4f2856818c1c5be56881d51d4dac9b255d26/caio-0.9.25-cp312-cp312-manylinux2010_x86_64.manylinux2014_x86_64.manylinux_2_12_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:97084e4e30dfa598449d874c4d8e0c8d5ea17d2f752ef5e48e150ff9d240cd64", size = 82012, upload-time = "2025-12-26T15:22:20.983Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/03/c4/8a1b580875303500a9c12b9e0af58cb82e47f5bcf888c2457742a138273c/caio-0.9.25-cp312-cp312-manylinux_2_34_aarch64.whl", hash = "sha256:4fa69eba47e0f041b9d4f336e2ad40740681c43e686b18b191b6c5f4c5544bfb", size = 81502, upload-time = "2026-03-04T22:08:22.381Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/d1/1c/0fe770b8ffc8362c48134d1592d653a81a3d8748d764bec33864db36319d/caio-0.9.25-cp312-cp312-manylinux_2_34_x86_64.whl", hash = "sha256:6bebf6f079f1341d19f7386db9b8b1f07e8cc15ae13bfdaff573371ba0575d69", size = 80200, upload-time = "2026-03-04T22:08:23.382Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/31/57/5e6ff127e6f62c9f15d989560435c642144aa4210882f9494204bc892305/caio-0.9.25-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:d6c2a3411af97762a2b03840c3cec2f7f728921ff8adda53d7ea2315a8563451", size = 36979, upload-time = "2025-12-26T15:21:35.484Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/a3/9f/f21af50e72117eb528c422d4276cbac11fb941b1b812b182e0a9c70d19c5/caio-0.9.25-cp313-cp313-manylinux2010_x86_64.manylinux2014_x86_64.manylinux_2_12_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:0998210a4d5cd5cb565b32ccfe4e53d67303f868a76f212e002a8554692870e6", size = 81900, upload-time = "2025-12-26T15:22:21.919Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/9c/12/c39ae2a4037cb10ad5eb3578eb4d5f8c1a2575c62bba675f3406b7ef0824/caio-0.9.25-cp313-cp313-manylinux_2_34_aarch64.whl", hash = "sha256:1a177d4777141b96f175fe2c37a3d96dec7911ed9ad5f02bac38aaa1c936611f", size = 81523, upload-time = "2026-03-04T22:08:25.187Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/22/59/f8f2e950eb4f1a5a3883e198dca514b9d475415cb6cd7b78b9213a0dd45a/caio-0.9.25-cp313-cp313-manylinux_2_34_x86_64.whl", hash = "sha256:9ed3cfb28c0e99fec5e208c934e5c157d0866aa9c32aa4dc5e9b6034af6286b7", size = 80243, upload-time = "2026-03-04T22:08:26.449Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/69/ca/a08fdc7efdcc24e6a6131a93c85be1f204d41c58f474c42b0670af8c016b/caio-0.9.25-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:fab6078b9348e883c80a5e14b382e6ad6aabbc4429ca034e76e730cf464269db", size = 36978, upload-time = "2025-12-26T15:21:41.055Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/5e/6c/d4d24f65e690213c097174d26eda6831f45f4734d9d036d81790a27e7b78/caio-0.9.25-cp314-cp314-manylinux2010_x86_64.manylinux2014_x86_64.manylinux_2_12_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:44a6b58e52d488c75cfaa5ecaa404b2b41cc965e6c417e03251e868ecd5b6d77", size = 81832, upload-time = "2025-12-26T15:22:22.757Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/87/a4/e534cf7d2d0e8d880e25dd61e8d921ffcfe15bd696734589826f5a2df727/caio-0.9.25-cp314-cp314-manylinux_2_34_aarch64.whl", hash = "sha256:628a630eb7fb22381dd8e3c8ab7f59e854b9c806639811fc3f4310c6bd711d79", size = 81565, upload-time = "2026-03-04T22:08:27.483Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/3f/ed/bf81aeac1d290017e5e5ac3e880fd56ee15e50a6d0353986799d1bc5cfd5/caio-0.9.25-cp314-cp314-manylinux_2_34_x86_64.whl", hash = "sha256:0ba16aa605ccb174665357fc729cf500679c2d94d5f1458a6f0d5ca48f2060a7", size = 80071, upload-time = "2026-03-04T22:08:28.751Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/86/93/1f76c8d1bafe3b0614e06b2195784a3765bbf7b0a067661af9e2dd47fc33/caio-0.9.25-py3-none-any.whl", hash = "sha256:06c0bb02d6b929119b1cfbe1ca403c768b2013a369e2db46bfa2a5761cf82e40", size = 19087, upload-time = "2025-12-26T15:22:00.221Z" }, +] + [[package]] name = "certifi" version = "2026.6.17" @@ -225,6 +258,7 @@ name = "chongming-zenoh" version = "0.1.0" source = { editable = "." } dependencies = [ + { name = "aiofile" }, { name = "amqtt" }, { name = "apscheduler" }, { name = "chongming-config" }, @@ -258,6 +292,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "aiofile", specifier = ">=3.11.1" }, { name = "amqtt", specifier = ">=0.11.3" }, { name = "apscheduler" }, { name = "chongming-config", git = "http://124.71.68.6:3000/chongming/chongming-utils-config.git" },