"""API_PULL 对端 Mock（任务书 P1-D）。 契约严格对齐 MdpApiPullExecutor.cs 与既有 doc/db/mdp/mock_api/mock_server.py： - GET 单次请求，不分页；响应数组位于 data.list（可切换 responsePath）； - 每行去重键 bizKey（dedup_key_path）；?cursor=<上批最后一行 bizKey> 只返回更大行； - 鉴权模式 NONE / TOKEN(Bearer) / BASIC / APIKEY，对齐执行器 ApplyAuth 分支； - 可注入延迟与强制 HTTP 错误（401/403/429/500）； - 端点复用 doc/db/mdp/mock_api/endpoints.json 与 samples/*.json（不迁移、不修改）， 另支持运行时 override：把模拟器适配后的样例（带 SIM 前缀）挂到任意路径。 每次调用写入内存审计（Header 脱敏），Secret 值以 redact_text 二次遮蔽。 """ from __future__ import annotations import base64 import json import sys import time from dataclasses import dataclass, field from pathlib import Path _SIM_ROOT = Path(__file__).resolve().parents[1] if str(_SIM_ROOT) not in sys.path: sys.path.insert(0, str(_SIM_ROOT)) from fastapi import FastAPI, HTTPException, Query, Request # noqa: E402 from catalog.sample_adapters import MOCK_API_DIR # noqa: E402 from clients.redaction import redact_headers # noqa: E402 @dataclass class MockApiSettings: auth_mode: str = "TOKEN" # NONE | TOKEN | BASIC | APIKEY token: str = "sim-mock-token" basic_user: str = "sim" basic_password: str = "sim-basic-password" api_key_header: str = "X-Api-Key" api_key_value: str = "sim-api-key" delay_seconds: float = 0.0 force_status: int | None = None # 401/403/429/500 等 response_path: str = "data.list" audit_limit: int = 200 @dataclass class MockApiState: settings: MockApiSettings = field(default_factory=MockApiSettings) overrides: dict[str, list[dict]] = field(default_factory=dict) audit: list[dict] = field(default_factory=list) def secrets(self) -> list[str]: s = self.settings return [v for v in (s.token, s.basic_password, s.api_key_value) if v] def _load_endpoints() -> dict[str, str]: with (MOCK_API_DIR / "endpoints.json").open("r", encoding="utf-8") as f: return json.load(f) def _load_rows(rel_path: str) -> list[dict]: p = MOCK_API_DIR / rel_path if not p.exists(): raise HTTPException(status_code=404, detail=f"sample not found: {rel_path}") with p.open("r", encoding="utf-8") as f: return json.load(f) def _build_response(rows: list[dict], response_path: str) -> dict: parts = [p for p in response_path.split(".") if p] body: dict = {} node = body for part in parts[:-1]: node[part] = {} node = node[part] node[parts[-1]] = rows if parts else rows body.setdefault("code", 0) body.setdefault("message", "ok") return body def _check_auth(state: MockApiState, request: Request) -> None: s = state.settings mode = (s.auth_mode or "NONE").upper() if mode == "NONE": return authorization = request.headers.get("Authorization") if mode in ("TOKEN", "OAUTH2"): if authorization != f"Bearer {s.token}": raise HTTPException(status_code=401, detail="invalid or missing bearer token") elif mode == "BASIC": expected = "Basic " + base64.b64encode(f"{s.basic_user}:{s.basic_password}".encode()).decode() if authorization != expected: raise HTTPException(status_code=401, detail="invalid or missing basic credentials") elif mode == "APIKEY": if request.headers.get(s.api_key_header) != s.api_key_value: raise HTTPException(status_code=401, detail=f"invalid or missing api key header {s.api_key_header}") else: raise HTTPException(status_code=401, detail=f"unsupported auth mode on mock side: {mode}") def _audit(state: MockApiState, entry: dict) -> None: state.audit.append(entry) if len(state.audit) > state.settings.audit_limit: del state.audit[: len(state.audit) - state.settings.audit_limit] def create_app(state: MockApiState | None = None) -> FastAPI: state = state or MockApiState() app = FastAPI(title="Ai-DOP Integration Simulator - Mock API (API_PULL)", version="1.0.0") app.state.sim = state @app.get("/health") def health(): return {"status": "UP", "authMode": state.settings.auth_mode} @app.get("/__endpoints") def list_endpoints(): return { "authMode": state.settings.auth_mode, "responsePath": state.settings.response_path, "endpoints": _load_endpoints(), "overrides": sorted(state.overrides.keys()), } @app.get("/__control/settings") def get_settings(): s = state.settings # 不回显任何凭据值 return { "authMode": s.auth_mode, "delaySeconds": s.delay_seconds, "forceStatus": s.force_status, "responsePath": s.response_path, "apiKeyHeader": s.api_key_header, } @app.post("/__control/settings") def update_settings(payload: dict): s = state.settings if "auth_mode" in payload: mode = str(payload["auth_mode"]).upper() if mode not in ("NONE", "TOKEN", "BASIC", "APIKEY", "OAUTH2"): raise HTTPException(status_code=400, detail=f"unknown auth_mode: {mode}") s.auth_mode = mode if "token" in payload and payload["token"]: s.token = str(payload["token"]) if "basic_user" in payload and payload["basic_user"]: s.basic_user = str(payload["basic_user"]) if "basic_password" in payload and payload["basic_password"]: s.basic_password = str(payload["basic_password"]) if "api_key_header" in payload and payload["api_key_header"]: s.api_key_header = str(payload["api_key_header"]) if "api_key_value" in payload and payload["api_key_value"]: s.api_key_value = str(payload["api_key_value"]) if "delay_seconds" in payload: s.delay_seconds = max(0.0, float(payload["delay_seconds"])) if "force_status" in payload: s.force_status = int(payload["force_status"]) if payload["force_status"] else None if "response_path" in payload and payload["response_path"]: s.response_path = str(payload["response_path"]) return get_settings() @app.post("/__control/overrides") def set_override(payload: dict): path = str(payload.get("path") or "") rows = payload.get("rows") if not path.startswith("/api/") or not isinstance(rows, list): raise HTTPException(status_code=400, detail="need {path:/api/..., rows:[...]}") state.overrides[path] = rows return {"path": path, "rowCount": len(rows)} @app.delete("/__control/overrides") def clear_overrides(): state.overrides.clear() return {"cleared": True} @app.get("/__control/audit") def get_audit(): return {"entries": state.audit[-50:]} @app.get("/api/{obj:path}") def get_object( obj: str, request: Request, cursor: str | None = Query(default=None, description="上批最后一行 bizKey（增量游标）"), ): started = time.time() path = request.url.path if state.settings.delay_seconds > 0: time.sleep(state.settings.delay_seconds) try: if state.settings.force_status: raise HTTPException(status_code=state.settings.force_status, detail=f"forced error {state.settings.force_status}") _check_auth(state, request) if path in state.overrides: rows = [dict(r) for r in state.overrides[path]] else: endpoints = _load_endpoints() if path not in endpoints: raise HTTPException(status_code=404, detail=f"no mock for {path}; add it to endpoints.json or set an override") rows = _load_rows(endpoints[path]) rows.sort(key=lambda r: str(r.get("bizKey", ""))) total = len(rows) if cursor: rows = [r for r in rows if str(r.get("bizKey", "")) > cursor] return _build_response(rows, state.settings.response_path) finally: _audit(state, { "time": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "path": path, "cursor": cursor, "authMode": state.settings.auth_mode, "forceStatus": state.settings.force_status, "elapsedMs": int((time.time() - started) * 1000), "headers": redact_headers(dict(request.headers), state.secrets()), }) return app