159 KiB
./pyproject.toml
[build-system]
build-backend = "hatchling.build"
requires = ["hatchling"]
[project]
dependencies = [
"apscheduler",
"boto3>=1.43.48",
"chongming-worker",
"diskcache",
"eclipse-zenoh",
"fastapi[uvicorn]",
"gunicorn",
"modelvault>=0.16.0",
"module-bank",
"nanio>=0.1.5",
"orjson>=3.11.9",
"python-multipart>=0.0.32",
"sqlite-vfs",
"tomli",
"typing-extensions",
"uvicorn",
]
description = "基于 modules-bank 的 fastapi 打包脚手架"
name = "chongming-zenoh"
readme = "README.md"
requires-python = ">=3.12"
version = "0.1.0"
[dependency-groups]
dev = [
"nuitka",
"pyarmor",
"pyinstaller",
"ruff",
]
[tool.hatch.build.targets.wheel]
packages = ["src/chongming"]
[[tool.uv.index]]
url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple"
[tool.uv.workspace]
members = [
"utils/worker",
"utils/request",
]
[tool.uv.sources]
chongming-worker = { workspace = true }
[project.scripts]
build = "chongming.scripts.build:build"
gen_models = "chongming.scripts.gen_models:gen_models"
gunicorn = "chongming:gunicorn_serve"
serve = "chongming:serve"
./config.toml
[default]
app.debug = false
app.description = "基于 FastAPI 的 chongming 打包脚手架"
app.name = "chongming"
app.version = "chongming"
env = "development"
prefix = "/api/v1"
upload_path = "uploads"
[database]
database_path = "./data/database.db"
type = "sqlite"
url = "sqlite+aiosqlite:///./data/database.db"
[database.sqlite]
connect_args.check_same_thread = false
connect_args.timeout = 15
echo = true
echo_pool = true
max_overflow = 10
pool_pre_ping = true
pool_recycle = 300
pool_size = 20
pool_timeout = 30
[scheduler]
job_store_path = "scheduler_jobs.db"
[cache]
cache_store_path = "./cache_data"
[security]
access_token_expire_minutes = 1440
algorithm = "HS256"
max_sessions_per_user = 5
refresh_token_expire_days = 7
secret_key = "your-secret-key-change-in-production"
[development]
debug = true
env = "development"
[development.server]
access_log = false
host = "0.0.0.0"
port = 8000
reload = true
[development.logging]
file = "logs"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
level = "DEBUG"
[development.cors]
allow_credentials = true
allow_headers = ["*"]
allow_methods = ["*"]
allow_origins = ["*"]
[plugins_worker]
config_path = "./config/workers/config.toml"
[pyinstaller_build_import]
base = [
"utils", # public 骨架
"utils.launch",
"utils.config",
"zenoh",
"chongming_worker",
"server",
"module_bank", # 模块化打包工具
"sqlite_vfs",
"sqlite_vfs.core",
"fastapi", # web 应用依赖
"fastapi.staticfiles",
"fastapi.middleware.cors",
"uvicorn",
"apscheduler",
"diskcache",
"orjson",
"python_multipart",
"nanio",
"boto3",
]
linux = [
"gunicorn.glogging",
"gunicorn.app",
"gunicorn.app.wsgiapp",
]
windows = []
[production.module_system]
path = [
["resources/app.mbank", "src/chongming/app", "app"],
["resources/plugins.mbank", "src/plugins", "plugins"],
]
type = "module-bank"
./README.md
./build_nuitka/config.toml
[default]
app.debug = false
app.description = "基于 FastAPI 的 chongming 打包脚手架"
app.name = "chongming"
app.version = "0.1.0"
env = "production"
prefix = "/api/v1"
upload_path = "uploads"
[database]
generate_schemas = true # 生产环境使用迁移工具
type = "sqlite"
url = "sqlite+aiosqlite:///./data/database.db"
[database.sqlite]
connect_args.check_same_thread = false
connect_args.timeout = 15
echo = false
echo_pool = false
max_overflow = 10
pool_pre_ping = true
pool_recycle = 300
pool_size = 20
pool_timeout = 30
[scheduler]
job_store_path = "scheduler_jobs.db"
[cache]
cache_store_path = "./"
[security]
access_token_expire_minutes = 30
algorithm = "HS256"
max_sessions_per_user = 5
refresh_token_expire_days = 7
secret_key = "LClT3DaMKZzvT1L4bgsxvCSM5q7uPeX3R2_FZ8xzwu8="
[development]
debug = true
env = "development"
[development.server]
access_log = false
host = "127.0.0.1"
port = 8001
reload = true
[development.logging]
file = "logs"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
level = "DEBUG"
[development.cors]
allow_credentials = true
allow_headers = ["*"]
allow_methods = ["*"]
allow_origins = ["*"]
[production]
debug = false
env = "production"
[production.server]
access_log = false
host = "0.0.0.0"
port = 8000
reload = false
timeout_keep_alive = 65
workers = 4
[production.logging]
backup_count = 5 # 最多保存 10 个日志文件
file = "logs"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
level = "INFO"
max_size = 10485760 # 10MB
[production.cors]
allow_credentials = true
allow_headers = ["*"]
allow_methods = ["*"]
allow_origins = ["http://192.168.137.5:8000", "http://localhost:8000", "http://127.0.0.1:8000"]
[production.module_system]
path = [
"resources/app.mbank",
"resources/plugins.mbank",
]
type = "module-bank"
[production.file_system]
path = "static.svfs"
type = "svfs"
[[workers]]
type = "example"
module = "plugins.example"
factory = "create_workers"
count = 3
config = { type = "example" }
./build_nuitka/config/workers/example/multiply.toml
# multiply 接口
[GET]
summary = "Multiply two numbers"
docstring = "Multiply two numbers via GET query params (a * b)"
params = ["a: float", "b: float"]
response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "multiply" }, timestamp = { type = "float", required = false, default = 0.0 } }
./build_nuitka/config/workers/example/divide.toml
# divide 接口
[GET]
summary = "Divide two numbers"
docstring = "Divide two numbers via GET query params (a / b). Returns error if b == 0."
params = ["a: float", "b: float"]
response_model = { result = { type = "float", required = false, default = 0.0 }, operation = { type = "str", default = "divide" }, timestamp = { type = "float", required = false, default = 0.0 }, error = { type = "str", required = false, default = "" } }
./build_nuitka/config/workers/example/subtract.toml
# subtract 接口
[GET]
summary = "Subtract two numbers"
docstring = "Subtract two numbers via GET query params (a - b)"
params = ["a: float", "b: float"]
response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "subtract" }, timestamp = { type = "float", required = false, default = 0.0 } }
./build_nuitka/config/workers/example/add.toml
# add 接口 — 按请求类型分节
# subject 会带上方法后缀: example.add.GET / example.add.POST
[GET]
cacheable = true
summary = "Add two numbers"
docstring = "Add two numbers via GET query params"
params = ["a: float", "b: float"]
response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "add" }, timestamp = { type = "float", required = false, default = 0.0 } }
[POST]
cacheable = true
summary = "Add two numbers (POST)"
docstring = "Add two numbers via POST JSON body"
params = ["a: float", "b: float"]
response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "add" }, timestamp = { type = "float", required = false, default = 0.0 } }
./build_nuitka/config/workers/example/config.toml
[worker]
name = "example"
version = "0.1.0"
description = "chongming worker example — demonstrates new config features"
config_version = "1.0"
defaults = { method = "GET", ttl = 30, timeout = 2.0, auth_required = true, cacheable = false }
[worker.registration]
type = "register"
service = "example"
worker_type = "calc-workers"
router_prefix = "/calc"
tags = ["example"]
./utils/worker/pyproject.toml
[build-system]
build-backend = "hatchling.build"
requires = ["hatchling"]
[project]
dependencies = [
"eclipse-zenoh>=1.9.0",
"tomli>=2.4.1",
]
description = "chongming worker"
name = "chongming-worker"
readme = "README.md"
requires-python = ">=3.12"
version = "0.1.0"
[tool.hatch.build.targets.wheel]
packages = ["src/chongming_worker"]
./utils/worker/README.md
./utils/worker/src/chongming_worker/decorators.py
"""
通用 Working 装饰器框架
提供 @rpc_method 装饰器，用于将普通函数注册为 Zenoh RPC 端点。
装饰器自动处理：
- 访问类型（GET/POST/PUT/DELETE）
- 接口名称（subject）
- 参数解析与类型转换
- 响应序列化
兼容 gen_models 命令生成的 Pydantic 模型：
- 通过 request_model / response_model 参数绑定已生成的模型
- 自动从插件 models 目录发现匹配的模型
- 请求参数自动校验与响应格式自动序列化
"""
import asyncio
import json
import time
import inspect
import importlib
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, get_type_hints, Union
from pydantic import BaseModel, Field, create_model, ValidationError
# ============================================================
# 类型系统
# ============================================================
TYPE_MAP: Dict[str, Any] = {
"float": float,
"int": int,
"str": str,
"bool": bool,
"any": Any,
}
def _resolve_type(annotation: Any) -> Any:
"""将类型注解解析为可用的 Python 类型"""
if annotation is inspect.Parameter.empty:
return str
origin = getattr(annotation, "__origin__", None)
if origin is list:
args = getattr(annotation, "__args__", (Any,))
return List[args[0]] if args else List[Any] # type: ignore
if origin is dict:
args = getattr(annotation, "__args__", (str, Any))
return Dict[args[0], args[1]] if len(args) == 2 else Dict[str, Any] # type: ignore
if origin is tuple:
args = getattr(annotation, "__args__", (Any,))
return Tuple[tuple(args)] # type: ignore
if origin is Optional:
args = getattr(annotation, "__args__", (Any,))
return Optional[args[0]] # type: ignore
return annotation
def _type_to_config_str(py_type: Any) -> str:
"""将 Python 类型转换为配置字符串（用于 gen_models 兼容）"""
origin = getattr(py_type, "__origin__", None)
# 处理 Union 类型（包括 Optional[X]）
if origin is Union:
args = getattr(py_type, "__args__", (Any,))
non_none_args = [a for a in args if a is not type(None)]
if len(non_none_args) == 1:
return _type_to_config_str(non_none_args[0])
return "str"
if origin is list:
args = getattr(py_type, "__args__", (Any,))
return f"list[{_type_to_config_str(args[0])}]"
if origin is dict:
args = getattr(py_type, "__args__", (str, Any))
return f"dict[{_type_to_config_str(args[0])}, {_type_to_config_str(args[1])}]"
if py_type is float:
return "float"
if py_type is int:
return "int"
if py_type is bool:
return "bool"
if py_type is str:
return "str"
return "str"
# ============================================================
# 装饰器定义
# ============================================================
class RPCMethod:
"""
装饰后的 RPC 方法元数据
存储原始函数、访问类型、接口名称、参数信息等，
供 Worker 在运行时发现和调用。
支持绑定 gen_models 生成的 Pydantic 模型：
- request_model: 用于请求参数的自动校验和转换
- response_model: 用于响应数据的格式化和序列化
"""
def __init__(
self,
func: Callable,
subject: str,
method: str = "GET",
summary: str = "",
timeout: float = 3.0,
request_model: Optional[Type[BaseModel]] = None,
response_model: Optional[Type[BaseModel]] = None,
):
self.func = func
self.subject = subject
self.method = method.upper()
self.summary = summary or func.__doc__ or ""
self.timeout = timeout
self.name = func.__name__
# 绑定的 Pydantic 模型（由 gen_models 生成，或手动传入）
self.request_model: Optional[Type[BaseModel]] = request_model
self.response_model: Optional[Type[BaseModel]] = response_model
# 解析函数签名
sig = inspect.signature(func)
hints = get_type_hints(func)
self.params: Dict[str, Any] = {}
self.model_param: Optional[str] = None # 如果有参数是 BaseModel，记录参数名
self.return_type: Any = str
for param_name, param in sig.parameters.items():
if param_name == "self":
continue
py_type = _resolve_type(hints.get(param_name, str))
# 检查参数类型是否是 BaseModel 子类
if isinstance(py_type, type) and issubclass(py_type, BaseModel):
self.model_param = param_name
# 如果是 BaseModel 参数，直接用它的字段作为 params（用于配置和文档）
for field_name, field in py_type.model_fields.items():
self.params[field_name] = _resolve_type(field.annotation)
else:
self.params[param_name] = py_type
# 返回值类型
self.return_type = _resolve_type(hints.get("return", Any))
def build_pydantic_request_model(self, model_name: str = "AutoRequest") -> Type[BaseModel]:
"""根据参数注解生成 Pydantic 请求模型"""
fields: Dict[str, Any] = {}
for param_name, py_type in self.params.items():
fields[param_name] = (py_type, Field(...))
return create_model(model_name, **fields)
def build_pydantic_response_model(self, model_name: str = "AutoResponse") -> Type[BaseModel]:
"""根据返回值注解生成 Pydantic 响应模型"""
fields: Dict[str, Any] = {}
# 如果返回类型是 BaseModel 子类，直接使用
if isinstance(self.return_type, type) and issubclass(self.return_type, BaseModel):
return self.return_type
# 否则包装为 {result: return_type}
fields["result"] = (self.return_type, Field(...))
return create_model(model_name, **fields)
def get_request_model(self) -> Optional[Type[BaseModel]]:
"""
获取有效的请求模型，优先级：
1. 绑定的 request_model（由用户或 gen_models 提供）
2. 如果函数参数本身就是 Pydantic 模型（model_param），从中提取字段生成模型
3. 自动生成的模型（根据函数参数注解，仅 POST/PUT/PATCH）
"""
if self.request_model is not None:
return self.request_model
# 如果函数接受了 Pydantic 模型参数，用它的字段生成请求模型
if self.model_param is not None:
sig = inspect.signature(self.func)
hints = get_type_hints(self.func)
model_type = hints.get(self.model_param)
if isinstance(model_type, type) and issubclass(model_type, BaseModel):
return model_type
# 对于 POST/PUT/PATCH，根据 params 自动生成
if self.params and self.method in ("POST", "PUT", "PATCH"):
return self.build_pydantic_request_model(f"{self.name.title()}_Request")
return None
def get_response_model(self) -> Optional[Type[BaseModel]]:
"""
获取有效的响应模型，优先级：
1. 绑定的 response_model（由用户或 gen_models 提供）
2. 自动生成的模型（根据函数返回值注解）
"""
if self.response_model is not None:
return self.response_model
return self.build_pydantic_response_model(f"{self.name.title()}_Response")
def validate_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
使用 Pydantic 模型校验请求参数
Args:
payload: 原始请求参数字典
Returns:
校验通过后的参数字典（包含类型转换）
"""
req_model = self.get_request_model()
if req_model is None:
return payload
try:
validated = req_model(**payload)
return validated.model_dump()
except ValidationError as e:
raise ValueError(f"请求参数校验失败: {e.errors()}")
def serialize_response(self, raw_result: Any) -> Any:
"""
使用 Pydantic 模型序列化响应数据
Args:
raw_result: 原始函数返回值
Returns:
序列化后的响应数据（dict 或列表等可 JSON 序列化的类型）
"""
resp_model = self.get_response_model()
if resp_model is None:
return raw_result
if isinstance(raw_result, dict):
try:
validated = resp_model(**raw_result)
return validated.model_dump()
except ValidationError:
# 如果无法用响应模型校验，返回原始结果
return raw_result
if isinstance(raw_result, resp_model):
return raw_result.model_dump()
return raw_result
async def call(self, payload: Dict[str, Any]) -> Any:
"""
调用原始函数，自动从 payload 提取参数
流程：
1. 校验请求参数（使用绑定的 request_model 或自动生成）
2. 提取函数需要的参数
3. 调用函数
4. 序列化响应（使用绑定的 response_model 或自动生成）
支持两种函数签名：
- 展开参数: def add(a: float, b: float) -> ... # 传统方式
- Pydantic 模型: def add(req: AddGetRequest) -> ... # 工程化方式
Args:
payload: 请求参数字典
Returns:
序列化后的响应数据
"""
# 1. 校验请求参数
validated_payload = self.validate_request(payload)
# 2. 提取函数需要的参数
kwargs = {}
if self.model_param is not None:
# 2a. 函数参数是 Pydantic 模型 — 将整个校验后的 payload 转为模型实例传入
req_model = self.get_request_model()
if req_model is not None:
kwargs[self.model_param] = req_model(**validated_payload)
else:
# 如果没有 request_model，从函数签名获取模型类型
sig = inspect.signature(self.func)
hints = get_type_hints(self.func)
model_type = hints.get(self.model_param)
if isinstance(model_type, type) and issubclass(model_type, BaseModel):
kwargs[self.model_param] = model_type(**validated_payload)
else:
kwargs[self.model_param] = validated_payload
else:
# 2b. 传统方式：逐字段提取参数
for param_name in self.params:
if param_name in validated_payload:
kwargs[param_name] = validated_payload[param_name]
# 3. 调用函数
if asyncio.iscoroutinefunction(self.func):
result = await self.func(**kwargs)
else:
result = self.func(**kwargs)
# 4. 序列化响应
return self.serialize_response(result)
# 全局注册表：worker_type -> {subject -> RPCMethod}
_RPC_REGISTRY: Dict[str, Dict[str, RPCMethod]] = {}
def rpc_method(
subject: str,
method: str = "GET",
summary: str = "",
timeout: float = 3.0,
request_model: Optional[Type[BaseModel]] = None,
response_model: Optional[Type[BaseModel]] = None,
):
"""
装饰器：将函数注册为 RPC 端点
用法:
# 基本用法（自动生成模型）
@rpc_method(subject="calc.add", method="GET", summary="Add two numbers")
def add(a: float, b: float) -> float:
return a + b
# 绑定 gen_models 生成的 Pydantic 模型
@rpc_method(
subject="calc.add",
method="GET",
summary="Add two numbers",
response_model=AddGetResponse,
)
def add(a: float, b: float):
return a + b
参数:
subject: 接口名称，格式为 "worker_type.operation"
method: HTTP 方法 (GET/POST/PUT/DELETE)
summary: 接口描述
timeout: 超时时间（秒）
request_model: 可选的 Pydantic 请求模型（用于验证请求参数）
response_model: 可选的 Pydantic 响应模型（用于格式化响应）
"""
def decorator(func: Callable) -> Callable:
rpc = RPCMethod(
func=func,
subject=subject,
method=method,
summary=summary,
timeout=timeout,
request_model=request_model,
response_model=response_model,
)
# 提取 worker_type（subject 的第一部分）
worker_type = subject.split(".")[0] if "." in subject else "default"
if worker_type not in _RPC_REGISTRY:
_RPC_REGISTRY[worker_type] = {}
# 使用 subject + "#" + method 作为唯一 key，允许多个 HTTP 方法共享同一 subject
registry_key = f"{subject}#{method.upper()}"
_RPC_REGISTRY[worker_type][registry_key] = rpc
# 给函数附加元数据，方便外部访问
func._rpc_meta = rpc # type: ignore
return func
return decorator
def bind_model(
subject: str = "",
method: str = "",
request_model: Optional[Type[BaseModel]] = None,
response_model: Optional[Type[BaseModel]] = None,
):
"""
装饰器：为已注册的 RPC 端点绑定 Pydantic 模型
用于 gen_models 生成模型后，将模型绑定到用 @rpc_method 定义的端点。
无需修改原有的 @rpc_method 调用。
用法:
from .models.pydantic import AddGetResponse
@bind_model(subject="example.add", method="GET", response_model=AddGetResponse)
def add():
pass # 函数体不重要，仅用于匹配已注册的端点
参数:
subject: 要绑定的接口名称（与 @rpc_method 中的 subject 一致）
method: 要绑定的 HTTP 方法
request_model: Pydantic 请求模型
response_model: Pydantic 响应模型
"""
def decorator(func: Callable) -> Callable:
nonlocal subject, method
if not subject and hasattr(func, '_rpc_meta'):
subject = func._rpc_meta.subject # type: ignore[attr-defined]
method = func._rpc_meta.method # type: ignore[attr-defined]
worker_type = subject.split(".")[0] if "." in subject else "default"
method_upper = method.upper()
registry = _RPC_REGISTRY.get(worker_type, {})
# 用新 key 格式查找: subject#METHOD
rpc = registry.get(f"{subject}#{method_upper}")
if rpc is None:
raise ValueError(
f"未找到已注册的 RPC 端点: subject='{subject}', method='{method}'. "
f"请确保 @rpc_method 装饰器已先于 @bind_model 执行。"
)
# 如果指定了 method，进一步筛选
if method and rpc.method != method.upper():
return func
if request_model is not None:
rpc.request_model = request_model
if response_model is not None:
rpc.response_model = response_model
func._rpc_meta = rpc # type: ignore[attr-defined]
return func
return decorator
def auto_bind_models(plugin_name: str):
"""
自动从插件 models 目录发现并绑定 Pydantic 模型
扫描 src/plugins/<plugin_name>/models/pydantic.py 中定义的所有模型，
尝试按命名约定匹配已注册的 RPC 端点。
命名约定：
- 请求模型: <Operation><Method>Request (e.g., AddPostRequest)
- 响应模型: <Operation><Method>Response (e.g., AddGetResponse)
用法:
# 在插件 __init__.py 中调用
from ...decorators import auto_bind_models
auto_bind_models("example")
"""
try:
models_module = importlib.import_module(f"plugins.{plugin_name}.models.pydantic")
except ImportError:
try:
models_module = importlib.import_module(f"src.plugins.{plugin_name}.models.pydantic")
except ImportError:
print(f"⚠️ auto_bind_models: 未找到 models 模块 (plugins.{plugin_name}.models.pydantic)")
return
# 获取模块中所有 BaseModel 子类
model_classes = {}
for name in dir(models_module):
obj = getattr(models_module, name)
if isinstance(obj, type) and issubclass(obj, BaseModel) and obj is not BaseModel:
model_classes[name] = obj
if not model_classes:
print(f"⚠️ auto_bind_models: 未找到任何 Pydantic 模型")
return
registry = _RPC_REGISTRY.get(plugin_name, {})
bound_count = 0
for _key, rpc in registry.items():
op_name = rpc.subject.split(".")[-1] if "." in rpc.subject else rpc.subject
op_camel = op_name.title().replace("-", "").replace("_", "")
method_tag = rpc.method.title()
# 尝试匹配请求模型: <Op><Method>Request
req_model_name = f"{op_camel}{method_tag}Request"
if req_model_name in model_classes:
rpc.request_model = model_classes[req_model_name]
bound_count += 1
print(f" ✓ 绑定请求模型: {req_model_name} -> {rpc.subject} [{rpc.method}]")
# 尝试匹配响应模型: <Op><Method>Response
resp_model_name = f"{op_camel}{method_tag}Response"
if resp_model_name in model_classes:
rpc.response_model = model_classes[resp_model_name]
bound_count += 1
print(f" ✓ 绑定响应模型: {resp_model_name} -> {rpc.subject} [{rpc.method}]")
if bound_count > 0:
print(f"✅ auto_bind_models: 已绑定 {bound_count} 个模型到 {plugin_name} 的 RPC 端点")
else:
print(f"⚠️ auto_bind_models: 未找到匹配的模型绑定")
def get_rpc_methods(worker_type: str) -> Dict[str, RPCMethod]:
"""获取指定 worker 类型的所有 RPC 方法"""
return _RPC_REGISTRY.get(worker_type, {})
def get_all_rpc_methods() -> Dict[str, Dict[str, RPCMethod]]:
"""获取所有已注册的 RPC 方法"""
return _RPC_REGISTRY
def clear_registry():
"""清空注册表（主要用于测试）"""
_RPC_REGISTRY.clear()
# ============================================================
# 路由配置生成器（从装饰器元数据生成 TOML 兼容的配置）
# ============================================================
def generate_route_configs(worker_type: str) -> List[Dict[str, Any]]:
"""
从装饰器注册表生成路由配置列表
返回的配置与 config/workers/<type>/*.toml 格式兼容，
可用于动态路由注册或生成配置文件。
"""
methods = get_rpc_methods(worker_type)
configs: List[Dict[str, Any]] = []
for _key, rpc in methods.items():
# 从 subject 提取操作名（最后一部分）
op_name = rpc.subject.split(".")[-1] if "." in rpc.subject else rpc.subject
configs.append({
"subject": rpc.subject,
"path": f"/{op_name}",
"method": rpc.method,
"params": [f"{name}: {_type_to_config_str(pt)}" for name, pt in rpc.params.items()],
"response_model": _build_response_spec(rpc),
"summary": rpc.summary,
"timeout": rpc.timeout,
})
return configs
def _build_response_spec(rpc: RPCMethod) -> Dict[str, Any]:
"""从 RPC 方法构建 response_model 配置"""
spec: Dict[str, Any] = {}
# 优先使用绑定的 response_model
resp_model = rpc.response_model
if resp_model is None:
resp_model = rpc.build_pydantic_response_model()
if isinstance(resp_model, type) and issubclass(resp_model, BaseModel):
# 遍历 Pydantic 模型字段
for field_name, field in resp_model.model_fields.items():
field_type = _type_to_config_str(field.annotation)
required = field.is_required()
default = field.default if not field.is_required() else None
spec[field_name] = {
"type": field_type,
"required": required,
}
if default is not None:
spec[field_name]["default"] = default
else:
spec["result"] = {"type": _type_to_config_str(rpc.return_type), "required": True}
return spec
./utils/worker/src/chongming_worker/runtime.py
"""
Worker 运行时 — 通用的生命周期管理
提供 create_worker / create_workers 工厂函数，供所有 Worker 插件共享。
"""
import os
import asyncio
import json
import orjson
from typing import Any, Dict, List, Optional
import zenoh
from chongming_worker.decorators import (
RPCMethod,
get_rpc_methods,
)
async def create_worker(
session: zenoh.Session,
worker_type: str,
worker_id: str,
config: Optional[Dict[str, Any]] = None,
worker_manager=None,
):
"""
创建一个 Worker 协程
自动发现通过 @rpc_method 注册的端点，为每个 subject 声明 Zenoh Queryable，
收到请求后按 _method 字段分派到具体方法。
Args:
session: Zenoh 会话
worker_type: Worker 类型标识（匹配 subject 的第一部分）
worker_id: Worker 唯一 ID
config: 自定义配置参数
worker_manager: WorkerManager 实例（用于本地注册）
"""
print(f"[Worker] {worker_id} 启动 (type={worker_type})")
# 1. 注册存活令牌
token = session.liveliness().declare_token(f"worker/{worker_type}/{worker_id}")
# 2. 注册到本地 WorkerManager（若有）
if worker_manager is not None:
worker_manager.register_worker(worker_type, worker_id)
# 3. 获取该 worker 类型的所有 RPC 方法
rpc_methods = get_rpc_methods(worker_type)
if not rpc_methods:
print(f"[Worker] {worker_id} 警告: 未找到任何 RPC 方法")
else:
print(f"[Worker] {worker_id} 发现 {len(rpc_methods)} 个端点:")
for _key, rpc in rpc_methods.items():
print(f" {rpc.method} {rpc.subject} -> {rpc.name}({', '.join(rpc.params.keys())})")
# 4. 按 subject 分组（多个方法可共享同一 subject，如 GET + POST）
subject_methods: Dict[str, List[RPCMethod]] = {}
for registry_key, rpc in rpc_methods.items():
subject_methods.setdefault(rpc.subject, []).append(rpc)
# 5. 为每个 subject 创建回调函数并声明 Queryable
stop_event = asyncio.Event()
def make_callback(subject: str, methods: List[RPCMethod]):
"""创建处理特定 subject 的回调函数"""
def callback(query: zenoh.Query):
try:
# 解析请求负载
# payload_str = str(query.payload) if query.payload else "{}"
payload_bytes = bytes(query.payload) if query.payload else b""
if not payload_bytes:
data = {}
else:
data = orjson.loads(payload_bytes)
http_method = data.get("_method", "GET").upper()
# 匹配方法（按 HTTP 方法）
matched = next((m for m in methods if m.method.upper() == http_method), None)
if not matched and methods:
matched = methods[0] # 降级到第一个
if not matched:
query.reply(
key_expr=query.key_expr,
payload=json.dumps({"error": f"No method for {subject}"}).encode("utf-8")
)
return
# 调用 RPC 方法（支持同步和异步）
result = asyncio.run(matched.call(data))
# 回复
query.reply(
key_expr=query.key_expr,
payload=orjson.dumps(result)
)
print(f"[Worker] {worker_id} 回复 {matched.method} {subject}: {result}")
except Exception as e:
error_response = {
"result": 0.0,
"operation": "error",
"error": str(e),
"worker": worker_id
}
query.reply(
key_expr=query.key_expr,
payload=json.dumps(error_response).encode("utf-8")
)
print(f"[Worker] {worker_id} 异常 {subject}: {e}")
return callback
# 声明所有 Queryable
queryables: List[zenoh.Queryable] = []
for subject, methods in subject_methods.items():
key = f"worker/{worker_type}/{worker_id}/{subject}"
# 使用回调方式声明
q = session.declare_queryable(key, make_callback(subject, methods))
queryables.append(q)
print(f"[Worker] {worker_id} Queryable: '{key}' [{', '.join(m.method for m in methods)}]")
# 6. 保持 worker 存活直到被取消
try:
await stop_event.wait()
except asyncio.CancelledError:
print(f"[Worker] {worker_id} 正在关闭...")
except Exception as e:
print(f"[Worker] {worker_id} 发生异常: {e}")
finally:
# 清理
if worker_manager is not None:
worker_manager.unregister_worker(worker_type, worker_id)
token.undeclare()
for q in queryables:
q.undeclare()
print(f"[Worker] {worker_id} 已关闭")
def create_workers(
session: zenoh.Session,
config: dict,
count: int = 1,
worker_manager=None,
) -> list[asyncio.Task]:
"""
创建一组 Worker 协程任务
工厂函数，供 WorkerManager 动态加载使用。
Args:
session: Zenoh 会话
config: Worker 配置字典（至少需要 "type" 字段）
count: 启动的 Worker 副本数
worker_manager: WorkerManager 实例（可选）
Returns:
list[asyncio.Task]: Worker 协程任务列表
"""
worker_type = config.get("type", "default")
pid = os.getpid()
tasks = []
for i in range(count):
worker_id = f"worker-{i}-{pid}"
task = asyncio.create_task(
create_worker(
session=session,
worker_type=worker_type,
worker_id=worker_id,
config=config,
worker_manager=worker_manager,
)
)
tasks.append(task)
print(f"[Worker] 已创建任务: {worker_id} (type={worker_type})")
return tasks
__all__ = [
"create_worker",
"create_workers",
]
./utils/worker/src/chongming_worker/config.py
import sys
from pathlib import Path
from typing import TypedDict, Literal, Optional, cast, List, Dict, Any
# 根据Python版本选择合适的TOML解析库和Required类型
if sys.version_info >= (3, 11):
import tomllib
from typing import Required
else:
import tomli as tomllib
from typing_extensions import Required
# ============================================================
# Worker 配置类型
# ============================================================
class WorkerParams(TypedDict, total=False):
"""Worker 操作参数定义"""
type: str
required: bool
default: Any
description: str
class WorkerResponse(TypedDict, total=False):
"""Worker 操作响应定义"""
type: str
required: bool
default: Any
class OperationConfig(TypedDict, total=False):
"""
Worker 操作配置结构
对应 config/workers/<name>/*.toml 中的每个操作定义。
"""
name: Required[str]
method: Required[Literal["GET", "POST", "PUT", "DELETE"]]
path: str
subject: Required[str]
summary: str
timeout: float
params: Dict[str, WorkerParams]
response_model: Dict[str, WorkerResponse]
class WorkerConfig(TypedDict, total=False):
"""
Worker 配置结构
对应 config/workers/<name>/config.toml 中的顶级配置。
"""
name: Required[str]
type: Required[str]
description: str
version: str
enabled: bool
operations: List[OperationConfig]
def load_worker_config(config_dir: Path) -> Dict[str, Any]:
"""
加载 Worker 配置目录中的所有 TOML 配置文件
Args:
config_dir: Worker 配置目录路径，如 Path("config/workers/example")
Returns:
解析后的配置字典，包含所有操作的配置
Raises:
FileNotFoundError: 当配置目录不存在时抛出异常
"""
if not config_dir.exists():
raise FileNotFoundError(f"Worker 配置目录不存在: {config_dir}")
config: Dict[str, Any] = {"operations": []}
# 首先加载 config.toml（如果存在）
config_file = config_dir / "config.toml"
if config_file.exists():
with open(config_file, "rb") as f:
base_config = tomllib.load(f)
config.update(base_config)
# 然后加载所有 *.toml 中的操作定义
for toml_file in sorted(config_dir.glob("*.toml")):
if toml_file.name == "config.toml":
continue
with open(toml_file, "rb") as f:
op_data = tomllib.load(f)
# 将文件名（不含扩展名）作为默认操作名
op_name = toml_file.stem
if "name" not in op_data:
op_data["name"] = op_name
config["operations"].append(op_data)
return config
# ============================================================
# 旧的配置系统（与主应用 config.py 兼容）
# ============================================================
# 设置 env 为 development 或 production
class DefaultApp(TypedDict):
"""应用默认配置结构"""
name: str
version: str
description: str
debug: bool
class Default(TypedDict):
"""系统默认配置结构，包含应用信息、环境、路径等基础配置"""
app: DefaultApp
env: Literal["production", "development"]
prefix: str
upload_path: str
class SqliteConnectionArgs(TypedDict, total=False):
"""SQLite数据库连接参数配置"""
timeout: int
check_same_thread: bool
class DatabaseSqlite(TypedDict, total=False):
"""SQLite数据库连接池及行为配置"""
echo: bool
pool_size: int
max_overflow: int
pool_timeout: int
pool_recycle: int
pool_pre_ping: bool
connect_args: SqliteConnectionArgs
class Database(TypedDict, total=False):
"""数据库配置结构，支持SQLite、MySQL、PostgreSQL三种数据库类型"""
type: Required[Literal["sqlite", "mysql", "postgresql"]]
url: Required[str]
generate_schemas: bool
sqlite: DatabaseSqlite
mysql: dict
postgresql: dict
database_path: str
class Security(TypedDict):
"""安全相关配置，包括JWT令牌和安全策略设置"""
secret_key: str
algorithm: Literal["HS256"]
access_token_expire_minutes: int
refresh_token_expire_days: int
max_sessions_per_user: int
class Server(TypedDict, total=False):
"""服务器运行配置，包含主机、端口、工作进程等设置"""
host: Required[str]
port: Required[int]
reload: bool
workers: int
access_log: bool
class Logging(TypedDict):
"""日志系统配置，定义日志级别、输出格式和文件管理策略"""
level: str
file: str
format: str
max_size: int
backup_count: int
class Cors(TypedDict, total=False):
"""跨域资源共享(CORS)配置"""
allow_origins: List[str]
allow_credentials: bool
allow_methods: List[str]
allow_headers: List[str]
class ModuleSystem(TypedDict, total=False):
"""模块系统配置，定义模块加载类型和路径"""
type: str
path: List[str]
class FileSystem(TypedDict, total=False):
"""文件系统配置，定义存储类型和路径"""
type: str
path: str
class Env(TypedDict, total=False):
"""环境特定配置结构，用于开发和生产环境的差异化配置"""
debug: Required[bool]
server: Required[Server]
logging: Required[Logging]
cors: Cors
module_system: ModuleSystem
file_system: FileSystem
class Scheduler(TypedDict):
"""任务调度器配置"""
job_store_path: str
class Cache(TypedDict):
"""缓存系统配置"""
cache_store_path: str
class Config(TypedDict):
"""完整配置结构，整合所有配置项，包含默认配置、数据库、调度器、缓存、安全和环境配置"""
default: Default
database: Database
scheduler: Scheduler
cache: Cache
security: Security
development: Env
production: Env
def load_config(config_path: Path = Path("config.toml")) -> Config:
"""
从TOML配置文件加载配置数据
Args:
config_path: 配置文件路径，默认为当前目录下的config.toml
Returns:
Config: 解析后的配置字典对象
Raises:
FileNotFoundError: 当配置文件不存在时抛出异常
"""
if not config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {config_path}")
with open(config_path, "rb") as f:
return cast(Config, tomllib.load(f))
./utils/worker/src/chongming_worker/__init__.py
"""
chongming_worker - 通用 Worker 框架
提供 Zenoh RPC 端点的装饰器定义、生命周期管理和配置管理。
"""
from chongming_worker.decorators import (
RPCMethod,
rpc_method,
bind_model,
auto_bind_models,
get_rpc_methods,
get_all_rpc_methods,
clear_registry,
generate_route_configs,
)
from chongming_worker.config import (
load_worker_config,
WorkerConfig,
OperationConfig,
)
from chongming_worker.runtime import (
create_worker,
create_workers,
)
__all__ = [
"RPCMethod",
"rpc_method",
"bind_model",
"auto_bind_models",
"get_rpc_methods",
"get_all_rpc_methods",
"clear_registry",
"generate_route_configs",
"load_worker_config",
"WorkerConfig",
"OperationConfig",
"create_worker",
"create_workers",
]
./utils/request/pyproject.toml
[build-system]
build-backend = "hatchling.build"
requires = ["hatchling"]
[project]
name = "chongming-request"
version = "0.1.0"
description = "worker direct communication"
readme = "README.md"
requires-python = ">=3.12"
dependencies = []
[tool.hatch.build.targets.wheel]
packages = ["src/chongming_request"]
./utils/request/README.md
./utils/request/src/chongming_request/__init__.py
./public/config.toml
[default]
app.debug = false
app.description = "基于 FastAPI 的 chongming 打包脚手架"
app.name = "chongming"
app.version = "0.1.0"
env = "production"
prefix = "/api/v1"
upload_path = "uploads"
[database]
generate_schemas = true # 生产环境使用迁移工具
type = "sqlite"
url = "sqlite+aiosqlite:///./data/database.db"
[database.sqlite]
connect_args.check_same_thread = false
connect_args.timeout = 15
echo = false
echo_pool = false
max_overflow = 10
pool_pre_ping = true
pool_recycle = 300
pool_size = 20
pool_timeout = 30
[scheduler]
job_store_path = "scheduler_jobs.db"
[cache]
cache_store_path = "./"
[security]
access_token_expire_minutes = 30
algorithm = "HS256"
max_sessions_per_user = 5
refresh_token_expire_days = 7
secret_key = "your-secret-key-change-in-production"
[production]
debug = false
env = "production"
[production.server]
access_log = false
host = "0.0.0.0"
port = 8000
reload = false
timeout_keep_alive = 65
workers = 4
[production.logging]
backup_count = 5 # 最多保存 10 个日志文件
file = "logs"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
level = "INFO"
max_size = 10485760 # 10MB
[production.cors]
allow_credentials = true
allow_headers = ["*"]
allow_methods = ["*"]
allow_origins = ["http://192.168.137.5:8000", "http://localhost:8000", "http://127.0.0.1:8000"]
[production.module_system]
path = [
"resources/app.mbank",
"resources/plugins.mbank",
]
type = "module-bank"
[production.file_system]
path = "static.svfs"
type = "svfs"
[[workers]]
type = "example"
module = "plugins.example"
factory = "create_workers"
count = 3
config = { type = "example" }
./public/server.py
# server.py
import sys
from utils.launch import init_module_bank
from utils.config import ConfigLoader
# 加载配置文件并初始化模块银行
config = ConfigLoader("config.toml")
init_module_bank(config)
# 从数据库动态导入 app 模块
# 该模块存储在 SQLite 数据库中，通过自定义导入机制加载
try:
import app # type: ignore
app_instance = app.app
except ImportError as e:
print(f"无法从数据库导入 app 模块: {e}")
sys.exit(1)
# 暴露 ASGI 应用实例，供 uvicorn 服务器使用
app = app_instance
./public/main.py
# main.py - 修复 reload 问题
#!/usr/bin/env python3
"""
FastAPI chongming 示例
主入口文件，适配 chongming 打包系统
"""
import os
import sys
import platform
from utils.launch import launch
from multiprocessing import freeze_support
import server
use_shell = platform.system() == "Windows"
def app_run(app_config: dict, default_config: dict):
"""
应用运行函数 - 适配模块银行环境
Args:
app_config: 应用配置字典，包含服务器配置、环境设置等
default_config: 默认配置字典，包含应用名称、版本等基础信息
Returns:
None: 此函数不返回值，直接启动服务器并阻塞运行
"""
from app.core.logger import get_logger # type: ignore
logger = get_logger("app")
# 获取服务器配置
server_config = app_config.get("server", {})
host = server_config.get("host", "0.0.0.0")
port = server_config.get("port", 8000)
timeout_keep_alive = server_config.get("timeout_keep_alive", 5)
# 打包环境强制禁用重载
is_frozen = getattr(sys, "frozen", False)
reload = False if is_frozen else server_config.get("reload", False)
# 支持通过环境变量 APP_WORKERS 覆盖 worker 数量（方便嵌入式设备调优）
default_workers = server_config.get("workers", 4)
workers = int(os.environ.get("APP_WORKERS", str(default_workers)))
# 显示启动信息
env = app_config.get("env", "development")
logger.info("=" * 50)
logger.info(f"🚀 启动 {default_config.get('app.name', 'FastAPI 应用')}")
logger.info(f"📦 版本: {default_config.get('app.version', '1.0.0')}")
logger.info(f"🌍 环境: {env}")
logger.info(f"📍 地址: http://{host}:{port}")
logger.info(f"📚 文档: http://{host}:{port}/docs")
logger.info(f"🔧 调试: {'启用' if app_config.get('debug', False) else '禁用'}")
logger.info(f"🔄 重载: {'启用' if reload else '禁用'}")
logger.info(f"👥 Workers: {workers}")
logger.info("=" * 50)
# 根据操作系统选择不同的服务器启动方式
if use_shell:
import uvicorn
uvicorn.run(
"server:app",
host=host,
port=port,
reload=reload,
workers=workers,
log_level="debug" if app_config.get("debug", False) else "info",
access_log=False,
timeout_keep_alive=timeout_keep_alive,
)
else:
from gunicorn.app.wsgiapp import run
sys.argv = [
"gunicorn",
"server:app",
"-k",
"uvicorn.workers.UvicornWorker",
"-w",
str(workers),
"-b",
f"{host}:{port}",
]
run()
def main():
"""
主函数入口 - 初始化多进程支持并启动应用
Returns:
None: 此函数不返回值，调用launch函数启动应用
"""
freeze_support()
launch(app_run, "production")
if __name__ == "__main__":
main()
./public/utils/config.py
# config_loader.py - 读取config.toml的示例代码
import os
import sys
from pathlib import Path
from typing import Any, Dict, Optional
# 根据Python版本选择合适的TOML解析库
if sys.version_info >= (3, 11):
import tomllib
else:
import tomli as tomllib
class ConfigLoader:
"""TOML配置文件加载器
提供配置文件的加载、访问和管理功能。支持自动查找配置文件、
点号分隔的嵌套键访问、配置节获取以及配置重载等功能。
Attributes:
config_path: 配置文件的路径
_config: 存储解析后的配置数据
"""
def __init__(self, config_path: Optional[str] = None):
"""
初始化配置加载器
Args:
config_path: 配置文件路径，如果为None则自动查找
Raises:
FileNotFoundError: 当无法找到配置文件时抛出异常
"""
if config_path is None:
# 自动查找配置文件
config_path = self._find_config_file()
self.config_path = Path(config_path)
self._config: Dict[str, Any] = {}
self.load()
def _find_config_file(self) -> str:
"""
自动查找配置文件
按照预定义的优先级顺序搜索配置文件：
1. 当前目录下的config.toml
2. config子目录下的config.toml
3. 用户配置目录~/.config/myapp/config.toml
4. 系统配置目录/etc/myapp/config.toml
Returns:
str: 找到的第一个配置文件路径
Raises:
FileNotFoundError: 当所有可能的位置都未找到配置文件时抛出异常
"""
possible_paths = [
"config.toml",
"config/config.toml",
"~/.config/myapp/config.toml",
"/etc/myapp/config.toml",
]
for path in possible_paths:
expanded_path = os.path.expanduser(path)
if os.path.exists(expanded_path):
return expanded_path
raise FileNotFoundError("未找到配置文件")
def load(self) -> None:
"""
加载配置文件
读取并解析TOML格式的配置文件，将配置数据存储到内部字典中。
Raises:
FileNotFoundError: 当配置文件不存在时抛出异常
tomllib.TOMLDecodeError: 当配置文件格式不正确时抛出异常
"""
if not self.config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {self.config_path}")
with open(self.config_path, "r", encoding="utf-8") as f:
self._config = tomllib.loads(f.read())
def get(self, key: str, default: Any = None) -> Any:
"""
获取配置值，支持点号分隔的路径
通过点号分隔的键路径访问嵌套的配置值。例如：
- "database.host" 访问 {"database": {"host": "localhost"}}
- "server.port" 访问 {"server": {"port": 8080}}
Args:
key: 配置键，支持点号分隔的嵌套路径
default: 当键不存在时返回的默认值，默认为None
Returns:
Any: 配置值，如果键不存在则返回默认值
"""
keys = key.split(".")
value = self._config
try:
for k in keys:
value = value[k]
return value
except (KeyError, TypeError):
return default
def get_section(self, section: str) -> Dict[str, Any]:
"""
获取整个配置节
返回指定配置节的完整字典内容，适用于需要访问某个配置节下
所有配置项的场景。
Args:
section: 配置节名称
Returns:
Dict[str, Any]: 配置节的内容字典，如果节不存在则返回空字典
"""
return self._config.get(section, {})
def reload(self) -> None:
"""
重新加载配置文件
从磁盘重新读取配置文件并更新内部配置数据。
适用于配置文件在运行时被修改的场景。
Raises:
FileNotFoundError: 当配置文件不存在时抛出异常
"""
self.load()
def __getitem__(self, key: str) -> Any:
"""
支持字典式访问
允许使用方括号语法访问配置值，如：config["database.host"]
Args:
key: 配置键
Returns:
Any: 配置值
Raises:
KeyError: 当键不存在且无默认值时抛出异常（通过get方法处理）
"""
return self.get(key)
def __contains__(self, key: str) -> bool:
"""
检查配置是否存在
支持使用in操作符检查配置键是否存在，如："database.host" in config
Args:
key: 要检查的配置键
Returns:
bool: 如果配置键存在且值不为None则返回True，否则返回False
"""
return self.get(key) is not None
@property
def all(self) -> Dict[str, Any]:
"""
获取所有配置
返回完整的配置字典，适用于需要遍历或导出所有配置的场景。
Returns:
Dict[str, Any]: 包含所有配置的字典
"""
return self._config
./public/utils/__init__.py
./public/utils/launch.py
from .config import ConfigLoader
import os
from module_bank import PythonToSQLite
from typing import Callable
def init_module_bank(config):
"""
初始化模块银行系统，将指定路径下的Python模块打包并安装导入器。
该函数会验证所有模块路径的有效性，创建PythonToSQLite打包器实例，
列出所有可用模块，并为每个打包器安装导入器以支持模块的动态加载。
Args:
config (dict): 配置字典，必须包含 'production' -> 'module_system' -> 'path' 键，
其值为模块路径列表
Raises:
ValueError: 当配置的模块路径中有任何一个不存在时抛出异常
"""
MODULE_BANK_PATH_LIST = config["production"]["module_system"]["path"]
for module_path in MODULE_BANK_PATH_LIST:
if not os.path.exists(module_path):
raise ValueError(f"模块路径不存在: {module_path}")
# 为每个模块路径创建PythonToSQLite打包器实例
packer_list = [
PythonToSQLite(module_path) for module_path in MODULE_BANK_PATH_LIST
]
# 遍历所有打包器，列出模块信息并安装导入器
for packer in packer_list:
key = packer.get_key()
modules = packer.list_modules()
for module in modules:
print(f"{module['module_name']} {'[包]' if module['is_package'] else ''}")
packer.install_importer(key=key)
def launch(run_fun: Callable[[dict, dict], None], run_mode: str):
"""
启动应用程序，加载配置并执行指定的运行函数。
该函数从config.toml文件中加载配置，根据运行模式获取对应的应用配置和默认配置，
然后调用传入的运行函数执行应用程序逻辑。
Args:
run_fun (Callable[[dict, dict], None]): 运行函数，接收两个参数：
- app_config (dict): 当前运行模式的配置
- default_config (dict): 默认配置
run_mode (str): 运行模式，用于从配置文件中获取对应模式的配置项
Returns:
None
"""
config = ConfigLoader("config.toml")
# 获取对应运行模式的配置
default_config = config.get("default", {})
app_config = config.get(run_mode, {})
run_fun(app_config, default_config)
./data/storage/chongming/.nanio-meta/string/95e51f17-ad95-48d9-a349-e0fe5deba291/k8s部署文档.md.json
{"etag":"\"99bb1226df2b46c13a8b2f099ecb4778\"","size":5756,"last_modified":"2026-07-16T09:55:58.795584+00:00","content_type":"text/markdown","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null}
./data/storage/chongming/.nanio-meta/test/1f7f0e0e-c347-489c-bfb9-1f3e8083303e/k8s部署文档.md.json
{"etag":"\"99bb1226df2b46c13a8b2f099ecb4778\"","size":5756,"last_modified":"2026-07-16T09:51:00.518358+00:00","content_type":"text/markdown","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null}
./data/storage/chongming/.nanio-meta/test/c6b0c0db-c681-4d44-b56a-350366a03866/test.mp4.json
{"etag":"\"cdbfb2788ef3025530e985b2c9ddb850\"","size":918742,"last_modified":"2026-07-16T01:36:13.643254+00:00","content_type":"application/octet-stream","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null}
./data/storage/chongming/.nanio-meta/test/e4269914-e2d1-48c1-875e-b114295943c3/test.mp4.json
{"etag":"\"cdbfb2788ef3025530e985b2c9ddb850\"","size":918742,"last_modified":"2026-07-16T01:34:26.862901+00:00","content_type":"application/octet-stream","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null}
./data/storage/chongming/string/95e51f17-ad95-48d9-a349-e0fe5deba291/k8s部署文档.md
# Kubernetes 部署
## 使用 containerd 配置 Kubernetes
自Kubernetes 1.24 版本起，不再支持 docker 作为容器运行时，因此 1.24 及以后版本默认使 用containerd 作为容器。
推荐使用1.28 及以上Kubernetes，本例程中以 1.28 版本为例。
## 安装 containerd
本节操作需要在所有节点上进行，包括Master节点与Node节点。
1. 卸载设备上原有containerd
```shell
sudo apt-get remove containerd
sudo apt-get remove containerd.io
sudo apt-get remove runc
-
更新 apt 包索引
sudo apt-get update sudo apt-get install ca-certificates curl gnupg lsb-release添加官方GPG密钥：
sudo mkdir -p /etc/apt/keyrings curl-fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg--dearmor-o /etc/apt/ →keyrings/docker.gpg置存储库：
echo "deb [arch=$(dpkg--print-architecture) signedby=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu $(lsb_release-cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null -
安装 containerd
安装最新版本 containerd
sudo apt-get update sudo apt-get install-y containerd.io
安装 Kubernetes
本节操作需要在所有节点上进行，包括Master节点与Node节点
-
卸载设备上已安装Kubernetes
sudo apt-get autoremove kube* -
更新 apt 包索引并安装 Kubernetes 依赖：
sudo apt-get update sudo apt-get install -y apt-transport-https ca-certificates curl -
下载密钥
sudo curl-fsSL https://pkgs.k8s.io/core:/stable:/v1.28/deb/Release.key | sudo gpg--dearmor-o /etc/apt/keyrings/kubernetes-apt-keyring.gpg -
添加 Kubernetes 仓库
sudo echo "deb [signed-by=/etc/apt/keyrings/kubernetes-apt-keyring.gpg] https://pkgs.k8s.io/core:/stable:/v1.28/deb/ /" | sudo tee /etc/apt/sources.list.d/kubernetes.list -
更新 apt 包索引，并安装 Kubernetes 管理工具
sudo apt-get update使用以下指令查看当前可用kubectl 版本
sudo apt-cache madison kubectl安装特定版本的 kubectl，kubeadm，kubectl
sudo apt-get install-y kubelet=1.28.0-1.1 kubeadm=1.28.0-1.1 kubectl=1.28.0-1.1 -
锁定 Kubernetes 版本
sudo apt-mark hold kubelet kubeadm kubectl
主节点
主节点初始化
-
关闭 swap 分区
sudo su swapoff -a free -m -
关闭 SELinux
setenforce 0 -
加载必要的内核模块并设置 sysctl 参数
modprobe overlay
modprobe br_netfilter
确保模块开机自动加载：
cat <<EOF | tee /etc/modules-load.d/k8s.conf
overlay
br_netfilter
EOF
设置 sysctl 参数（包括桥接流量转发和 IP 转发）
cat <<EOF | tee /etc/sysctl.d/k8s.conf
net.bridge.bridge-nf-call-iptables = 1
net.bridge.bridge-nf-call-ip6tables = 1
net.ipv4.ip_forward = 1
EOF
sysctl --system
执行后，/proc/sys/net/bridge/bridge-nf-call-iptables 文件应该会出现。
-
修改 containerd 的 sandbox_images 为阿里云镜像
sed -i 's|sandbox_image = ".*"|sandbox_image = "registry.aliyuncs.com/google_containers/pause:3.9"|' /etc/containerd/config.toml systemctl restart containerd -
初始化 Kubernetes
kubeadm init --service-cidr=10.96.0.0/12 --pod-network-cidr=10.244.0.0/16 --kubernetes-version=1.28.0 --image-repository registry.aliyuncs.com/google_containers
{
"GBDeviceId":"",
"GBTaskChnId":"",
"GPS":{
"KSpeed":0.0,
"NSpeed":0.0,
"angleCourse":0.0,
"available":false,
"latitude":0.0,
"longitude":0.0
},
"alarmId":"dc3e6d26-d825-4a84-af95-a0585ee34294",
"boardId":"183918002357768",
"boardIp":"192.168.150.1",
"dataId":0,
"id":0,
"localLabeledPath":"",
"localRawPath":"",
"media":{
"GBTransport":false,
"mediaDesc":"909",
"mediaHeight":1080,
"mediaName":"418",
"mediaUrl":"rtsp://admin:xm666888@192.168.8.10:554/cam/realmonitor?channel=1&subtype=0",
"mediaWidth":1920,
"rtspTransport":false
},
"result":{
"cropped":false,
"description":"No mask",
"noHelmet":false,
"properties":[],
"relativeBox":[652.8,302.4,1536.0,756.0],
"type":"S6_electrolytic_room"
},
"summary":"mask",
"taskDesc":"",
"taskSession":"89097472",
"time":"2026-07-13 16:14:01",
"timeStamp":1783930441921425,
"type":0,
"uniqueId":"ALARM_dc3e6d26-d825-4a84-af95-a0585ee34294",
"videoFile":"alarm_20260713_161404_355"
}
python3 ~/ppe_pose_delivery/main.py
--input 'rtsp://admin:datagoo6081559@192.168.1.145:554/ch1/main/av_stream'
--bmodel ~/ppe_pose_delivery/model_config.json
--dev_id 0
--taskId '37665280'
--push 'http://dg.datagoo.cn:60014/open-api/event-alarm/receive'
--push_only_alarms --board_id '183918002357767' -
-board_ip '192.168.1.253'
--media_name '418'
--media_url 'rtsp://admin:datagoo6081559@192.168.1.145:554/ch1/main/av_stream'
--media_desc '909-electrolyte-room-1'
--scene_type 'clean_room'
--roi '0.61,0.32,0.80,0.82'
--frame_stride 10
--alarm_votes 3
--alarm_cooldown 10
--log_interval 1
--no_pose
--max_frames 3
`./data/storage/chongming/test/1f7f0e0e-c347-489c-bfb9-1f3e8083303e/k8s部署文档.md`
Kubernetes 部署
使用 containerd 配置 Kubernetes
自Kubernetes 1.24 版本起，不再支持 docker 作为容器运行时，因此 1.24 及以后版本默认使 用containerd 作为容器。
推荐使用1.28 及以上Kubernetes，本例程中以 1.28 版本为例。
安装 containerd
本节操作需要在所有节点上进行，包括Master节点与Node节点。
-
卸载设备上原有containerd
sudo apt-get remove containerd sudo apt-get remove containerd.io sudo apt-get remove runc -
更新 apt 包索引
sudo apt-get update sudo apt-get install ca-certificates curl gnupg lsb-release添加官方GPG密钥：
sudo mkdir -p /etc/apt/keyrings curl-fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg--dearmor-o /etc/apt/ →keyrings/docker.gpg置存储库：
echo "deb [arch=$(dpkg--print-architecture) signedby=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu $(lsb_release-cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null -
安装 containerd
安装最新版本 containerd
sudo apt-get update sudo apt-get install-y containerd.io
安装 Kubernetes
本节操作需要在所有节点上进行，包括Master节点与Node节点
-
卸载设备上已安装Kubernetes
sudo apt-get autoremove kube* -
更新 apt 包索引并安装 Kubernetes 依赖：
sudo apt-get update sudo apt-get install -y apt-transport-https ca-certificates curl -
下载密钥
sudo curl-fsSL https://pkgs.k8s.io/core:/stable:/v1.28/deb/Release.key | sudo gpg--dearmor-o /etc/apt/keyrings/kubernetes-apt-keyring.gpg -
添加 Kubernetes 仓库
sudo echo "deb [signed-by=/etc/apt/keyrings/kubernetes-apt-keyring.gpg] https://pkgs.k8s.io/core:/stable:/v1.28/deb/ /" | sudo tee /etc/apt/sources.list.d/kubernetes.list -
更新 apt 包索引，并安装 Kubernetes 管理工具
sudo apt-get update使用以下指令查看当前可用kubectl 版本
sudo apt-cache madison kubectl安装特定版本的 kubectl，kubeadm，kubectl
sudo apt-get install-y kubelet=1.28.0-1.1 kubeadm=1.28.0-1.1 kubectl=1.28.0-1.1 -
锁定 Kubernetes 版本
sudo apt-mark hold kubelet kubeadm kubectl
主节点
主节点初始化
-
关闭 swap 分区
sudo su swapoff -a free -m -
关闭 SELinux
setenforce 0 -
加载必要的内核模块并设置 sysctl 参数
modprobe overlay
modprobe br_netfilter
确保模块开机自动加载：
cat <<EOF | tee /etc/modules-load.d/k8s.conf
overlay
br_netfilter
EOF
设置 sysctl 参数（包括桥接流量转发和 IP 转发）
cat <<EOF | tee /etc/sysctl.d/k8s.conf
net.bridge.bridge-nf-call-iptables = 1
net.bridge.bridge-nf-call-ip6tables = 1
net.ipv4.ip_forward = 1
EOF
sysctl --system
执行后，/proc/sys/net/bridge/bridge-nf-call-iptables 文件应该会出现。
-
修改 containerd 的 sandbox_images 为阿里云镜像
sed -i 's|sandbox_image = ".*"|sandbox_image = "registry.aliyuncs.com/google_containers/pause:3.9"|' /etc/containerd/config.toml systemctl restart containerd -
初始化 Kubernetes
kubeadm init --service-cidr=10.96.0.0/12 --pod-network-cidr=10.244.0.0/16 --kubernetes-version=1.28.0 --image-repository registry.aliyuncs.com/google_containers
{
"GBDeviceId":"",
"GBTaskChnId":"",
"GPS":{
"KSpeed":0.0,
"NSpeed":0.0,
"angleCourse":0.0,
"available":false,
"latitude":0.0,
"longitude":0.0
},
"alarmId":"dc3e6d26-d825-4a84-af95-a0585ee34294",
"boardId":"183918002357768",
"boardIp":"192.168.150.1",
"dataId":0,
"id":0,
"localLabeledPath":"",
"localRawPath":"",
"media":{
"GBTransport":false,
"mediaDesc":"909",
"mediaHeight":1080,
"mediaName":"418",
"mediaUrl":"rtsp://admin:xm666888@192.168.8.10:554/cam/realmonitor?channel=1&subtype=0",
"mediaWidth":1920,
"rtspTransport":false
},
"result":{
"cropped":false,
"description":"No mask",
"noHelmet":false,
"properties":[],
"relativeBox":[652.8,302.4,1536.0,756.0],
"type":"S6_electrolytic_room"
},
"summary":"mask",
"taskDesc":"",
"taskSession":"89097472",
"time":"2026-07-13 16:14:01",
"timeStamp":1783930441921425,
"type":0,
"uniqueId":"ALARM_dc3e6d26-d825-4a84-af95-a0585ee34294",
"videoFile":"alarm_20260713_161404_355"
}
python3 ~/ppe_pose_delivery/main.py
--input 'rtsp://admin:datagoo6081559@192.168.1.145:554/ch1/main/av_stream'
--bmodel ~/ppe_pose_delivery/model_config.json
--dev_id 0
--taskId '37665280'
--push 'http://dg.datagoo.cn:60014/open-api/event-alarm/receive'
--push_only_alarms --board_id '183918002357767' -
-board_ip '192.168.1.253'
--media_name '418'
--media_url 'rtsp://admin:datagoo6081559@192.168.1.145:554/ch1/main/av_stream'
--media_desc '909-electrolyte-room-1'
--scene_type 'clean_room'
--roi '0.61,0.32,0.80,0.82'
--frame_stride 10
--alarm_votes 3
--alarm_cooldown 10
--log_interval 1
--no_pose
--max_frames 3
`./config/workers/config.toml`
workers type = "example" module = "plugins.example" factory = "create_workers" count = 3 config = { type = "example" }
`./config/workers/example/multiply.toml`
multiply 接口
[GET] summary = "Multiply two numbers" docstring = "Multiply two numbers via GET query params (a * b)" params = ["a: float", "b: float"] response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "multiply" }, timestamp = { type = "float", required = false, default = 0.0 } }
`./config/workers/example/divide.toml`
divide 接口
[GET] summary = "Divide two numbers" docstring = "Divide two numbers via GET query params (a / b). Returns error if b == 0." params = ["a: float", "b: float"] response_model = { result = { type = "float", required = false, default = 0.0 }, operation = { type = "str", default = "divide" }, timestamp = { type = "float", required = false, default = 0.0 }, error = { type = "str", required = false, default = "" } }
`./config/workers/example/subtract.toml`
subtract 接口
[GET] summary = "Subtract two numbers" docstring = "Subtract two numbers via GET query params (a - b)" params = ["a: float", "b: float"] response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "subtract" }, timestamp = { type = "float", required = false, default = 0.0 } }
`./config/workers/example/add.toml`
add 接口 — 按请求类型分节
subject 会带上方法后缀: example.add.GET / example.add.POST
[GET] cacheable = true summary = "Add two numbers" docstring = "Add two numbers via GET query params" params = ["a: float", "b: float"] response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "add" }, timestamp = { type = "float", required = false, default = 0.0 } }
[POST] cacheable = true summary = "Add two numbers (POST)" docstring = "Add two numbers via POST JSON body" params = ["a: float", "b: float"] response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "add" }, timestamp = { type = "float", required = false, default = 0.0 } }
`./config/workers/example/config.toml`
[worker] name = "example" version = "0.1.0" description = "chongming worker example — demonstrates new config features" config_version = "1.0" defaults = { method = "GET", ttl = 30, timeout = 2.0, auth_required = true, cacheable = false }
[worker.registration] type = "register" service = "example" worker_type = "calc-workers" router_prefix = "/calc" tags = ["example"]
`./src/plugins/__init__.py`
`./src/plugins/example/__init__.py`
""" Example Worker 插件 — 演示 @rpc_method + 显式模型绑定用法
本插件展示如何使用 chongming_worker 框架定义 RPC 端点。
核心工作流:
- 编写 config/workers/example/*.toml 配置文件
- 执行 uv run gen_models 生成 Pydantic 模型
- 在 __init__.py 中用 @rpc_method 定义业务函数
- 显式导入模型并绑定到对应端点（替代 auto_bind_models）
关键约定:
- 模型命名: {Request|Response} (如 AddGetRequest, AddGetResponse 由 gen_models 自动生成)
- 装饰器 subject 中的操作名 → 模型前缀（驼峰） example.add → Add, example.subtract → Subtract """ import time
from chongming_worker import rpc_method, auto_bind_models from chongming_worker.runtime import create_workers as _create_workers
============================================================
显式导入 Pydantic 模型（避免 PyArmor 打包后 importlib 动态导入异常）
============================================================
from .models.pydantic import ( AddGetRequest, AddGetResponse, AddPostRequest, AddPostResponse, SubtractGetRequest, SubtractGetResponse, MultiplyGetRequest, MultiplyGetResponse, DivideGetRequest, DivideGetResponse, )
============================================================
注册 RPC 端点
============================================================
直接在 @rpc_method 中指定 response_model（和可选的 request_model），
不再依赖 auto_bind_models 进行动态导入绑定。
这样 PyArmor 打包后不会因 importlib.import_module 路径解析失败而出问题。
@rpc_method( subject="example.add", method="GET", summary="Add two numbers via GET", response_model=AddGetResponse, ) def add(add_get_request:AddGetRequest) -> AddGetResponse: """加法运算（GET）— 通过查询参数传入 a、b，返回它们的和""" return AddGetResponse( result=add_get_request.a + add_get_request.b, operation="add", timestamp=time.time() )
@rpc_method( subject="example.add", method="POST", summary="Add two numbers via POST", response_model=AddPostResponse, ) def add_post(add_post_request: AddPostRequest) -> AddPostResponse: """加法运算（POST）— 通过 JSON Body 传入 a、b，返回它们的和""" return AddPostResponse( result=add_post_request.a + add_post_request.b, operation="add", timestamp=time.time(), )
@rpc_method( subject="example.subtract", method="GET", summary="Subtract two numbers", response_model=SubtractGetResponse, ) def subtract(subtract_get_request: SubtractGetRequest) -> SubtractGetResponse: """减法运算：a - b""" return SubtractGetResponse( result=subtract_get_request.a - subtract_get_request.b, operation="subtract", timestamp=time.time(), )
@rpc_method( subject="example.multiply", method="GET", summary="Multiply two numbers", response_model=MultiplyGetResponse, ) def multiply(multiply_get_request=MultiplyGetRequest) -> MultiplyGetResponse: """乘法运算：a * b""" return MultiplyGetResponse( result=multiply_get_request.a * multiply_get_request.b, operation="multiply", timestamp=time.time(), )
@rpc_method( subject="example.divide", method="GET", summary="Divide two numbers", response_model=DivideGetResponse, ) def divide(divide_get_request: DivideGetRequest) -> DivideGetResponse: """除法运算：a / b（b 不能为 0）""" if divide_get_request.b == 0: return DivideGetResponse( error="除数不能为零", operation="divide", result=0.0, timestamp=time.time(), ) return DivideGetResponse( result=divide_get_request.a / divide_get_request.b, operation="divide", timestamp=time.time(), )
============================================================
自动绑定 Pydantic 模型（作为备用，但在显式绑定后实际不再需要）
============================================================
保留 auto_bind_models 调用作为 fallback，确保任何未显式指定模型的端点
也能自动匹配到合适的模型。由于大部分端点已在 @rpc_method 中显式指定，
auto_bind_models 会跳过已绑定的端点。
auto_bind_models("example")
============================================================
工厂函数 — 供 app lifespan 动态加载
============================================================
框架通过 import + getattr 获取此函数，
然后调用 create_workers(session, config, count, worker_manager=worker_manager)。
详见 src/chongming/app/__init__.py 的 lifespan。
def create_workers(session, config, count=1, worker_manager=None): """插件入口 — 委托给 chongming_worker 运行时创建 Worker 协程""" return _create_workers(session, config, count, worker_manager)
`./src/plugins/example/models/__init__.py`
""" 模型导出 — example 自动生成，请勿手动修改 """
from .pydantic import AddGetRequest from .pydantic import AddGetResponse from .pydantic import AddPostRequest from .pydantic import AddPostResponse from .pydantic import DivideGetRequest from .pydantic import DivideGetResponse from .pydantic import ExampleResponse from .pydantic import MultiplyGetRequest from .pydantic import MultiplyGetResponse from .pydantic import SubtractGetRequest from .pydantic import SubtractGetResponse
all = [ "AddGetRequest", "AddGetResponse", "AddPostRequest", "AddPostResponse", "DivideGetRequest", "DivideGetResponse", "ExampleResponse", "MultiplyGetRequest", "MultiplyGetResponse", "SubtractGetRequest", "SubtractGetResponse", ]
`./src/plugins/example/models/pydantic.py`
""" 自动生成的 Pydantic 模型 — example 来源: config/workers/example/ """
from future import annotations
from typing import Any, Optional from pydantic import BaseModel, Field
class AddGetRequest(BaseModel): """Add two numbers — 请求体""" a: float = Field(...) b: float = Field(...)
class AddGetResponse(BaseModel): """Add two numbers — 响应体""" result: float = Field(...) operation: str = Field(default='add') timestamp: Optional[float] = Field(default=0.0)
class AddPostRequest(BaseModel): """Add two numbers (POST) — 请求体""" a: float = Field(...) b: float = Field(...)
class AddPostResponse(BaseModel): """Add two numbers (POST) — 响应体""" result: float = Field(...) operation: str = Field(default='add') timestamp: Optional[float] = Field(default=0.0)
class DivideGetRequest(BaseModel): """Divide two numbers — 请求体""" a: float = Field(...) b: float = Field(...)
class DivideGetResponse(BaseModel): """Divide two numbers — 响应体""" result: Optional[float] = Field(default=0.0) operation: str = Field(default='divide') timestamp: Optional[float] = Field(default=0.0) error: Optional[str] = Field(default='')
class MultiplyGetRequest(BaseModel): """Multiply two numbers — 请求体""" a: float = Field(...) b: float = Field(...)
class MultiplyGetResponse(BaseModel): """Multiply two numbers — 响应体""" result: float = Field(...) operation: str = Field(default='multiply') timestamp: Optional[float] = Field(default=0.0)
class SubtractGetRequest(BaseModel): """Subtract two numbers — 请求体""" a: float = Field(...) b: float = Field(...)
class SubtractGetResponse(BaseModel): """Subtract two numbers — 响应体""" result: float = Field(...) operation: str = Field(default='subtract') timestamp: Optional[float] = Field(default=0.0)
所有响应模型的联合类型
ExampleResponse = MultiplyGetResponse | AddGetResponse | SubtractGetResponse | AddPostResponse | DivideGetResponse
`./src/chongming/__init__.py`
import uvicorn from .app.core.config import load_config from .app.core.logger import get_logger
def serve(): """ 使用 uvicorn 启动服务
从配置文件中加载服务器配置和日志配置，然后使用 uvicorn 启动应用服务。
该函数适用于开发环境或单进程部署场景。
Returns:
None: 该函数不返回值，会阻塞运行直到服务停止
"""
config = load_config()
env = config["default"]["env"]
server_config = config[env]["server"]
log_config = config[env]["logging"]
logger = get_logger("app")
logger.info(f"启动服务: {server_config}")
logger.info(f"启动日志: {log_config}")
uvicorn.run("chongming.app:app", **server_config)
def gunicorn_serve(): """ 使用 gunicorn 启动服务
通过 gunicorn 作为进程管理器启动应用，使用 UvicornWorker 作为 worker 类型，
支持多进程部署，适用于生产环境。
从配置文件中读取服务器配置，动态设置 gunicorn 的 worker 数量、绑定地址和端口等参数。
Returns:
None: 该函数不返回值，会阻塞运行直到服务停止
"""
import sys
from gunicorn.app.wsgiapp import run
config = load_config()
env = config["default"]["env"]
server_config = config[env]["server"]
# 构建 gunicorn 命令行参数并启动服务
sys.argv = [
"gunicorn",
"chongming.app:app",
"-k",
"uvicorn.workers.UvicornWorker",
"-w",
f"{server_config.get('workers', 4)}",
"-b",
f"{server_config.get('host', '0.0.0.0')}:{server_config.get('port', 8000)}",
]
run()
if name == "main": serve()
`./src/chongming/scripts/code_content.py`
import os import sys
def generate_markdown_from_py_files(directory, output_file):
with open(output_file, "w", encoding="utf-8") as md_file:
for root, dirs, files in os.walk(directory):
# 排除 venv 目录
dirs[:] = [d for d in dirs if d != ".venv"]
dirs[:] = [d for d in dirs if d != ".vscode"]
dirs[:] = [d for d in dirs if d != "node_modules"]
dirs[:] = [d for d in dirs if d != "build"]
dirs[:] = [d for d in dirs if d != "dist"]
dirs[:] = [d for d in dirs if d != "chongming-web"]
for file in files:
if (
file.endswith(".py")
or file.endswith(".rs")
or file.endswith(".toml")
or file.endswith(".md")
or file.endswith(".css")
or file.endswith(".vue")
or file.endswith(".ts")
or file.endswith(".json")
or file.endswith(".html")
or file.endswith(".sh")
or file == "Dockerfile"
):
file_path = os.path.join(root, file)
if file == "output.md" or file == "package-lock.json":
continue
if "generated" in file_path:
continue
md_file.write(f"{file_path}\n")
md_file.write("\n") with open(file_path, "r", encoding="utf-8") as py_file: md_file.write(py_file.read()) md_file.write("\n\n\n")
if name == "main": # 指定目录和输出文件名 target_directory = sys.argv[1] # 替换为你的目标目录 output_markdown_file = "output.md" # 输出的 Markdown 文件名
generate_markdown_from_py_files(target_directory, output_markdown_file)
print(f"Markdown 文件已生成：{output_markdown_file}")
`./src/chongming/scripts/gen_models.py`
""" 模型生成脚本 — 从 config/workers/<plugin_name>/*.toml 生成 Pydantic 模型
用法: uv run gen_models config/workers/example src/plugins/example
工作流: 1. 编写 config/workers/<plugin_name>/*.toml 配置文件 2. 执行 gen_models 生成 Pydantic 模型 3. 在业务代码中 import 生成的模型
功能: 1. 扫描配置目录下的所有 .toml 文件 2. 解析每个路由的 params 和 response_model 3. 生成 Pydantic 请求/响应模型 4. 写入 src/plugins/<plugin_name>/models/__init__.py 和 models/pydantic.py """
import sys import json from pathlib import Path from typing import Any
def parse_type_str(type_str: str) -> str: """将 TOML 中的类型字符串映射为 Python 类型注解""" mapping = { "float": "float", "int": "int", "str": "str", "bool": "bool", "string": "str", "integer": "int", "number": "float", "boolean": "bool", "any": "Any", }
if type_str.startswith("list[") and type_str.endswith("]"):
inner = type_str[5:-1]
inner_py = mapping.get(inner, inner)
return f"list[{inner_py}]"
if type_str == "dict":
return "dict[str, Any]"
if type_str.startswith("dict[") and type_str.endswith("]"):
inner = type_str[5:-1]
if "," in inner:
parts = [p.strip() for p in inner.split(",", 1)]
inner_py = mapping.get(parts[-1], parts[-1])
else:
inner_py = mapping.get(inner, inner)
return f"dict[str, {inner_py}]"
if type_str.startswith("Optional[") and type_str.endswith("]"):
inner = type_str[9:-1]
inner_py = mapping.get(inner, inner)
return f"Optional[{inner_py}]"
return mapping.get(type_str, type_str)
def parse_params(params: list[str]) -> list[dict[str, str]]: """解析 params 列表，如 ['a: float', 'b: float']""" result = [] for p in params: if ":" in p: name, type_str = p.split(":", 1) result.append({"name": name.strip(), "type": parse_type_str(type_str.strip())}) else: result.append({"name": p.strip(), "type": "str"}) return result
def parse_response_model(response_model: dict[str, Any]) -> list[dict[str, Any]]: """解析 response_model 字典，返回字段列表""" fields: list[dict[str, Any]] = [] for field_name, field_spec in response_model.items(): entry: dict[str, Any] = {"name": field_name} if isinstance(field_spec, dict): entry["type"] = parse_type_str(field_spec.get("type", "str")) entry["required"] = field_spec.get("required", True) if "default" in field_spec: entry["default"] = field_spec["default"] else: entry["type"] = parse_type_str(str(field_spec)) entry["required"] = True fields.append(entry) return fields
def model_name(base: str, method: str, suffix: str) -> str: """生成模型名: e.g. add + GET + Request -> AddGetRequest""" return f"{base.title().replace('-', '').replace('', '')}{method.title()}{suffix}"
def generate_pydantic_code( plugin_name: str, routes: list[dict[str, Any]], ) -> str: """生成 pydantic.py 文件内容""" lines = [] lines.append('"""') lines.append(f"自动生成的 Pydantic 模型 — {plugin_name}") lines.append(f"来源: config/workers/{plugin_name}/") lines.append('"""') lines.append("") lines.append("from future import annotations") lines.append("") lines.append("from typing import Any, Optional") lines.append("from pydantic import BaseModel, Field") lines.append("")
seen_models: set[str] = set()
for route in routes:
route_name = route["name"]
method = route["method"]
summary = route.get("summary", "")
# --- Request Model ---
params = route.get("parsed_params", [])
if params:
model_name = _model_name(route_name, method, "Request")
if model_name not in seen_models:
seen_models.add(model_name)
lines.append(f"class {model_name}(BaseModel):")
lines.append(f' """{summary} — 请求体"""')
for p in params:
lines.append(f" {p['name']}: {p['type']} = Field(...)")
lines.append("")
# --- Response Model ---
response_fields = route.get("parsed_response", [])
if response_fields:
model_name = _model_name(route_name, method, "Response")
if model_name not in seen_models:
seen_models.add(model_name)
lines.append(f"class {model_name}(BaseModel):")
lines.append(f' """{summary} — 响应体"""')
for f in response_fields:
field_type = f["type"]
if f.get("required", True):
if "default" in f:
default_repr = (
json.dumps(f["default"])
if not isinstance(f["default"], str)
else f"'{f['default']}'"
)
lines.append(f" {f['name']}: {field_type} = Field(default={default_repr})")
else:
lines.append(f" {f['name']}: {field_type} = Field(...)")
else:
if "default" in f:
default_repr = (
json.dumps(f["default"])
if not isinstance(f["default"], str)
else f"'{f['default']}'"
)
lines.append(f" {f['name']}: Optional[{field_type}] = Field(default={default_repr})")
else:
lines.append(f" {f['name']}: Optional[{field_type}] = Field(default=None)")
lines.append("")
# --- Union type for all responses ---
response_models = [m for m in seen_models if m.endswith("Response")]
if response_models:
plugin_title = plugin_name.title().replace("-", "").replace("_", "")
lines.append("")
lines.append("# 所有响应模型的联合类型")
lines.append(f"{plugin_title}Response = " + " | ".join(response_models))
lines.append("")
return "\n".join(lines)
def generate_init_code( plugin_name: str, routes: list[dict[str, Any]], ) -> str: """生成 __init__.py 文件内容""" lines = [] lines.append('"""') lines.append(f"模型导出 — {plugin_name}") lines.append("自动生成，请勿手动修改") lines.append('"""') lines.append("")
seen_models: set[str] = set()
for route in routes:
rname = route["name"]
method = route["method"]
if route.get("parsed_params"):
seen_models.add(_model_name(rname, method, "Request"))
if route.get("parsed_response"):
seen_models.add(_model_name(rname, method, "Response"))
response_models = [m for m in seen_models if m.endswith("Response")]
if response_models:
plugin_title = plugin_name.title().replace("-", "").replace("_", "")
seen_models.add(f"{plugin_title}Response")
for model_name in sorted(seen_models):
lines.append(f"from .pydantic import {model_name}")
lines.append("")
lines.append("__all__ = [")
for model_name in sorted(seen_models):
lines.append(f' "{model_name}",')
lines.append("]")
lines.append("")
return "\n".join(lines)
def scan_config_dir(config_dir: Path) -> list[dict[str, Any]]: """扫描配置目录，解析所有 .toml 文件中的路由定义""" import tomli
routes = []
toml_files = sorted(config_dir.glob("*.toml"))
group_config = {}
config_file = config_dir / "config.toml"
if config_file.exists():
with open(config_file, "rb") as f:
group_config = tomli.load(f)
defaults = group_config.get("worker", {}).get("defaults", {})
router_prefix = (
group_config.get("worker", {})
.get("registration", {})
.get("router_prefix", "")
)
for toml_file in toml_files:
if toml_file.name == "config.toml":
continue
with open(toml_file, "rb") as f:
data = tomli.load(f)
route_name = toml_file.stem
for method in ("GET", "POST", "PUT", "DELETE", "PATCH"):
if method not in data:
continue
entry = data[method]
params_raw = entry.get("params", [])
response_raw = entry.get("response_model", {})
route = {
"name": route_name,
"method": method,
"summary": entry.get("summary", ""),
"docstring": entry.get("docstring", ""),
"params": params_raw,
"response_model": response_raw,
"parsed_params": parse_params(params_raw),
"parsed_response": parse_response_model(response_raw),
"path": entry.get("path", f"{router_prefix}/{route_name}"),
"timeout": entry.get("timeout", defaults.get("timeout", 3.0)),
"auth_required": entry.get("auth_required", defaults.get("auth_required", False)),
}
routes.append(route)
return routes
def main(): if len(sys.argv) < 3: print("用法: uv run gen_models <config_dir> <plugin_dir>") print("示例:") print(" uv run gen_models config/workers/example src/plugins/example") sys.exit(1)
config_dir = Path(sys.argv[1])
plugin_dir = Path(sys.argv[2])
plugin_name = plugin_dir.name
if not config_dir.exists():
print(f"❌ 配置目录不存在: {config_dir}")
sys.exit(1)
if not plugin_dir.exists():
print(f"❌ 插件目录不存在: {plugin_dir}")
sys.exit(1)
print(f"🔍 扫描配置目录: {config_dir}")
routes = scan_config_dir(config_dir)
print(f" 发现 {len(routes)} 个路由定义")
if not routes:
print("⚠️ 未找到路由定义，跳过模型生成")
return
models_dir = plugin_dir / "models"
models_dir.mkdir(parents=True, exist_ok=True)
pydantic_code = generate_pydantic_code(plugin_name, routes)
pydantic_file = models_dir / "pydantic.py"
pydantic_file.write_text(pydantic_code, encoding="utf-8")
print(f"✅ 生成: {pydantic_file}")
init_code = generate_init_code(plugin_name, routes)
init_file = models_dir / "__init__.py"
init_file.write_text(init_code, encoding="utf-8")
print(f"✅ 生成: {init_file}")
print(f"\n📋 生成的模型:")
for route in routes:
method = route["method"]
name = route["name"]
params = route.get("parsed_params", [])
response = route.get("parsed_response", [])
if params:
req_model = _model_name(name, method, "Request")
param_strs = [f"{p['name']}: {p['type']}" for p in params]
print(f" 📥 {req_model}: {', '.join(param_strs)}")
if response:
resp_model = _model_name(name, method, "Response")
field_strs = [f"{f['name']}: {f['type']}" for f in response]
print(f" 📤 {resp_model}: {', '.join(field_strs)}")
def gen_models(): """CLI 入口""" main()
if name == "main": main()
`./src/chongming/scripts/build.py`
""" 构建脚本 - 将 chongming 应用打包为可执行文件
功能阶段:
- 模块打包(module-bank): 加密打包 Python 模块
- 资源准备: 创建 build 目录, 复制必要文件
- 混淆与打包: pyarmor 混淆 + PyInstaller/Nuitka 打包
- 数据库初始化: 创建表结构和默认数据
- 前端构建: 编译 Vue 项目并打包为 VFS
用法: python -m chongming.scripts.build [pyinstaller|nuitka] """
import json import ast import platform import shutil import subprocess import sys import asyncio import time from dataclasses import dataclass, field from pathlib import Path
from module_bank import PythonToSQLite from module_bank.encryption import Encryption from sqlite_vfs.core import SQLiteVFS from sqlite_vfs.folder_packer import FolderPacker from ..app.core.config import get_config, Config
── 常量 ────────────────────────────────────────────────────────────────────
PYINSTALLER = "pyinstaller" NUITKA = "nuitka" MODES = (PYINSTALLER, NUITKA)
is_windows = platform.system() == "Windows"
── 构建结果数据结构 ─────────────────────────────────────────────────────────
@dataclass class BuildStepResult: """单个构建步骤的执行结果"""
name: str # 步骤名称
success: bool # 是否成功
duration: float # 耗时(秒)
message: str = "" # 附加信息(如包大小)
@dataclass class BuildReport: """构建总结报告"""
mode: str
start_time: float
end_time: float = 0.0
steps: list[BuildStepResult] = field(default_factory=list)
@property
def total_duration(self) -> float:
return self.end_time - self.start_time if self.end_time else 0.0
@property
def all_success(self) -> bool:
return all(s.success for s in self.steps)
def add_step(self, name: str, success: bool, duration: float, message: str = ""):
self.steps.append(BuildStepResult(name, success, duration, message))
def print_summary(self):
"""打印结构化的构建总结"""
sep = "=" * 60
print(f"\n{sep}")
print(f" 🏗️ CHONGMING 构建报告")
print(f"{sep}")
print(f" 构建模式 : {self.mode}")
print(f" 总耗时 : {self.total_duration:.2f}s")
print(f" 构建状态 : {'✅ 成功' if self.all_success else '❌ 失败'}")
print(f"{sep}")
print(f" 步骤明细:")
for i, step in enumerate(self.steps, 1):
icon = "✅" if step.success else "❌"
msg = f" ({step.message})" if step.message else ""
print(
f" {icon} [{i}/{len(self.steps)}] {step.name:<20s} {step.duration:>7.2f}s{msg}"
)
print(f"{sep}\n")
── 工具函数 ────────────────────────────────────────────────────────────────
def _run_cmd( cmd: list[str], cwd: str | None = None, capture: bool = True, shell: bool | None = None, ) -> subprocess.CompletedProcess: """统一执行外部命令""" if shell is None: shell = is_windows return subprocess.run( cmd, cwd=cwd, capture_output=capture, text=True, shell=shell, encoding="utf-8", errors="replace", )
def _format_size(path: str | Path) -> str: """获取文件可读大小""" size = Path(path).stat().st_size for unit in ("B", "KB", "MB", "GB"): if size < 1024: return f"{size:.2f} {unit}" size /= 1024.0 return f"{size:.2f} TB"
def _ensure_dir(path: str | Path): """确保目录存在""" Path(path).mkdir(parents=True, exist_ok=True)
def _resolve_mode() -> str: """解析构建模式""" mode = PYINSTALLER if len(sys.argv) >= 2: mode = sys.argv[1].lower() if mode not in MODES: print(f" ❌ 不支持的构建模式: {mode}，可选: {', '.join(MODES)}") sys.exit(1) return mode
── 阶段 1: 模块打包 (module-bank) ──────────────────────────────────────────
def _stage_pack_modules(config: Config) -> str: """ 使用 module-bank 加密打包 Python 模块为 .mbank 文件。 返回打包过程中的提示信息。 """ _ensure_dir("resources") mbank_list = config.get("production", {}).get("module_system", {}).get("path", []) if not mbank_list: return "⚠️ 未配置 module_system.path，跳过模块打包"
messages = []
for output_path, source_dir, package_name in mbank_list:
key = Encryption.generate_key()
packer = PythonToSQLite(
output_path,
key,
pyarmor_obfuscate=True,
)
packer.pack_directory(source_dir, package_name)
packer.verify_package_structure()
packer.packer_key(key)
packer.delete_source_code(None)
size = _format_size(output_path)
messages.append(f"{Path(output_path).name} ({size})")
return " - " + "\n - ".join(messages)
def _stage_prepare_build_dir(config: Config, mode: str) -> str: """准备 build 目录并复制必要文件""" build_path = Path("build") if build_path.exists(): shutil.rmtree(str(build_path)) _ensure_dir("build")
# 复制 utils
utils_src = Path("public/utils")
if utils_src.exists():
shutil.copytree(utils_src, build_path / "utils")
# 复制 resources
resources_src = Path("resources")
if resources_src.exists():
shutil.copytree(resources_src, build_path / "resources")
resources_src = Path("config")
if resources_src.exists():
shutil.copytree(resources_src, build_path / "config")
# 复制核心文件
for f in ("main.py", "server.py", "config.toml"):
shutil.copy(f"public/{f}", f"build/{f}")
# 注入 JWT 密钥
jwt_key = Encryption.generate_key()
cfg_path = build_path / "config.toml"
content = cfg_path.read_text(encoding="utf-8")
content = content.replace("your-secret-key-change-in-production", jwt_key)
cfg_path.write_text(content, encoding="utf-8")
return "资源准备完成"
── 阶段 3: PyArmor 混淆 ───────────────────────────────────────────────────
def _stage_pyarmor_obfuscate() -> str: """对关键脚本执行 PyArmor 混淆""" files = [ ("main.py", "./"), ("server.py", "./"), ("utils/launch.py", "./utils"), ("utils/config.py", "./utils"), ] failed = [] for filename, output in files: result = _run_cmd( ["pyarmor", "gen", f"--output={output}", filename], cwd="build", ) if result.returncode != 0: failed.append(filename) if failed: return f"❌ 混淆失败: {', '.join(failed)}" return "✅ 所有文件混淆成功"
── 阶段 4: PyInstaller 打包 ────────────────────────────────────────────────
def _stage_pyinstaller(config: Config) -> tuple[str, str]: """执行 PyInstaller 打包，返回 (状态消息, 可执行文件路径)""" base_imports = config.get("pyinstaller_build_import", {}) build_imports = list(base_imports.get("base", [])) build_imports.extend( base_imports.get("windows", []) if is_windows else base_imports.get("linux", []) )
icon = Path("../../public/chongming.ico")
cmd = [
sys.executable,
"-m",
"PyInstaller",
"--onefile",
"--noconfirm",
"--optimize",
"2",
"--name",
"chongming",
"--distpath",
".",
"--workpath",
"./build_temp",
"--specpath",
"./specs",
"--clean",
]
if not is_windows:
cmd.append("--strip")
if icon.exists():
cmd.extend(["--icon", str(icon)])
for imp in build_imports:
cmd.extend(["--hidden-import", imp])
cmd.append("main.py")
result = _run_cmd(cmd, cwd="build", capture=True)
if result.returncode != 0:
return f"❌ PyInstaller 失败", ""
exe_path = Path("build/chongming")
if is_windows:
exe_path = exe_path.with_suffix(".exe")
suffix = ".exe" if is_windows else ""
exe = Path("build") / f"chongming{suffix}"
if exe.exists():
size = _format_size(exe)
return f"✅ PyInstaller 打包成功 ({size})", str(exe)
return "✅ PyInstaller 打包完成 (但未找到可执行文件)", ""
── 阶段 5: Nuitka 打包 ────────────────────────────────────────────────────
def _stage_nuitka(config: Config) -> tuple[str, str]: """执行 Nuitka 打包，返回 (状态消息, 可执行文件路径)""" base_imports = config.get("pyinstaller_build_import", {}).get("base", []) extra = config.get("pyinstaller_build_import", {}).get( "windows" if is_windows else "linux", [] )
packages = [
"module_bank",
"sqlite_vfs",
"fastapi",
"uvicorn",
"gunicorn",
"diskcache",
"apscheduler",
"tomli",
"typing_extensions",
]
cmd = [
sys.executable,
"-m",
"nuitka",
"--standalone",
"--onefile",
"--follow-imports",
"--static-libpython=no",
"--remove-output",
"--assume-yes-for-downloads",
"--lto=yes",
"--disable-plugin=tk-inter",
"--disable-plugin=numpy",
"--output-dir=.",
"--output-filename=chongming",
"main.py",
"--jobs=2",
]
if shutil.which("upx"):
cmd.append("--enable-plugin=upx")
for pkg in packages:
cmd.append(f"--include-package={pkg}")
for mod in list(base_imports) + extra:
cmd.append(f"--include-module={mod}")
if is_windows:
icon = Path("../../public/chongming.ico")
if icon.exists():
cmd.append(f"--windows-icon-from-ico={icon}")
try:
_run_cmd(cmd, cwd="build", capture=False)
except subprocess.CalledProcessError:
# 降级重试（移除可能不兼容的参数）
fallback = [
arg for arg in cmd if arg not in ("--lto=yes", "--enable-plugin=upx")
]
_run_cmd(fallback, cwd="build", capture=False)
exe = Path("build/chongming")
if exe.exists():
# 后处理瘦身
if not is_windows:
subprocess.run(["strip", str(exe)], check=False)
if shutil.which("upx"):
subprocess.run(["upx", "--best", str(exe)], check=False)
size = _format_size(exe)
return f"✅ Nuitka 编译完成 ({size})", str(exe)
return "❌ Nuitka 编译失败 (未生成可执行文件)", ""
── 阶段 6: 清理临时文件 ────────────────────────────────────────────────────
def _stage_cleanup(mode: str) -> str: """清理构建过程中的临时文件""" build = Path("build") dirs_to_remove = [ "build_temp", "specs", "utils", "pyarmor_runtime_000000", "pycache", ] files_to_remove = ["main.py", "server.py"]
for d in dirs_to_remove:
target = build / d
if target.exists():
shutil.rmtree(str(target), ignore_errors=True)
for f in files_to_remove:
target = build / f
if target.exists():
target.unlink(missing_ok=True)
return "临时文件已清理"
── 阶段 8: Vue 前端构建 ────────────────────────────────────────────────────
def _stage_build_vue() -> str: """构建 Vue 前端并打包为 SQLite VFS""" dist_dir = "./src/chongming-web/dist" output_db = "./build/static.svfs" fs_name = "frontend"
# 1. npm build
result = _run_cmd(
["npm", "run", "build"],
cwd="./src/chongming-web",
capture=True,
)
if result.returncode != 0:
err = (result.stderr or "").strip()[:200]
return f"❌ Vue 构建失败: {err}"
# 2. 检查产物
if not Path(dist_dir).exists():
return "❌ Vue 构建成功但未找到 dist 目录"
# 3. 打包为 VFS
vfs = SQLiteVFS(output_db, compress=False)
packer = FolderPacker(sqlite_vfs=vfs, exclude_patterns=[], fs_name=fs_name)
try:
packer.pack_folder(dist_dir)
finally:
vfs.close()
size = _format_size(output_db)
return f"✅ Vue 打包完成 ({size})"
── 主流程 ──────────────────────────────────────────────────────────────────
async def main(): """规范化构建主流程""" mode = _resolve_mode() report = BuildReport(mode=mode, start_time=time.time()) config = get_config()
async def run_step(name: str, fn, *args, **kwargs) -> str:
"""执行一个构建步骤并记录结果（支持同步和异步函数）"""
start = time.time()
try:
result = fn(*args, **kwargs)
if asyncio.iscoroutine(result):
msg = await result
else:
msg = result
elapsed = time.time() - start
report.add_step(name, success=True, duration=elapsed, message=msg or "")
print(f" ✅ {name:<25s} {elapsed:>7.2f}s {msg}")
return msg
except Exception as e:
elapsed = time.time() - start
err_msg = str(e).strip()[:200]
report.add_step(name, success=False, duration=elapsed, message=err_msg)
print(f" ❌ {name:<25s} {elapsed:>7.2f}s {err_msg}")
return ""
print("\n" + "=" * 60)
print(" 🏗️ CHONGMING 构建开始")
print(f" 模式: {mode} | 平台: {platform.system()}")
print("=" * 60 + "\n")
# 阶段 1: 模块打包
await run_step("模块打包(module-bank)", _stage_pack_modules, config)
# 阶段 2: 准备构建目录
await run_step("准备构建目录", _stage_prepare_build_dir, config, mode)
# 阶段 3+4: 混淆 + 打包
if mode == PYINSTALLER:
await run_step("PyArmor 混淆", _stage_pyarmor_obfuscate)
start = time.time()
try:
msg, exe_path = _stage_pyinstaller(config)
elapsed = time.time() - start
success = "❌" not in msg
report.add_step(
"PyInstaller 打包", success, elapsed, msg.replace("✅ PyInstaller ", "")
)
print(
f" {'✅' if success else '❌'} PyInstaller 打包{' ' * 12} {elapsed:>7.2f}s {msg}"
)
except Exception as e:
elapsed = time.time() - start
report.add_step("PyInstaller 打包", False, elapsed, str(e)[:200])
print(f" ❌ PyInstaller 打包 {elapsed:>7.2f}s {str(e)[:200]}")
else:
await run_step("Nuitka 编译", lambda: _stage_nuitka(config)[0])
# 阶段 5: 清理
await run_step("清理临时文件", _stage_cleanup, mode)
# 阶段 7: Vue 前端构建
await run_step("Vue 前端构建", _stage_build_vue)
# 总结
report.end_time = time.time()
report.print_summary()
if not report.all_success:
sys.exit(1)
def build(): """构建入口""" asyncio.run(main())
if name == "main": build()
`./src/chongming/app/__init__.py`
""" chongming FastAPI 应用入口
重构为配置驱动的微服务网关：
- WorkerManager: 统一管理 Zenoh worker 存活检测、负载均衡、超时重试、缓存
- RouteGenerator: 根据配置自动生成 FastAPI 端点
- 插件工厂函数动态加载 """ import os
import time import subprocess import asyncio from contextlib import asynccontextmanager import boto3 from botocore.exceptions import ClientError from botocore.config import Config from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware
from .core.config import get_config, get_workers_config from .core.logger import get_logger from .core.static_files import SVFSStaticFiles from .routes import storage_router
config = get_config() logger = get_logger("app")
try: env = config["default"]["env"] except KeyError: # pragma: no cover raise ValueError("配置不存在")
@asynccontextmanager async def lifespan(app: FastAPI): import zenoh from diskcache import Cache from .core.worker_manager import WorkerManager from .core.route_generator import register_dynamic_routes
env = os.environ.copy()
env["NANIO_ACCESS_KEY"] = "minioadmin"
env["NANIO_SECRET_KEY"] = "minioadmin"
nanio_process = subprocess.Popen(
["nanio", "serve", "--data-dir", "./data/storage"],
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
app.state.nanio_process = nanio_process
await asyncio.sleep(1) # 等待 nanio 启动完成
s3_client_for_init = boto3.client(
"s3",
endpoint_url="http://127.0.0.1:9000",
aws_access_key_id="minioadmin",
aws_secret_access_key="minioadmin",
config=Config(signature_version="s3v4"),
region_name="us-east-1"
)
try:
s3_client_for_init.head_bucket(Bucket="chongming")
except ClientError as e:
if e.response["Error"]["Code"] == "404":
s3_client_for_init.create_bucket(Bucket="chongming")
logger.info("Bucket 'chongming' 已创建")
else:
raise
app.state.s3_client = s3_client_for_init
app.state.storage_bucket = "chongming"
app.state.storage_presign_expire = 300
app.state.storage_max_size = 100 * 1024 * 1024 # 100MB
# 1. 建立 Zenoh 会话
zconfig = zenoh.Config()
zconfig.from_file("config/zenoh_config.json5")
try:
session = zenoh.open(zconfig)
logger.info("Zenoh 会话已建立 (TCP 自连接模式)")
except Exception as e:
logger.error(f"Zenoh 会话建立失败: {e}")
raise
app.state.zenoh_session = session
logger.info("Zenoh 会话已建立")
# 2. 创建 WorkerManager
cache = Cache(config["cache"]["cache_store_path"])
worker_manager = WorkerManager(session, cache)
app.state.worker_manager = worker_manager
# 3. 启动 Worker 存活刷新任务
refresh_task = asyncio.create_task(worker_manager.refresh_workers())
app.state.refresh_task = refresh_task
# 4. 动态加载 worker 插件（根据配置）
workers_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml")
workers_conf = get_workers_config(workers_config_path).get("workers", [])
print(workers_conf)
worker_tasks = []
for wc in workers_conf:
module_name = wc.get("module", "")
factory_name = wc.get("factory", "")
count = wc.get("count", 1)
worker_config = wc.get("config", {})
if not module_name or not factory_name:
logger.warning(f"Worker 配置不完整: {wc}")
continue
try:
# 动态导入模块
module = __import__(module_name, fromlist=[factory_name])
factory = getattr(module, factory_name)
print(f"[factory]({factory_name}):{factory}")
# 调用工厂函数，传入 session、自定义配置、副本数和 worker_manager
# worker_manager 会传递给每个 worker，让 worker 启动后直接注册到本地管理器
tasks = factory(session, worker_config, count, worker_manager=worker_manager)
worker_tasks.extend(tasks)
logger.info(f"已加载 Worker 插件: {module_name}.{factory_name} (副本数: {count})")
except Exception as e:
logger.error(f"加载 Worker 插件失败 {module_name}: {e}")
app.state.worker_tasks = worker_tasks
logger.info(f"已启动 {len(worker_tasks)} 个 Worker")
# 5. 注册动态路由
register_dynamic_routes(app, worker_manager)
# 短暂等待确保 worker 状态已同步
await asyncio.sleep(0.5)
# 初始同步完成后打印 worker 状态
worker_status = {
wtype: len(workers)
for wtype, workers in app.state.worker_manager._workers_cache.items()
}
logger.info(f"Worker 就绪状态: {worker_status}")
yield # 应用运行中...
# 6. 清理
logger.info("正在关闭应用...")
# 停止 Worker 刷新任务
refresh_task.cancel()
try:
await refresh_task
except asyncio.CancelledError:
pass
# 取消所有 Worker 任务
for task in worker_tasks:
task.cancel()
if worker_tasks:
await asyncio.gather(*worker_tasks, return_exceptions=True)
# 关闭 Zenoh 会话
session.close()
logger.info("应用已关闭")
nanio_process.terminate()
nanio_process.wait()
logger.info("Nanio 进程已终止")
cors = config[env].get("cors", {}) app = FastAPI( title=config["default"]["app"]["name"], version=config["default"]["app"]["version"], description=config["default"]["app"]["description"], docs_url="/docs" if config[env]["debug"] else None, redoc_url="/redoc" if config[env]["debug"] else None, openapi_url="/openapi.json" if config[env]["debug"] else None, lifespan=lifespan, )
app.add_middleware(CORSMiddleware, **cors)
@app.middleware("http") async def access_log_middleware(request: Request, call_next): """ HTTP访问日志中间件，记录每个请求的详细信息。
记录的信息包括：
- 请求方法和路径
- 响应状态码
- 请求处理时长（毫秒）
- 客户端IP地址
参数:
request: FastAPI请求对象
call_next: 下一个中间件或路由处理函数
返回:
Response: 处理后的响应对象
"""
start = time.perf_counter()
response = await call_next(request)
duration = (time.perf_counter() - start) * 1000
client = request.client.host if request.client else "-"
logger.info(
f"{request.method} {request.url.path} status={response.status_code} "
f"duration={duration:.2f}ms client={client}",
extra={"status_code": response.status_code},
)
return response
@app.get("/", summary="根路径") async def root(): """ 根路径端点，返回应用的基本信息。
返回:
dict: 包含应用名称、版本、文档地址和健康检查地址的字典
"""
return {
"message": f"欢迎使用 {config['default']['app']['name']}",
"version": config["default"]["app"]["version"],
"docs": "/docs" if config[env]["debug"] else None,
"health": f"{config['default']['prefix']}/health",
}
@app.exception_handler(Exception) async def global_exception_handler(_: Request, exc: Exception): """ 全局异常处理器，捕获并处理应用中未捕获的异常。
在调试模式下返回详细的异常信息，在生产模式下返回通用错误提示。
参数:
_: FastAPI请求对象（未使用）
exc: 捕获到的异常对象
返回:
JSONResponse: 包含错误信息的JSON响应，状态码为500
"""
logger.error(f"全局异常: {exc}")
return JSONResponse(
status_code=500,
content={
"error": "内部服务器错误",
"detail": str(exc) if config[env]["debug"] else "请查看服务器日志",
},
)
静态文件服务（可选）
if vfs_db_path := config[env].get("file_system", {}).get("path", None): logger.info("静态文件已存在，将使用静态文件服务") static_files_handler = SVFSStaticFiles(directory="/", vfs_db_path=vfs_db_path) app.mount("/static", static_files_handler, name="static")
app.include_router(storage_router)
`./src/chongming/app/core/route_generator.py`
""" 动态路由生成器
根据配置文件中的 worker 路由定义，动态生成 FastAPI 端点。 为每个端点生成带有正确 Pydantic 类型注解的处理函数，确保 OpenAPI 文档 正确反映参数类型、响应结构，并注册到 components/schemas。 """ import asyncio import json from typing import Any, Dict, List, Optional, get_type_hints from inspect import Parameter, Signature
from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Depends from pydantic import create_model, Field, BaseModel
from .worker_manager import WorkerManager from .logger import get_logger
logger = get_logger("route_generator")
简单类型映射
TYPE_MAP: Dict[str, Any] = { "float": float, "int": int, "str": str, "bool": bool, "any": Any, }
def _parse_python_type(type_str: str) -> Any: """解析配置中的类型字符串为 Python 类型对象""" type_str = type_str.strip() if type_str.startswith("list[") or type_str.startswith("List["): inner = type_str[5:-1].strip() inner_type = _parse_python_type(inner) return List[inner_type] # type: ignore if type_str.startswith("dict[") or type_str.startswith("Dict["): inner = type_str[5:-1].strip() parts = [p.strip() for p in inner.split(",", 1)] if len(parts) == 2: key_type = _parse_python_type(parts[0]) val_type = _parse_python_type(parts[1]) return Dict[key_type, val_type] # type: ignore return Dict[str, Any] # type: ignore return TYPE_MAP.get(type_str, str)
def generate_request_model( param_spec: Dict[str, str], model_name: str = "AutoRequest" ) -> type[BaseModel]: """生成 Pydantic 请求体模型""" fields: Dict[str, Any] = {} for param_name, param_type_str in param_spec.items(): py_type = _parse_python_type(param_type_str) fields[param_name] = (py_type, Field(...)) return create_model(model_name, **fields)
def generate_response_model( response_spec: Dict[str, Any], model_name: str = "AutoResponse" ) -> Optional[type[BaseModel]]: """生成 Pydantic 响应模型""" if not response_spec: return None fields: Dict[str, Any] = {} for field_name, field_spec in response_spec.items(): if isinstance(field_spec, dict): py_type = _parse_python_type(field_spec.get("type", "str")) required = field_spec.get("required", True) default_val = field_spec.get("default", ...) if required: fields[field_name] = (py_type, Field(...)) else: fields[field_name] = (py_type, Field(default=default_val if default_val is not ... else None)) else: py_type = _parse_python_type(str(field_spec)) fields[field_name] = (py_type, Field(...)) return create_model(model_name, **fields)
class RouteConfig: """单个路由配置模型"""
def __init__(self, data: Dict[str, Any]):
self.subject: str = data.get("subject", "")
self.path: str = data.get("path", "")
self.method: str = data.get("method", "GET").upper()
self.params: List[str] = data.get("params", [])
self.response_model: Dict[str, Any] = data.get("response_model", {})
self.summary: str = data.get("summary", "")
self.docstring: str = data.get("docstring", "")
self.timeout: float = data.get("timeout", 3.0)
self.cacheable: bool = data.get("cacheable", False)
def parse_params(self) -> Dict[str, str]:
"""解析 params 列表为 {name: type} 字典"""
result: Dict[str, str] = {}
for param in self.params:
if ":" in param:
name, type_str = param.split(":", 1)
result[name.strip()] = type_str.strip()
else:
result[param.strip()] = "str"
return result
def extract_path_params(self) -> List[str]:
"""从路径中提取路径参数，如 /add/{a}/{b} -> ['a', 'b']"""
return [
seg[1:-1] for seg in self.path.split("/")
if seg.startswith("{") and seg.endswith("}")
]
默认的 worker 配置目录
WORKER_CONFIG_DIR = "config/workers"
def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]: """ 从 config/workers/{worker_type}/ 目录加载路由配置。
目录结构:
config/workers/{worker_type}/
config.toml # [worker] 分组信息 + registration + defaults
add.toml # 接口名 = 文件名（不含扩展名）
[GET]
summary = ...
params = [...]
response_model = { ... }
[POST]
... # 同一个接口支持多个请求方法
返回: List[Dict], 每个 dict 包含 flatten 后的路由配置,
每个方法节(如 GET) 生成一个独立的路由条目。
"""
from pathlib import Path
import tomllib
base_dir = Path(WORKER_CONFIG_DIR) / worker_type
if not base_dir.is_dir():
logger.warning(f"Worker 配置目录不存在: {base_dir}")
return []
# 1. 读取分组配置
group_cfg_path = base_dir / "config.toml"
group_defaults: Dict[str, Any] = {}
registration: Dict[str, Any] = {}
router_prefix: str = ""
if group_cfg_path.exists():
with open(group_cfg_path, "rb") as f:
cfg = tomllib.load(f)
w = cfg.get("worker", {})
group_defaults = w.get("defaults", {})
registration = w.get("registration", {})
router_prefix = registration.get("router_prefix", "")
# 2. 遍历所有接口 toml 文件
routes: List[Dict[str, Any]] = []
for route_file in sorted(base_dir.glob("*.toml")):
if route_file.name == "config.toml":
continue
# 文件名（不含 .toml）作为路径名
route_name = route_file.stem # 例如 "add"
base_path = router_prefix + "/" + route_name # 例如 "/calc/add"
with open(route_file, "rb") as f:
route_data = tomllib.load(f)
# 每个方法节（GET/POST/PUT/DELETE）生成一个独立路由
for method, spec in route_data.items():
if not isinstance(spec, dict):
logger.warning(f"忽略非 dict 的方法节: {route_file.name}[{method}]")
continue
# 合并: spec 显式字段优先，然后 group_defaults 兜底
route = dict(spec)
for key, default_val in group_defaults.items():
if key not in route:
route[key] = default_val
# 覆盖关键字段 - method 始终从 TOML 节键获取，不被 defaults 覆盖
route["method"] = method.upper()
route["path"] = route.get("path", base_path)
if "subject" not in route:
# 自动生成 subject: worker_type.route_name (如 "calc.add")
route["subject"] = f"{worker_type}.{route_name}"
routes.append(route)
logger.debug(
f" 已加载: {route['method']} {route['path']} -> {route['subject']}"
)
logger.info(f"从 {base_dir} 加载了 {len(routes)} 个路由")
return routes
def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager): """根据 config/workers/ 目录结构注册所有动态路由""" from pathlib import Path
workers_base = Path(WORKER_CONFIG_DIR)
if not workers_base.is_dir():
logger.info(f"Worker 配置目录不存在: {WORKER_CONFIG_DIR}")
return
router = APIRouter()
extra_schemas: Dict[str, type[BaseModel]] = {}
# 扫描所有 worker 子目录（每个子目录名 = worker_type）
for worker_dir in sorted(workers_base.iterdir()):
if not worker_dir.is_dir():
continue
worker_type = worker_dir.name
routes = load_worker_routes_from_dir(worker_type)
if not routes:
continue
for route_conf in routes:
_register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas)
app.include_router(router)
# 将额外 schema 注册到 OpenAPI 的 components/schemas
if extra_schemas:
_inject_schemas_to_openapi(app, extra_schemas)
logger.info(f"动态路由注册完成，worker 配置目录: {WORKER_CONFIG_DIR}")
def _inject_schemas_to_openapi(app: FastAPI, schemas: Dict[str, type[BaseModel]]): """ 将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。
通过重写 app.openapi() 方法注入，因为 FastAPI 在生命周期内只生成一次
OpenAPI schema，之后会被缓存。
"""
original_openapi = app.openapi
def patched_openapi() -> dict[str, Any]:
if app.openapi_schema is not None:
return app.openapi_schema
schema = original_openapi()
if schema is None:
return {"openapi": "3.1.0", "info": {"title": "Chongming", "version": "1.0.0"}, "paths": {}}
# 注入我们的 schema
schemas_dict = schema.setdefault("components", {}).setdefault("schemas", {})
for name, model in schemas.items():
if name not in schemas_dict:
schemas_dict[name] = model.model_json_schema()
app.openapi_schema = schema
return schema
app.openapi = patched_openapi
def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str, RequestModel: type[BaseModel]): """创建用于 Body 请求（POST/PUT/PATCH）的端点函数""" path = rc.path subject = rc.subject timeout = rc.timeout http_method = rc.method cacheable = rc.cacheable
async def endpoint(
body: RequestModel, # type: ignore
) -> Any:
try:
payload = body.model_dump()
payload.setdefault("_method", http_method)
result = await worker_manager.call(
worker_type=worker_type,
subject=subject,
payload=payload,
timeout=timeout,
cacheable=cacheable,
)
return result
except RuntimeError as e:
raise HTTPException(status_code=503, detail=str(e))
except Exception as e:
logger.error(f"路由 {path} 调用异常: {e}")
raise HTTPException(status_code=500, detail="内部服务器错误")
return endpoint
def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str, path_params: List[str], query_params: Dict[str, str]): """ 创建用于 Query 请求（GET/DELETE）的端点函数。
通过设置函数 __signature__ 和 __annotations__ 为每个参数赋予正确的 Pydantic 类型，
使 FastAPI 能在 OpenAPI schema 中生成正确的 type / format 信息。
请求模型 schema 通过 extra_schemas 机制单独注入到 components/schemas。
"""
path = rc.path
subject = rc.subject
timeout = rc.timeout
http_method = rc.method
cacheable = rc.cacheable
# 合并路径参数和查询参数
param_spec_combined: Dict[str, str] = {}
param_spec_combined.update({k: query_params.get(k, "str") for k in path_params if k in query_params})
param_spec_combined.update(query_params)
# 无参数情况
if not param_spec_combined:
async def _simple_endpoint() -> Any:
try:
result = await worker_manager.call(
worker_type=worker_type,
subject=subject,
payload={"_method": http_method},
timeout=timeout,
cacheable=cacheable,
)
return result
except RuntimeError as e:
raise HTTPException(status_code=503, detail=str(e))
except Exception as e:
logger.error(f"路由 {path} 调用异常: {e}")
raise HTTPException(status_code=500, detail="内部服务器错误")
return _simple_endpoint
# 构造参数名列表（用于从 kwargs 中提取）
param_names = list(param_spec_combined.keys())
async def endpoint(**kwargs: Any) -> Any:
payload = {k: kwargs[k] for k in param_names}
payload.setdefault("_method", http_method)
try:
result = await worker_manager.call(
worker_type=worker_type,
subject=subject,
payload=payload,
timeout=timeout,
cacheable=cacheable,
)
return result
except RuntimeError as e:
raise HTTPException(status_code=503, detail=str(e))
except Exception as e:
logger.error(f"路由 {path} 调用异常: {e}")
raise HTTPException(status_code=500, detail="内部服务器错误")
# 构建带类型注解的 Parameter 列表
params: List[Parameter] = []
annotations: Dict[str, Any] = {}
for name, type_str in param_spec_combined.items():
py_type = _parse_python_type(type_str)
annotations[name] = py_type
if name in path_params:
params.append(Parameter(name, Parameter.KEYWORD_ONLY, default=FastAPIPath(...), annotation=py_type))
else:
params.append(Parameter(name, Parameter.KEYWORD_ONLY, default=Query(...), annotation=py_type))
# 设置 __annotations__（FastAPI 通过 get_type_hints 读取）
endpoint.__annotations__ = annotations
# 设置 __signature__（FastAPI 优先于此获取参数列表）
endpoint.__signature__ = Signature(params, return_annotation=Dict[str, Any]) # type: ignore
return endpoint
def _register_single_route( router: APIRouter, worker_manager: WorkerManager, worker_type: str, route_conf: Dict[str, Any], extra_schemas: Optional[Dict[str, type[BaseModel]]] = None, ): """注册单个路由，生成正确的函数签名以便 OpenAPI 正确生成文档""" rc = RouteConfig(route_conf) param_spec = rc.parse_params() path_params = rc.extract_path_params() query_params = {k: v for k, v in param_spec.items() if k not in path_params}
# 生成干净的模型名称（替换路径中的 / 和 {} 为 _）
# 模型名称包含 HTTP method 以区分同名路径的不同调用类型
clean_path = rc.path.replace("/", "_").replace("{", "").replace("}", "").strip("_")
if not clean_path:
clean_path = "root"
method_tag = rc.method.upper()
response_model_name = f"{worker_type}_{clean_path}_{method_tag}_Response"
request_model_name = f"{worker_type}_{clean_path}_{method_tag}_Request"
# 生成响应模型
ResponseModel = generate_response_model(
rc.response_model, response_model_name
)
if rc.method in ("POST", "PUT", "PATCH"):
RequestModel = generate_request_model(
param_spec, request_model_name
)
endpoint = _make_body_endpoint(worker_manager, rc, worker_type, RequestModel)
else:
# 对 GET 端点也生成 Request 模型并注册到 extra_schemas
if param_spec:
RequestModel = generate_request_model(
param_spec, request_model_name
)
if extra_schemas is not None and request_model_name not in extra_schemas:
extra_schemas[request_model_name] = RequestModel
endpoint = _make_query_endpoint(worker_manager, rc, worker_type, path_params, query_params)
# 注册路由 - 使用 worker_type 作为 OpenAPI 分组标签
# 每个 worker 类型自动成为一个独立的分组，显示在 Swagger UI 中
router.add_api_route(
path=rc.path,
endpoint=endpoint,
methods=[rc.method],
summary=rc.summary,
description=rc.docstring,
response_model=ResponseModel,
tags=[worker_type],
include_in_schema=True,
)
logger.info(f"已注册动态路由: {rc.method} {rc.path} -> worker {worker_type}.{rc.subject} | tags=[{worker_type}]")
`./src/chongming/app/core/worker_manager.py`
import asyncio import random import json import orjson import time from typing import Any, Dict, List, Optional, Set import zenoh from diskcache import Cache from .logger import get_logger
logger = get_logger("worker_manager")
_SENTINEL = object() # 哨兵值，用于标记"No reply received"
class WorkerManager: """ 统一 Worker 管理器
封装 Zenoh 会话，维护每个 worker 类型的存活节点列表，
提供负载均衡、超时重试、缓存等能力的统一 RPC 调用入口。
支持两种 worker 注册方式：
1. 分布式：通过 Zenoh liveliness 自动发现
2. 本地：通过 register_worker() 直接注册（用于同进程 worker）
"""
def __init__(self, session: zenoh.Session, cache: Optional[Cache] = None):
self.session = session
self.cache = cache
self._workers_cache: Dict[str, List[str]] = {} # type -> list of worker key_exprs
self._local_workers: Dict[str, List[str]] = {} # type -> list of locally registered workers
self._lock = asyncio.Lock()
self._refresh_interval = 15 # seconds
def register_worker(self, worker_type: str, worker_id: str):
"""
直接注册一个本地 worker（同进程内使用）
这样无需等待 liveliness 发现，立即可用。
同步执行（仅修改内存字典），确保调用后立即可被 get_worker 发现。
"""
key_expr = f"worker/{worker_type}/{worker_id}"
# 同步直接写入 _local_workers，无需异步锁
# 因为 _local_workers 和 _workers_cache 在调用 call() 时通过 await get_worker() 读取
self._local_workers.setdefault(worker_type, []).append(key_expr)
logger.debug(f"本地 worker 已注册: {key_expr}")
def unregister_worker(self, worker_type: str, worker_id: str):
"""取消注册一个本地 worker"""
key_expr = f"worker/{worker_type}/{worker_id}"
if worker_type in self._local_workers:
try:
self._local_workers[worker_type].remove(key_expr)
except ValueError:
pass
async def refresh_workers(self):
"""定期刷新存活 worker 列表（合并 Zenoh liveliness + 本地注册）"""
while True:
try:
# 1. 从 Zenoh liveliness 获取分布式 workers
replies = self.session.liveliness().get("worker/*/*")
new_cache: Dict[str, List[str]] = {}
for reply in replies:
if reply.ok:
key = str(reply.ok.key_expr) # e.g. "worker/example/worker-0"
parts = key.split("/")
if len(parts) >= 3:
worker_type = parts[1]
new_cache.setdefault(worker_type, []).append(key)
# 2. 合并本地注册的 workers
for wtype, workers in self._local_workers.items():
existing = set(new_cache.get(wtype, []))
for w in workers:
if w not in existing:
new_cache.setdefault(wtype, []).append(w)
async with self._lock:
self._workers_cache = new_cache
logger.debug(f"Worker 存活列表更新: {self._workers_cache}")
except Exception as e:
logger.error(f"刷新 worker 列表失败: {e}")
await asyncio.sleep(self._refresh_interval)
async def get_worker(self, worker_type: str) -> Optional[str]:
"""获取一个可用的 worker 地址（简单随机负载均衡）"""
async with self._lock:
workers = self._workers_cache.get(worker_type, [])
# 如果缓存为空，回退到本地注册表（适用于启动初期 refresh_workers 尚未执行时）
if not workers:
local = self._local_workers.get(worker_type, [])
if local:
return random.choice(local)
return None
return random.choice(workers)
def is_success_response(self, result: dict) -> bool:
"""根据业务约定判断响应是否成功。
假设成功响应没有 'error' 字段，或 'code' 为 0。
"""
if "error" in result and result["error"]:
return False
if "code" in result and result["code"] != 0:
return False
return True
async def call(
self,
worker_type: str,
subject: str,
payload: Dict[str, Any],
timeout: float = 3.0,
retries: int = 1,
cacheable: bool = False,
) -> Any:
"""
统一调用入口：缓存 > 负载均衡 > 超时重试
Args:
worker_type: Worker 类型标识（如 "example"）
subject: 调用的主题（如 "calc.add"）
payload: 请求参数
timeout: 单次调用超时（秒）
retries: 失败重试次数
cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写）
Returns:
Dict[str, Any]: 调用结果
Raises:
RuntimeError: 无可用 worker 或所有重试均失败
"""
# 1. 缓存检查（仅在 cacheable=True 时检查）
cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}"
if cacheable and self.cache is not None:
if cache_key in self.cache:
logger.debug(f"命中缓存: {cache_key}")
return self.cache[cache_key]
# 2. 获取可用 worker
worker_expr = await self.get_worker(worker_type)
if not worker_expr:
raise RuntimeError(f"No available worker for type {worker_type}")
full_key = f"{worker_expr}/{subject}"
payload_bytes = orjson.dumps(payload)
for attempt in range(retries + 1):
try:
reply_iter = self.session.get(
full_key,
payload=payload_bytes,
consolidation=zenoh.ConsolidationMode.NONE,
)
# 使用辅助函数获取第一个回复，避免 StopIteration 在 asyncio.to_thread 中传播的问题
def _get_first(iterator):
"""线程安全地获取迭代器的第一个元素"""
for item in iterator:
return item
return _SENTINEL
first_reply = await asyncio.wait_for(
asyncio.to_thread(_get_first, reply_iter),
timeout=timeout,
)
if first_reply is _SENTINEL:
logger.warning(
f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
)
continue
reply = first_reply
if reply.ok: # type: ignore
# result_str = str(reply.ok.payload)
payload_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"" # type: ignore
if not payload_bytes:
result = {} # 或根据业务返回默认值
else:
result = orjson.loads(payload_bytes)
# 缓存结果（仅在 cacheable=True 时写入）
if cacheable and self.cache is not None and self.is_success_response(result):
self.cache[cache_key] = result
return result
else:
error_payload = reply.err.payload # type: ignore
error_msg = str(error_payload) if error_payload else "unknown error"
logger.warning(f"Worker 返回错误: {error_msg} (尝试 {attempt+1}/{retries+1})")
continue # 重试
except asyncio.TimeoutError:
logger.warning(
f"Worker {worker_expr} 超时，尝试重试 {attempt+1}/{retries+1}"
)
continue
except Exception as e:
logger.error(f"调用 worker 异常: {e}")
raise
raise RuntimeError(
f"Failed to call worker {worker_type} after {retries} retries"
)
`./src/chongming/app/core/config.py`
import sys from pathlib import Path from typing import TypedDict, Literal, Optional, cast, List
根据Python版本选择合适的TOML解析库和Required类型
if sys.version_info >= (3, 11): import tomllib from typing import Required else: import tomli as tomllib from typing_extensions import Required
设置 env 为 development 或 production
class DefaultApp(TypedDict): """应用默认配置结构"""
name: str
version: str
description: str
debug: bool
class Default(TypedDict): """系统默认配置结构，包含应用信息、环境、路径等基础配置"""
app: DefaultApp
env: Literal["production", "development"]
prefix: str
upload_path: str
class SqliteConnectionArgs(TypedDict, total=False): """SQLite数据库连接参数配置"""
timeout: int
check_same_thread: bool
class DatabaseSqlite(TypedDict, total=False): """SQLite数据库连接池及行为配置"""
echo: bool
pool_size: int
max_overflow: int
pool_timeout: int
pool_recycle: int
pool_pre_ping: bool
connect_args: SqliteConnectionArgs
class Database(TypedDict, total=False): """数据库配置结构，支持SQLite、MySQL、PostgreSQL三种数据库类型"""
type: Required[Literal["sqlite", "mysql", "postgresql"]]
url: Required[str]
generate_schemas: bool
sqlite: DatabaseSqlite
mysql: dict
postgresql: dict
database_path: str
class Security(TypedDict): """安全相关配置，包括JWT令牌和安全策略设置"""
secret_key: str
algorithm: Literal["HS256"]
access_token_expire_minutes: int
refresh_token_expire_days: int
max_sessions_per_user: int
class Server(TypedDict, total=False): """服务器运行配置，包含主机、端口、工作进程等设置"""
host: Required[str]
port: Required[int]
reload: bool
workers: int
access_log: bool
class Logging(TypedDict): """日志系统配置，定义日志级别、输出格式和文件管理策略"""
level: str
file: str
format: str
max_size: int
backup_count: int
class Cors(TypedDict, total=False): """跨域资源共享(CORS)配置"""
allow_origins: List[str]
allow_credentials: bool
allow_methods: List[str]
allow_headers: List[str]
class ModuleSystem(TypedDict, total=False): """模块系统配置，定义模块加载类型和路径"""
type: str
path: List[str]
class FileSystem(TypedDict, total=False): """文件系统配置，定义存储类型和路径"""
type: str
path: str
class Env(TypedDict, total=False): """环境特定配置结构，用于开发和生产环境的差异化配置"""
debug: Required[bool]
server: Required[Server]
logging: Required[Logging]
cors: Cors
module_system: ModuleSystem
file_system: FileSystem
class Scheduler(TypedDict): """任务调度器配置"""
job_store_path: str
class Cache(TypedDict): """缓存系统配置"""
cache_store_path: str
class Config(TypedDict): """完整配置结构，整合所有配置项，包含默认配置、数据库、调度器、缓存、安全和环境配置"""
default: Default
database: Database
scheduler: Scheduler
cache: Cache
security: Security
development: Env
production: Env
def load_config(config_path: Path = Path("config.toml")) -> Config: """ 从TOML配置文件加载配置数据
Args:
config_path: 配置文件路径，默认为当前目录下的config.toml
Returns:
Config: 解析后的配置字典对象
Raises:
FileNotFoundError: 当配置文件不存在时抛出异常
"""
if not config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {config_path}")
with open(config_path, "rb") as f:
return cast(Config, tomllib.load(f))
全局配置缓存变量，避免重复加载配置文件
_config: Optional[Config] = None
def get_config(config_path: str = "config.toml") -> Config: """ 获取应用程序配置（带缓存机制）
首次调用时加载配置文件并缓存，后续调用直接返回缓存的配置对象
Args:
config_path: 配置文件路径，默认为config.toml
Returns:
Config: 配置字典对象
"""
global _config
if _config is None:
_config = load_config(Path(config_path))
return _config
def get_workers_config(config_path: str = "config/workers/config.toml") -> dict: """ 获取Worker插件配置列表
Args:
config_path: Worker配置文件路径，默认为config/workers/config.toml
Returns:
list: Worker配置字典列表
"""
if not Path(config_path).exists():
raise FileNotFoundError(f"Worker配置文件不存在: {config_path}")
with open(config_path, "rb") as f:
return cast(dict, tomllib.load(f))
`./src/chongming/app/core/__init__.py`
`./src/chongming/app/core/logger.py`
import re import logging import datetime from pathlib import Path from logging.handlers import RotatingFileHandler
from ..core.config import get_config
config = get_config() env = config["default"]["env"] server_config = config[env]["server"] log_config = config[env]["logging"]
class ColoredFormatter(logging.Formatter): """ 为 HTTP 状态码添加颜色的日志格式化器
根据 HTTP 状态码的首位数字为日志消息中的状态码添加不同颜色：
- 2xx: 绿色（成功）
- 3xx: 青色（重定向）
- 4xx: 黄色（客户端错误）
- 5xx: 红色（服务器错误）
Attributes:
COLORS: 状态码首位数字到 ANSI 颜色代码的映射字典
RESET: ANSI 重置颜色代码
"""
COLORS = {
"2": "\033[32m", # 绿色 2xx
"3": "\033[36m", # 青色 3xx
"4": "\033[33m", # 黄色 4xx
"5": "\033[31m", # 红色 5xx
}
RESET = "\033[0m"
def format(self, record):
"""
格式化日志记录，为状态码添加颜色
如果日志记录包含 status_code 属性，则查找消息中的 "status=数字" 模式
并为其添加对应的颜色代码。格式化完成后恢复原始消息以避免影响其他 handler。
Args:
record: 日志记录对象，应包含 msg 属性和可选的 status_code 属性
Returns:
str: 格式化后的日志消息字符串，状态码部分可能包含 ANSI 颜色代码
"""
# 保存原始消息
original_msg = record.msg
# 如果日志记录包含 status_code 属性，则为其添加颜色
status_code = getattr(record, "status_code", None)
if status_code and isinstance(status_code, int):
first_digit = str(status_code)[0]
color = self.COLORS.get(first_digit, "")
colored_status = f"{color}{status_code}{self.RESET}"
# 替换消息中的 "status=数字" 为带颜色的版本
record.msg = re.sub(
rf"status={status_code}\b", f"status={colored_status}", original_msg
)
result = super().format(record)
# 恢复原始消息，避免影响其他 handler
record.msg = original_msg
return result
class ExcludeWatchFilesFilter(logging.Filter): """ 日志过滤器，用于排除第三方库的日志输出
过滤掉来自 watchfiles、aiosqlite、apscheduler、sqlalchemy 等库的日志记录，
避免这些库的详细日志污染应用日志输出。
"""
def __init__(self):
"""
初始化过滤器，设置需要过滤的日志名称前缀列表
"""
super().__init__()
self._filter_starts_with_tulpe = (
"watchfiles",
"aiosqlite",
"apscheduler",
"sqlalchemy",
"sqlitedict",
"sqlalchemy.engine.Engine",
)
def filter(self, record):
"""
判断是否应该保留该日志记录
Args:
record: 日志记录对象，包含 name 属性表示日志来源
Returns:
bool: 如果日志名称不以过滤列表中的任何前缀开头则返回 True（保留），
否则返回 False（过滤掉）
"""
return not record.name.startswith(self._filter_starts_with_tulpe)
_configured_loggers = set()
def setup_logging(logger: logging.Logger, log_name: str = "app"): """ 配置日志系统，包括文件和控制台输出
为指定的 logger 配置以下功能：
1. 设置日志级别和格式
2. 创建按日期命名的轮转文件处理器（支持文件大小限制和备份数量控制）
3. 创建带颜色格式化的控制台处理器
4. 应用过滤器排除第三方库日志
该函数具有防重复配置机制，同一个 log_name 只会配置一次。
Args:
logger: 需要配置的 logging.Logger 实例
log_name: 日志名称标识，用于区分不同的日志配置和生成日志文件名，默认为 "app"
Returns:
None
Raises:
无显式抛出异常，但可能因文件系统权限问题导致目录创建或文件写入失败
"""
if log_name in _configured_loggers:
return
_configured_loggers.add(log_name)
log_level = getattr(logging, log_config["level"], logging.INFO)
log_format = log_config["format"]
# 创建文件夹
log_file_path = Path(log_config["file"])
if log_file_path != Path("."):
log_file_path.mkdir(parents=True, exist_ok=True)
# 配置根日志记录器
logger.setLevel(log_level)
# 创建格式化器
plain_formatter = logging.Formatter(log_format)
color_formatter = ColoredFormatter(log_format)
# 创建轮转文件处理器 - 限制文件大小为 10MB，保留 5 个备份文件
max_bytes = log_config.get("max_size", 10 * 1024 * 1024) # 默认 10MB
backup_count = log_config.get("backup_count", 5) # 默认保留 5 个备份
file_handler = RotatingFileHandler(
f"{log_config['file']}/{log_name}-{datetime.date.today()}.log",
encoding="utf-8",
maxBytes=max_bytes,
backupCount=backup_count,
)
file_handler.setFormatter(plain_formatter)
file_handler.addFilter(ExcludeWatchFilesFilter())
# 创建控制台处理器并添加过滤器
console_handler = logging.StreamHandler()
console_handler.setFormatter(color_formatter)
console_handler.addFilter(ExcludeWatchFilesFilter())
# 添加处理器到根日志记录器
logger.addHandler(file_handler)
logger.addHandler(console_handler)
print(
f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} "
f"(最大：{max_bytes / 1024 / 1024:.1f}MB, 备份：{backup_count})"
)
def get_logger(name: str) -> logging.Logger: """ 获取配置好的日志记录器实例
根据名称获取或创建 logger 实例，并自动调用 setup_logging 进行配置。
如果该名称的 logger 已经配置过，则直接返回已配置的实例。
Args:
name: 日志记录器名称，通常使用模块名如 __name__
Returns:
logging.Logger: 已配置完成的日志记录器实例，可直接用于记录日志
"""
logger = logging.getLogger(name)
setup_logging(logger, name)
return logger
`./src/chongming/app/core/static_files.py`
from fastapi import HTTPException from fastapi.responses import Response from fastapi.staticfiles import StaticFiles from sqlite_vfs.core import SQLiteVFS
class SVFSStaticFiles(StaticFiles): """ 基于 SQLite VFS 的静态文件服务类
该类继承自 FastAPI 的 StaticFiles，使用 SQLite 虚拟文件系统来存储和提供静态文件。
支持单页应用（SPA）的路由回退机制，当请求的文件不存在时，会自动返回 index.html。
"""
def __init__(self, directory, vfs_db_path="static.svfs"):
"""
初始化 SVFSStaticFiles 实例
参数:
directory (str): 静态文件的根目录路径
vfs_db_path (str): SQLite VFS 数据库文件路径，默认为 "static.svfs"
返回值:
无
"""
vfs_db_path = vfs_db_path
self._vfs = SQLiteVFS(vfs_db_path, compress=False)
super().__init__(directory=directory)
async def get_response(self, path: str, scope):
"""
根据请求路径获取对应的静态文件响应
该方法处理静态文件的请求逻辑，包括：
1. 路径标准化处理
2. 文件存在性检查
3. SPA 路由回退支持
4. 目录索引处理
5. MIME 类型自动识别
参数:
path (str): 请求的文件路径
scope: ASGI 作用域对象，包含请求上下文信息
返回值:
Response: FastAPI 响应对象，包含文件内容和正确的 Content-Type
异常:
HTTPException: 当文件不存在且无法回退到 index.html 时，抛出 404 错误
"""
# 统一路径格式：以 / 开头，并将 Windows 反斜杠转为正斜杠
normalized_path = "/" + path.replace("\\", "/")
# 获取文件信息
file_info = self._vfs.get_file_info(normalized_path)
# 处理文件不存在的情况，尝试 SPA 回退机制
if file_info is None:
# SPA 回退：如果不是静态资源扩展名，尝试返回 index.html
if not any(
normalized_path.endswith(ext)
for ext in [
".js",
".css",
".png",
".jpg",
".ico",
".svg",
".webp",
".json",
]
):
index_info = self._vfs.get_file_info("/index.html")
if index_info:
content = self._vfs.read_file("/index.html")
return Response(content=content, media_type="text/html")
raise HTTPException(status_code=404, detail="File not found")
# 处理目录请求，返回目录下的 index.html
if file_info["is_directory"]:
index_info = self._vfs.get_file_info(f"{normalized_path}/index.html")
if index_info:
content = self._vfs.read_file(f"{normalized_path}/index.html")
return Response(content=content, media_type="text/html")
raise HTTPException(status_code=404, detail="Directory index not found")
# 读取文件内容并确定 MIME 类型
content = self._vfs.read_file(normalized_path)
content_type = file_info.get("content_type") or "application/octet-stream"
# 简单根据扩展名补充 MIME 类型（可选）
if normalized_path.endswith(".js"):
content_type = "application/javascript"
elif normalized_path.endswith(".css"):
content_type = "text/css"
elif normalized_path.endswith(".html"):
content_type = "text/html"
elif normalized_path.endswith(".ico"):
content_type = "image/x-icon"
return Response(content=content, media_type=content_type)
`./src/chongming/app/routes/storage.py`
src/chongming/app/routes/storage.py
import uuid import urllib.parse from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Request from fastapi.responses import StreamingResponse from botocore.exceptions import ClientError
router = APIRouter(prefix="/api/v1/storage", tags=["Storage"])
@router.post("/upload", summary="上传文件") async def upload_file( request: Request, file: UploadFile = File(...), worker_type: str = Form(...), ): """后端代理上传，直接写入 nanio""" s3_client = request.app.state.s3_client bucket = request.app.state.storage_bucket object_key = f"{worker_type}/{uuid.uuid4()}/{file.filename}"
try:
s3_client.upload_fileobj(
file.file,
bucket,
object_key,
ExtraArgs={"ContentType": file.content_type or "application/octet-stream"}
)
except ClientError as e:
raise HTTPException(status_code=500, detail=f"S3 上传失败: {str(e)}")
return {
"file_key": object_key,
"bucket": bucket,
"filename": file.filename,
"content_type": file.content_type,
"size": file.size or 0, # 注意：UploadFile.size 可能为 None，需读取后计算
}
@router.get("/download/{file_key:path}") async def download_file(request: Request, file_key: str): s3_client = request.app.state.s3_client bucket = request.app.state.storage_bucket
try:
response = s3_client.get_object(Bucket=bucket, Key=file_key)
content_type = response.get('ContentType', 'application/octet-stream')
raw_filename = file_key.split('/')[-1]
# 百分号编码，确保仅 ASCII 字符
encoded = urllib.parse.quote(raw_filename, safe='')
# 仅使用 filename*（现代浏览器都支持）
content_disposition = f"attachment; filename*=UTF-8''{encoded}"
return StreamingResponse(
response['Body'].iter_chunks(),
media_type=content_type,
headers={'Content-Disposition': content_disposition}
)
except ClientError as e:
if e.response['Error']['Code'] == 'NoSuchKey':
raise HTTPException(404, "文件不存在")
raise HTTPException(500, f"下载失败: {e}")
`./src/chongming/app/routes/__init__.py`
from .storage import router as storage_router
all = ["storage_router"]