From 247005bc79c1fecc8e8d0c39a629eea5845c20a2 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 17 Jul 2026 12:04:34 +0800 Subject: [PATCH] modify: chongming-request --- .gitignore | 1 + pyproject.toml | 2 +- src/plugins/example/__init__.py | 26 ++++++++--- .../request/src/chongming_request/__init__.py | 1 + .../worker/src/chongming_worker/decorators.py | 43 ++++++++----------- utils/worker/src/chongming_worker/runtime.py | 20 ++++++--- 6 files changed, 56 insertions(+), 37 deletions(-) diff --git a/.gitignore b/.gitignore index 0c2bd04..9aae615 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,4 @@ data logs resources cache_data +output.md diff --git a/pyproject.toml b/pyproject.toml index 9b3cdfd..e7f26b7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,8 +49,8 @@ members = [ ] [tool.uv.sources] -chongming-worker = { workspace = true } chongming-request = { workspace = true } +chongming-worker = { workspace = true } [project.scripts] build = "chongming.scripts.build:build" diff --git a/src/plugins/example/__init__.py b/src/plugins/example/__init__.py index 7fbe9d1..b099766 100644 --- a/src/plugins/example/__init__.py +++ b/src/plugins/example/__init__.py @@ -17,8 +17,11 @@ Example Worker 插件 — 演示 @rpc_method + 显式模型绑定用法 """ import time +import zenoh + from chongming_worker import rpc_method, auto_bind_models from chongming_worker.runtime import create_workers as _create_workers +from chongming_request import Client # ============================================================ # 显式导入 Pydantic 模型（避免 PyArmor 打包后 importlib 动态导入异常） @@ -51,7 +54,7 @@ from .models.pydantic import ( summary="Add two numbers via GET", response_model=AddGetResponse, ) -def add(add_get_request:AddGetRequest) -> AddGetResponse: +async def add(add_get_request:AddGetRequest) -> AddGetResponse: """加法运算（GET）— 通过查询参数传入 a、b，返回它们的和""" return AddGetResponse( result=add_get_request.a + add_get_request.b, @@ -65,8 +68,19 @@ def add(add_get_request:AddGetRequest) -> AddGetResponse: summary="Add two numbers via POST", response_model=AddPostResponse, ) -def add_post(add_post_request: AddPostRequest) -> AddPostResponse: +async def add_post(add_post_request: AddPostRequest, session: zenoh.Session) -> AddPostResponse: """加法运算（POST）— 通过 JSON Body 传入 a、b，返回它们的和""" + async with Client(session=session) as client: + add_result = await client.call( + worker_type="example", + subject="example.add", + payload={ + "a":add_post_request.a, + "b":add_post_request.b + }, + method="GET" + ) + print(add_result) return AddPostResponse( result=add_post_request.a + add_post_request.b, operation="add", @@ -79,8 +93,8 @@ def add_post(add_post_request: AddPostRequest) -> AddPostResponse: method="GET", summary="Subtract two numbers", response_model=SubtractGetResponse, -) -def subtract(subtract_get_request: SubtractGetRequest) -> SubtractGetResponse: +) +async def subtract(subtract_get_request: SubtractGetRequest) -> SubtractGetResponse: """减法运算：a - b""" return SubtractGetResponse( result=subtract_get_request.a - subtract_get_request.b, @@ -95,7 +109,7 @@ def subtract(subtract_get_request: SubtractGetRequest) -> SubtractGetResponse: summary="Multiply two numbers", response_model=MultiplyGetResponse, ) -def multiply(multiply_get_request=MultiplyGetRequest) -> MultiplyGetResponse: +async def multiply(multiply_get_request=MultiplyGetRequest) -> MultiplyGetResponse: """乘法运算：a * b""" return MultiplyGetResponse( result=multiply_get_request.a * multiply_get_request.b, @@ -109,7 +123,7 @@ def multiply(multiply_get_request=MultiplyGetRequest) -> MultiplyGetResponse: summary="Divide two numbers", response_model=DivideGetResponse, ) -def divide(divide_get_request: DivideGetRequest) -> DivideGetResponse: +async def divide(divide_get_request: DivideGetRequest) -> DivideGetResponse: """除法运算：a / b（b 不能为 0）""" if divide_get_request.b == 0: return DivideGetResponse( diff --git a/utils/request/src/chongming_request/__init__.py b/utils/request/src/chongming_request/__init__.py index e69de29..3ead93a 100644 --- a/utils/request/src/chongming_request/__init__.py +++ b/utils/request/src/chongming_request/__init__.py @@ -0,0 +1 @@ +from .client import Client \ No newline at end of file diff --git a/utils/worker/src/chongming_worker/decorators.py b/utils/worker/src/chongming_worker/decorators.py index 1be18b1..91aee92 100644 --- a/utils/worker/src/chongming_worker/decorators.py +++ b/utils/worker/src/chongming_worker/decorators.py @@ -235,59 +235,54 @@ class RPCMethod: return raw_result.model_dump() return raw_result - async def call(self, payload: Dict[str, Any]) -> Any: + async def call(self, payload: Dict[str, Any], **extra_kwargs) -> Any: """ - 调用原始函数，自动从 payload 提取参数 - - 流程： - 1. 校验请求参数（使用绑定的 request_model 或自动生成） - 2. 提取函数需要的参数 - 3. 调用函数 - 4. 序列化响应（使用绑定的 response_model 或自动生成） + 调用原始函数，支持从 payload 和 extra_kwargs 提取参数。 支持两种函数签名： - - 展开参数: def add(a: float, b: float) -> ... # 传统方式 - - Pydantic 模型: def add(req: AddGetRequest) -> ... # 工程化方式 + - 展开参数: def add(a: float, b: float) -> ... + - Pydantic 模型: def add(req: AddGetRequest) -> ... Args: payload: 请求参数字典 + **extra_kwargs: 额外关键字参数（如 session, worker_id） Returns: 序列化后的响应数据 """ - # 1. 校验请求参数 validated_payload = self.validate_request(payload) - # 2. 提取函数需要的参数 + sig = inspect.signature(self.func) + param_names = set(sig.parameters.keys()) - {'self'} 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) + model_type = sig.parameters[self.model_param].annotation 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] + param_names.discard(self.model_param) - # 3. 调用函数 + # 普通参数：优先 validated_payload，其次 extra_kwargs + for name in param_names: + if name in validated_payload: + kwargs[name] = validated_payload[name] + elif name in extra_kwargs: + kwargs[name] = extra_kwargs[name] + # 否则函数使用默认值（如果有） + + # 调用函数 if inspect.iscoroutinefunction(self.func): result = await self.func(**kwargs) else: result = self.func(**kwargs) - # 4. 序列化响应 return self.serialize_response(result) diff --git a/utils/worker/src/chongming_worker/runtime.py b/utils/worker/src/chongming_worker/runtime.py index 284fa75..0d496b5 100644 --- a/utils/worker/src/chongming_worker/runtime.py +++ b/utils/worker/src/chongming_worker/runtime.py @@ -64,15 +64,13 @@ async def create_worker( def make_callback(subject: str, methods: List[RPCMethod]): """创建处理特定 subject 的回调函数""" - def callback(query: zenoh.Query): + loop = asyncio.get_running_loop() + async def _handle_query(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) + data = orjson.loads(payload_bytes) if payload_bytes else {} http_method = data.get("_method", "GET").upper() # 匹配方法（按 HTTP 方法） @@ -87,7 +85,11 @@ async def create_worker( return # 调用 RPC 方法（支持同步和异步） - result = asyncio.run(matched.call(data)) + result = await matched.call( + data, + session=session, + worker_id=worker_id + ) # 回复 query.reply( @@ -107,6 +109,11 @@ async def create_worker( payload=json.dumps(error_response).encode("utf-8") ) print(f"[Worker] {worker_id} 异常 {subject}: {e}") + + def callback(query: zenoh.Query): + # 当前 create_worker 运行在事件循环中，可直接 create_task + asyncio.run_coroutine_threadsafe(_handle_query(query), loop) + return callback # 声明所有 Queryable @@ -119,6 +126,7 @@ async def create_worker( print(f"[Worker] {worker_id} Queryable: '{key}' [{', '.join(m.method for m in methods)}]") # 6. 保持 worker 存活直到被取消 + stop_event = asyncio.Event() try: await stop_event.wait() except asyncio.CancelledError: