""" BaseApiHandler - API Handler 基类 职责： - 实现 Handler 接口（load, create, update, delete, poll） - 使用 self.api_client 发送 HTTP 请求 - 生成 push_id 和 biz_id（业务标识） - 从 collection 获取依赖节点数据 - 管理节点的 context（存储 project_id, contract_id 等） - 实现 poll 方法，调用 api_client.get_push_logs 批量查询 - 提取 created_id（因为不同业务的 id 字段名和位置可能不同） - 返回 TaskResult（SUCCESS/FAILED/IN_PROGRESS） """ import uuid import asyncio from abc import abstractmethod from copy import deepcopy from typing import ClassVar, Dict, List, Any, Optional, TYPE_CHECKING, TypeVar from pydantic import BaseModel, ValidationError from ...common.type_safety import get_pydantic_model_fields from ..handler import BaseNodeHandler from ..task_result import TaskResult from ..handler import ValidationResult # Import ValidationResult if TYPE_CHECKING: from ...common.sync_node import SyncNode from ...common.collection import DataCollection from .client import ApiClient T = TypeVar("T", bound=BaseModel) class BaseApiHandler(BaseNodeHandler[T]): """API Handler 基类""" # 子类需要设置这些属性 _node_type: str _node_class: type['SyncNode'] _schema: Any def __init__(self, **handler_config: Any): # 子类应该在调用 super().__init__() 之前或之后设置 _node_type, _node_class, _schema super().__init__() self.api_client: 'ApiClient' = None # type: ignore # 由 DataSource 注入 self.poll_mode: str = "async" # async | sync self._collection: 'DataCollection' = None # type: ignore # 由 DataSource 注入 self.handler_config = dict(handler_config) # 子类在 __init__ 中设置此列表以声明 create/update 中涉及的 Pydantic schema 类； # post-check 将只比较这些 schema 覆盖的字段，过滤后端只读的派生字段。 self.update_schemas: List[type] = [] def get_update_fields(self, *, exclude: frozenset = frozenset({'id'})) -> List[str]: """从 update_schemas 中提取所有可写字段名，排除标识键（默认排除 id）。""" fields: set = set() for schema in self.update_schemas: for field_name in get_pydantic_model_fields(schema): if field_name not in exclude: fields.add(field_name) return sorted(fields) def set_api_client(self, client: 'ApiClient') -> None: self.api_client = client def set_collection(self, collection: 'DataCollection') -> None: """设置 Collection（由 DataSource 调用）""" self._collection = collection @property def node_type(self) -> str: """返回节点类型""" return self._node_type @property def node_class(self) -> type['SyncNode']: """返回节点类""" return self._node_class @property def schema(self) -> Any: """返回 schema（子类可重写）""" return self._schema def _generate_push_id(self) -> str: """ 生成 push_id（推送唯一标识） Returns: UUID 格式的 push_id """ return str(uuid.uuid4()) def _generate_biz_id(self, node: 'SyncNode') -> str: """ 生成 biz_id（业务标识） 子类可重写此方法实现自定义业务 ID 生成逻辑。 默认使用 node_id 作为 biz_id。 Args: node: 同步节点 Returns: 业务标识符 """ return node.node_id def _create_basic_node(self, data: Dict[str, Any], depend_ids: Optional[List[str]] = None) -> "SyncNode": """创建或复用基础节点（默认使用 data['id'] 或 data['_id']）""" data_id = str(data.get("id") or data.get("_id") or "") if self._collection is None: raise RuntimeError("Collection not set. Call set_collection() before creating nodes.") if data_id: existing_node = self._collection.get_by_data_id(self.node_type, data_id) if existing_node is not None: existing_node.depend_ids = list(depend_ids or []) existing_node.set_data(data.copy()) existing_node.set_origin_data(data.copy()) return existing_node node_id = self._collection.generate_node_id() node = self.node_class( node_id=node_id, data_id=data_id, data=data.copy(), depend_ids=depend_ids or [], origin_data=data.copy(), ) return node def _create_node(self, data: Dict[str, Any]) -> "SyncNode": """默认节点创建（子类可覆盖并补充 context）""" return self._create_basic_node(data) def _get_committed_approval_snapshot(self, node: 'SyncNode') -> Dict[str, Any]: snapshot = node.context.get("approval_snapshot") return snapshot if isinstance(snapshot, dict) else {} def _get_pending_approval_snapshot(self, node: 'SyncNode') -> Dict[str, Any]: snapshot = node.context.get("pending_approval_snapshot") return snapshot if isinstance(snapshot, dict) else {} def _get_active_approval_snapshot(self, node: 'SyncNode') -> Dict[str, Any]: pending_snapshot = self._get_pending_approval_snapshot(node) if pending_snapshot: return pending_snapshot return self._get_committed_approval_snapshot(node) def _get_request_steps(self, node: 'SyncNode') -> List[Dict[str, Any]]: snapshot = self._get_active_approval_snapshot(node) steps = snapshot.get("steps") if isinstance(steps, list): return deepcopy(steps) return [] def _steps_changed(self, node: 'SyncNode') -> bool: snapshot = self._get_active_approval_snapshot(node) return bool(snapshot.get("steps_changed")) def _attach_request_steps(self, request_data: Dict[str, Any], node: Optional['SyncNode']) -> Dict[str, Any]: payload = dict(request_data) payload["steps"] = self._get_request_steps(node) if node is not None else [] return payload def _mark_approval_snapshot_synced(self, node: 'SyncNode') -> None: snapshot = self._get_active_approval_snapshot(node) if snapshot: committed_snapshot = deepcopy(snapshot) committed_snapshot.pop("steps_changed", None) node.context["approval_snapshot"] = committed_snapshot node.context.pop("pending_approval_snapshot", None) def extract_id(self, data: Dict[str, Any]) -> str: """从原始数据提取 ID（默认优先 id，其次 _id）""" return str(data.get("id") or data.get("_id") or "") def extract_created_id(self, push_log: Dict[str, Any]) -> Optional[str]: """ 从 push_log 提取 created_id 子类可重写此方法以适配不同的响应结构。 默认从 push_log['data']['biz_id'] 提取。 Args: push_log: Push Log 响应数据 Returns: 创建的对象 ID，未找到返回 None """ try: data = push_log.get('data', {}) if isinstance(data, dict): return data.get('biz_id') return None except (KeyError, AttributeError): return None def set_poll_mode(self, mode: str) -> None: """设置轮询模式（sync/async）""" if mode not in {"sync", "async"}: raise ValueError(f"Unsupported poll mode: {mode}") self.poll_mode = mode def _extract_created_id_from_response(self, response: Dict[str, Any]) -> Optional[str]: """从同步创建响应提取创建的 ID（默认 data.biz_id）""" data = response.get("data") if isinstance(response, dict) else None if isinstance(data, dict): return data.get("biz_id") return None def _build_create_task_result( self, node_id: str, response: Optional[Dict[str, Any]] = None, task_id: Optional[str] = None, fallback_data_id: Optional[str] = None, ) -> TaskResult: """根据轮询模式构建 create 的 TaskResult""" if self.poll_mode == "sync": created_id = None if response is not None: created_id = self._extract_created_id_from_response(response) if not created_id: created_id = fallback_data_id return TaskResult.success(node_id=node_id, data_id=created_id) if not task_id: return TaskResult.failed(node_id=node_id, error="Missing task_id for async create") return TaskResult.in_progress(node_id=node_id, task_id=task_id) # ========== 通用辅助方法 ========== def _validate_schema(self, data: Dict[str, Any], schema: Any) -> Optional[str]: """ 验证数据是否符合 schema Args: data: 待验证数据 schema: Pydantic schema Returns: 错误信息，无错误返回 None """ try: schema.model_validate(data) return None except Exception as e: return str(e) def _get_update_schema_payload( self, schema: Any, data: Dict[str, Any], *, require_valid: bool, ) -> Dict[str, Any] | None: """Update 内部实现：提取指定 schema 下的 payload。 这里明确只服务“update schema”，不负责 response model 一致性校验。 这个方法只给 BaseApiHandler 的 update 入口方法复用，不作为 handler 入口。 处理顺序： 1. 先尝试用 update schema 校验，成功则直接使用校验后的 model_dump。 2. 如果校验失败且 require_valid=False，则退化为只抽取 schema 定义的字段。 之所以允许退化，是因为 update 比较时两边快照常常不完整： 某些字段在旧快照缺失，但新数据已经有值。此时应该继续比较 schema 相关字段，而不是把它误判成“无差异”。 空字符串和 None 会被统一过滤掉，因为当前 update handler 并不把它们 当成“清空字段”的显式语义，而是当作“不参与本次更新”。 """ try: validated = schema.model_validate(data) payload = validated.model_dump(exclude_unset=True) except ValidationError: if require_valid: return None schema_fields = get_pydantic_model_fields(schema) if not schema_fields: return {} payload = { field_name: data.get(field_name) for field_name in schema_fields.keys() if field_name in data } return { field_name: value for field_name, value in payload.items() if value not in (None, "") } def get_update_schema_diff( self, schema, data: Dict[str, Any], origin_data: Dict[str, Any], node_id: Optional[str] = None, force_full_schema: bool = False, ) -> Dict[str, Any] | None: """Update 入口：获取指定 update schema 下的差异字段。 Args: schema: Pydantic schema 类 data: 新数据 origin_data: 原始数据 node_id: 节点ID（用于日志） Returns: Dict[str, Any] | None: 有差异返回差异字段字典，无差异或字段不全返回 None """ # 当前待发送的数据必须能构成合法 update payload；允许旧快照不完整。 new_dict = self._get_update_schema_payload(schema, data, require_valid=True) origin_dict = self._get_update_schema_payload(schema, origin_data, require_valid=False) if new_dict is None or origin_dict is None: return None if new_dict == origin_dict: if force_full_schema: return new_dict return None diff_fields = {k: v for k, v in new_dict.items() if origin_dict.get(k) != v} if diff_fields: node_info = f"[{node_id}]" if node_id else "" print(f" [DIFF] {self.node_type}{node_info} schema={schema.__name__}") for k, v in diff_fields.items(): orig_v = origin_dict.get(k) print(f" • {k}: {orig_v!r} → {v!r}") return diff_fields if diff_fields else None def build_update_request_data( self, new_data: Dict[str, Any], original_data: Dict[str, Any], schema: Optional[Any] = None, force_full_schema: bool = False, ) -> Dict[str, Any]: """Update 入口：构造最终要发送给更新接口的请求数据。 规则： 1. 提取 schema 定义的所有字段（包括必填） 2. 检查是否有任何字段发生变化 3. 如果有变化，返回所有 schema 字段；否则返回空 Args: new_data: 新数据 original_data: 原始数据 schema: 更新 schema（可选） Returns: 包含 schema 所有字段的数据（如果有变化），否则返回空字典 """ schema_fields = get_pydantic_model_fields(schema) if not schema_fields: # 无 schema，按原逻辑处理 update_data = {} for key, new_value in new_data.items(): if new_value == "" or new_value is None: continue original_value = original_data.get(key) if new_value != original_value: update_data[key] = new_value return update_data schema_data = self._get_update_schema_payload(schema, new_data, require_valid=True) or {} origin_schema_data = self._get_update_schema_payload(schema, original_data, require_valid=False) or {} if force_full_schema or schema_data != origin_schema_data: return schema_data return {} def _schema_changed_fields( self, schema: Any, new_data: Dict[str, Any], origin_data: Dict[str, Any], ) -> List[str]: new_payload = self._get_update_schema_payload(schema, new_data, require_valid=True) or {} origin_payload = self._get_update_schema_payload(schema, origin_data, require_valid=False) or {} field_names = sorted(set(new_payload) | set(origin_payload)) return [field_name for field_name in field_names if new_payload.get(field_name) != origin_payload.get(field_name)] def build_update_schema_status( self, *, schema_name: str, schema: Any, new_data: Dict[str, Any], origin_data: Dict[str, Any], will_update: bool, update_error: Optional[str] = None, ) -> str: """Update 入口：生成 update schema 的状态说明文本。""" new_error = self._validate_schema(new_data, schema) origin_error = self._validate_schema(origin_data, schema) changed_fields = self._schema_changed_fields(schema, new_data, origin_data) if new_error: compact = new_error.replace("\n", " ").strip() if len(compact) > 220: compact = compact[:220] + "..." return f"{schema_name}: NEW_INVALID({compact})" if not changed_fields: return f"{schema_name}: CONSISTENT" changed_text = ",".join(changed_fields) if update_error: compact = update_error.replace("\n", " ").strip() if len(compact) > 220: compact = compact[:220] + "..." return f"{schema_name}: DIFF[{changed_text}] -> UPDATE_FAILED({compact})" if will_update: if origin_error: compact = origin_error.replace("\n", " ").strip() if len(compact) > 160: compact = compact[:160] + "..." return f"{schema_name}: DIFF[{changed_text}] -> UPDATE_TRIGGERED (origin_invalid={compact})" return f"{schema_name}: DIFF[{changed_text}] -> UPDATE_TRIGGERED" if origin_error: compact = origin_error.replace("\n", " ").strip() if len(compact) > 160: compact = compact[:160] + "..." return f"{schema_name}: DIFF[{changed_text}] -> NOT_TRIGGERED (origin_invalid={compact})" return f"{schema_name}: DIFF[{changed_text}] -> NOT_TRIGGERED" async def _execute_with_retry( self, operation, max_retries: int = 3, retry_delay: float = 1.0 ) -> Any: """ 通用重试逻辑 子类可重写此方法实现自定义重试策略。 默认策略：网络错误/5xx 重试，4xx 不重试。 Args: operation: 异步操作（协程函数） max_retries: 最大重试次数 retry_delay: 重试延迟（秒） Returns: 操作执行结果 Raises: 最后一次重试的异常 """ import httpx last_error = None for attempt in range(max_retries): try: return await operation() except httpx.HTTPStatusError as e: # 4xx 客户端错误不重试 if 400 <= e.response.status_code < 500: raise last_error = e if attempt < max_retries - 1: await asyncio.sleep(retry_delay * (attempt + 1)) except (httpx.NetworkError, httpx.TimeoutException) as e: # 网络错误重试 last_error = e if attempt < max_retries - 1: await asyncio.sleep(retry_delay * (attempt + 1)) # 重试次数用尽 if last_error: raise last_error else: raise RuntimeError("Retry failed with unknown error") # ========== 批量接口（子类必须实现） ========== @abstractmethod async def create_all(self, nodes: List['SyncNode']) -> List[TaskResult]: """ 批量创建节点 Args: nodes: 需要创建的节点列表（DataSource 已筛选为 CREATE + PENDING） Returns: List[TaskResult]，每个结果关联 node_id """ pass @abstractmethod async def update_all(self, nodes: List['SyncNode']) -> List[TaskResult]: """ 批量更新节点 Args: nodes: 需要更新的节点列表（DataSource 已筛选为 UPDATE + PENDING） Returns: List[TaskResult]，每个结果关联 node_id """ pass @abstractmethod async def delete_all(self, nodes: List['SyncNode']) -> List[TaskResult]: """ 批量删除节点 Args: nodes: 需要删除的节点列表（DataSource 已筛选为 DELETE + PENDING） Returns: List[TaskResult]，每个结果关联 node_id """ pass async def poll_tasks(self, task_ids: List[str]) -> Dict[str, TaskResult]: """ 批量轮询异步任务状态 Args: task_ids: 任务 ID 列表（push_id 列表） Returns: {task_id: TaskResult} 映射 """ if not task_ids: return {} if not self.api_client: raise RuntimeError("api_client not injected") # 批量查询 push_logs try: push_logs = await self.api_client.get_push_logs(task_ids) except Exception as e: # 查询失败，所有任务标记为失败 return { task_id: TaskResult.failed(error=f"Poll failed: {str(e)}") for task_id in task_ids } results = {} for task_id in task_ids: push_log = push_logs.get(task_id, {}) if 'error' in push_log: results[task_id] = TaskResult.failed(error=push_log['error']) elif push_log.get('code') == 200: # 通过code==200判断成功 created_id = self.extract_created_id(push_log) results[task_id] = TaskResult.success(data_id=created_id) elif push_log.get('code') and push_log.get('code') != 200: # code存在但不是200表示失败 results[task_id] = TaskResult.failed(error=push_log.get('message', 'Unknown error')) else: # 仍在进行中 results[task_id] = TaskResult.in_progress(task_id=task_id) return results async def validate(self, data: Dict[str, Any]) -> ValidationResult: """ 数据质量检查（使用 schema 验证） Args: data: 原始数据 Returns: ValidationResult（是否有效 + 错误列表） """ error = self._validate_schema(data, self.schema) if error: return ValidationResult(is_valid=False, errors=[error]) return ValidationResult(is_valid=True) # ========== 抽象方法：子类必须实现 ========== @abstractmethod async def load(self) -> List['SyncNode']: """ 从 API 加载数据并创建节点 Handler 负责： 1. 加载原始数据（可使用 self._collection 查询依赖） 2. 创建 SyncNode（调用 self.create_node） 3. 设置 context（如 project_id, contract_id 等） Returns: SyncNode 列表 """ pass