""" ApiDataSource - API 数据源协调层 职责： - 持有 ApiClient 实例，提供给 Handler 使用 - 实现 DataSource 接口（load_all, sync_all） - 协调多个 Handler 的执行顺序 - 统一异常处理和日志记录 """ import json from typing import Optional from pathlib import Path from ..datasource import BaseDataSource from ..handler import NodeHandler from ..task_result import TaskResult from .client import ApiClient class ApiDataSource(BaseDataSource): """基于 API 的数据源实现""" _TRACE_MAX_PARAMS_LEN = 1000 _TRACE_MAX_RESPONSE_LEN = 1000 def __init__( self, base_url: str, uid: str, secret: str, poll_mode: str = "async", debug: bool = False, log_file: Optional[Path] = None, poll_max_retries: int = 1, poll_interval: float = 0.5, rate_limit_max_requests: int = 30, rate_limit_window_seconds: float = 10.0, ): """ 初始化 API 数据源 Args: base_url: API 基础 URL uid: 用户标识 secret: 签名密钥 debug: 是否启用调试模式 log_file: 日志文件路径（可选） """ super().__init__(poll_max_retries=poll_max_retries, poll_interval=poll_interval) self.client = ApiClient( base_url, uid, secret, debug=debug, log_file=log_file, rate_limit_max_requests=rate_limit_max_requests, rate_limit_window_seconds=rate_limit_window_seconds, ) self.poll_mode = poll_mode async def initialize(self): """初始化数据源（初始化 HTTP 客户端）""" await self.client.initialize() async def close(self): """关闭数据源（关闭 HTTP 客户端）""" await self.client.close() def add_handler(self, handler: NodeHandler) -> None: """ 添加业务 Handler Args: handler: API Handler 实例 """ handler.set_api_client(self.client) handler.set_poll_mode(self.poll_mode) # 调用基类添加方法 super().add_handler(handler) # 如果 collection 已设置，确保新 handler 拿到引用 if self._collection is not None: handler.set_collection(self._collection) # type: ignore[arg-type] @staticmethod def _fmt_trace_value(value, max_len: int = 600) -> str: try: text = json.dumps(value, ensure_ascii=False, separators=(",", ":")) except Exception: text = str(value) if len(text) > max_len: return text[:max_len] + "..." return text def _append_trace_log(self, node, trace: dict, *, stage: str) -> None: method = str(trace.get("method", "")).upper() endpoint = str(trace.get("endpoint", "")) url = str(trace.get("url", "")) params = trace.get("params_text") if not isinstance(params, str): params = self._fmt_trace_value(trace.get("params", {}), max_len=self._TRACE_MAX_PARAMS_LEN) payload = trace.get("payload_text") if not isinstance(payload, str): payload = self._fmt_trace_value(trace.get("payload", {}), max_len=self._TRACE_MAX_PARAMS_LEN) response = trace.get("response_text") if not isinstance(response, str): response = self._fmt_trace_value(trace.get("response", {}), max_len=self._TRACE_MAX_RESPONSE_LEN) elapsed_ms = trace.get("elapsed_ms", 0) node.append_log(f"API链路[{stage}] {method} {endpoint}") node.append_log(f" url={url}") if params not in {"{}", "null"}: node.append_log(f" params={params}") if payload not in {"{}", "null"}: node.append_log(f" payload={payload}") node.append_log(f" response={response}") node.append_log(f" elapsed_ms={elapsed_ms}") def _consume_and_append_push_traces(self, node, push_id: str, *, stage: str) -> int: if not push_id: return 0 traces = self.client.consume_traces_by_push_id(push_id) for trace in traces: self._append_trace_log(node, trace, stage=stage) return len(traces) @staticmethod def _extract_push_id_from_trace(trace: dict | None) -> Optional[str]: if trace is None: return None payload = trace.get("payload") if not isinstance(payload, dict): return None push_id = payload.get("push_id") return push_id if isinstance(push_id, str) and push_id else None def _resolve_push_id_for_success(self, node, result: TaskResult) -> Optional[str]: if result.task_id: return result.task_id if result.push_id: return result.push_id action_name = node.action.value.lower() biz_id = node.data_id or "" if biz_id: by_biz_and_op = self.client.peek_latest_trace_by_biz_id(biz_id, op_kind=action_name) if by_biz_and_op is not None: push_id = self._extract_push_id_from_trace(by_biz_and_op) if push_id: self.client.consume_latest_trace_by_biz_id(biz_id, op_kind=action_name) return push_id by_biz_any = self.client.peek_latest_trace_by_biz_id(biz_id) if by_biz_any is not None: push_id = self._extract_push_id_from_trace(by_biz_any) if push_id: self.client.consume_latest_trace_by_biz_id(biz_id) return push_id return None def _append_missing_push_id_debug_trace(self, node) -> None: action_name = node.action.value.lower() biz_id = node.data_id or "" if biz_id: trace = self.client.peek_latest_trace_by_biz_id(biz_id, op_kind=action_name) if trace is None: trace = self.client.peek_latest_trace_by_biz_id(biz_id) if trace is not None: self._append_trace_log(node, trace, stage="submit-missing-push-id") return node.append_log( f"API链路[submit-missing-push-id] 未找到匹配 biz_id 的请求记录，action={node.action.value}, data_id={node.data_id or ''}" ) return node.append_log( f"API链路[submit-missing-push-id] 未找到可用请求记录（已禁用按操作类型兜底），action={node.action.value}, data_id={node.data_id or ''}" ) return None def _on_node_loaded(self, handler: NodeHandler, node) -> None: data_id = node.data_id or "" if not data_id: return traces = self.client.consume_load_traces(node.node_type, data_id) for trace in traces: self._append_trace_log(node, trace, stage="load") async def _intercept_success_result(self, handler: NodeHandler, node, result: TaskResult, async_tasks: dict) -> bool: if node.action.value not in {"CREATE", "UPDATE", "DELETE"}: return False push_id = self._resolve_push_id_for_success(node, result) if not push_id: self._append_missing_push_id_debug_trace(node) raise RuntimeError( f"[{node.node_type}] unified polling requires push_id for {node.action.value}, " f"but none resolved for node={node.node_id}, data_id={node.data_id}" ) synthetic = TaskResult.in_progress(node_id=node.node_id, task_id=push_id) appended = self._consume_and_append_push_traces(node, push_id, stage="submit") if appended == 0: node.append_log(f"API链路[submit] 未捕获请求记录，push_id={push_id}") await self._handle_in_progress_result(node, synthetic, async_tasks) return True def _on_in_progress_result(self, handler: NodeHandler, node, result: TaskResult) -> None: if result.task_id: appended = self._consume_and_append_push_traces(node, result.task_id, stage="submit") if appended == 0: node.append_log(f"API链路[submit] 未捕获请求记录，push_id={result.task_id}") def _on_poll_result(self, handler: NodeHandler, node, task_id: str, result: TaskResult) -> None: appended = self._consume_and_append_push_traces(node, task_id, stage="poll") if appended == 0: node.append_log(f"API链路[poll] 未捕获请求记录，push_id={task_id}") def _on_failed_result(self, handler: NodeHandler, node, result: TaskResult) -> None: if node.action.value not in {"CREATE", "UPDATE", "DELETE"}: return if result.push_id: traces = self.client.consume_traces_by_push_id(result.push_id) if traces: for trace in traces: self._append_trace_log(node, trace, stage="submit-failed") return node.append_log(f"API链路[submit-failed] 未捕获请求记录，push_id={result.push_id}") action_name = node.action.value.lower() biz_id = node.data_id or "" trace = None if biz_id: trace = self.client.consume_latest_trace_by_biz_id(biz_id, op_kind=action_name) if trace is None: trace = self.client.consume_latest_trace_by_biz_id(biz_id) if trace is not None: self._append_trace_log(node, trace, stage="submit-failed") return if biz_id: node.append_log( f"API链路[submit-failed] 未找到匹配 biz_id 的请求记录，action={node.action.value}, data_id={biz_id}" ) return node.append_log( f"API链路[submit-failed] 未找到可用请求记录（已禁用按操作类型兜底），action={node.action.value}, data_id=" ) # load_all 和 sync_all 继承自 BaseDataSource # BaseDataSource 会按顺序调用 Handler 的方法