"""HTTP klient pro PPL CPL API. Zajišťuje: - získání a cachování OAuth Bearer tokenu (client_credentials, scope myapi2), - minimální rozestup mezi requesty (PPL vyžaduje >= 40 ms), - jeden retry s čerstvým tokenem, pokud PPL vrátí 401 (token mohl být revokován), - pomocné funkce pro předání odpovědi PPL klientovi (JSON / binární etikety). Secrets se nikdy nelogují — loguje se pouze metoda, cesta a status. """ import asyncio import time from typing import Any import httpx from fastapi.responses import JSONResponse, Response from . import token_cache from .config import ( CPL_OAUTH_SCOPE, MIN_REQUEST_INTERVAL_SECONDS, UPSTREAM_TIMEOUT_SECONDS, ) from .credentials import Credentials from .errors import CredentialsError, UpstreamError, raise_for_upstream from .logging_config import get_logger log = get_logger("pplcpl.client") TOKEN_PATH = "/login/getAccessToken" # Hlavičky PPL, které má smysl předat klientovi (paging, korelace, Location). _RELAY_HEADERS = ( "location", "x-correlation-id", "x-paging-total-items-count", "x-paging-offset", "x-paging-limit", "content-disposition", ) _client: httpx.AsyncClient | None = None _throttle_lock = asyncio.Lock() _last_request_at = 0.0 def _http() -> httpx.AsyncClient: global _client if _client is None: _client = httpx.AsyncClient(timeout=UPSTREAM_TIMEOUT_SECONDS) return _client async def close_client() -> None: global _client if _client is not None: await _client.aclose() _client = None async def _throttle() -> None: """PPL vyžaduje minimálně 40 ms rozestup mezi po sobě jdoucími requesty.""" global _last_request_at async with _throttle_lock: now = time.monotonic() wait = MIN_REQUEST_INTERVAL_SECONDS - (now - _last_request_at) if wait > 0: await asyncio.sleep(wait) _last_request_at = time.monotonic() async def _fetch_token(creds: Credentials) -> str: await _throttle() try: resp = await _http().post( creds.base_url + TOKEN_PATH, data={ "grant_type": "client_credentials", "client_id": creds.client_id.strip(), "client_secret": creds.client_secret.strip(), "scope": CPL_OAUTH_SCOPE, }, ) except httpx.HTTPError as exc: log.error("Token endpoint PPL nedostupný (%s): %s", creds.environment, exc.__class__.__name__) raise UpstreamError("PPL CPL API (token endpoint) je nedostupné.") from exc if resp.status_code != 200: log.warning( "PPL odmítlo vydání tokenu (%s): HTTP %s", creds.environment, resp.status_code ) raise CredentialsError( f"PPL odmítlo přihlašovací údaje při vydávání tokenu (HTTP {resp.status_code}).", detail=(resp.text or "")[:500], ) payload = resp.json() token = payload.get("access_token") if not token: log.error("Token endpoint PPL vrátil 200 bez access_token.") raise UpstreamError("PPL vrátilo neplatnou odpověď z token endpointu.") expires_in = float(payload.get("expires_in") or 1800) token_cache.store(creds.cache_key, token, expires_in) log.info("Vydán nový PPL token (%s), platnost %ss.", creds.environment, int(expires_in)) return token async def get_access_token(creds: Credentials, force_refresh: bool = False) -> str: creds.require() if not force_refresh: cached = token_cache.get_cached(creds.cache_key) if cached: return cached lock = await token_cache.acquire_lock(creds.cache_key) async with lock: if not force_refresh: cached = token_cache.get_cached(creds.cache_key) if cached: return cached return await _fetch_token(creds) async def cpl_request( creds: Credentials, method: str, path: str, *, params: Any = None, json_body: Any = None, content: bytes | None = None, content_type: str | None = None, files: Any = None, ) -> httpx.Response: """Provede autentizovaný request na PPL CPL API a vrátí surovou odpověď. Na 401 zkusí jednou obnovit token a request zopakovat. Chybové statusy NEvyhazuje — o mapování rozhoduje volající (typované endpointy mapují, generická proxy předává 1:1). """ token = await get_access_token(creds) for attempt in (1, 2): headers: dict[str, str] = {"Authorization": f"Bearer {token}"} if creds.accept_language: headers["Accept-Language"] = creds.accept_language if content_type and content is not None: headers["Content-Type"] = content_type await _throttle() try: resp = await _http().request( method, creds.base_url + path, params=params, json=json_body, content=content, files=files, headers=headers, ) except httpx.TimeoutException as exc: log.error("Timeout při volání PPL %s %s", method, path) raise UpstreamError(f"PPL CPL API neodpovědělo včas ({method} {path}).") from exc except httpx.HTTPError as exc: log.error( "Chyba spojení na PPL %s %s: %s", method, path, exc.__class__.__name__ ) raise UpstreamError(f"PPL CPL API je nedostupné ({method} {path}).") from exc if resp.status_code == 401 and attempt == 1: log.info("PPL vrátilo 401, obnovuji token a opakuji request.") token_cache.invalidate(creds.cache_key) token = await get_access_token(creds, force_refresh=True) continue if resp.status_code >= 400: log.warning("PPL %s %s -> HTTP %s", method, path, resp.status_code) else: log.info("PPL %s %s -> HTTP %s", method, path, resp.status_code) return resp raise UpstreamError("PPL CPL API opakovaně odmítlo request.") # pragma: no cover def _relay_headers(resp: httpx.Response) -> dict[str, str]: return { name: value for name, value in resp.headers.items() if name.lower() in _RELAY_HEADERS } def ensure_success(resp: httpx.Response) -> None: """Vyhodí typovanou chybu služby, pokud PPL vrátilo chybový status.""" if resp.status_code >= 400: raise_for_upstream( resp.status_code, resp.text, resp.headers.get("content-type", "") ) def relay_json(resp: httpx.Response) -> JSONResponse: """Předá JSON odpověď PPL klientovi vč. paging hlaviček. Chyby mapuje.""" ensure_success(resp) body = None if resp.content: body = resp.json() return JSONResponse( status_code=resp.status_code, content=body, headers=_relay_headers(resp) ) def relay_binary(resp: httpx.Response) -> Response: """Předá binární odpověď PPL (etikety PDF/ZPL/JPG...) klientovi. Chyby mapuje.""" ensure_success(resp) return Response( content=resp.content, status_code=resp.status_code, media_type=resp.headers.get("content-type", "application/octet-stream"), headers=_relay_headers(resp), ) def relay_raw(resp: httpx.Response) -> Response: """Předá odpověď PPL 1:1 (vč. chybových statusů) — pro generickou proxy.""" return Response( content=resp.content, status_code=resp.status_code, media_type=resp.headers.get("content-type"), headers=_relay_headers(resp), ) def batch_id_from_location(resp: httpx.Response) -> str: """Vytáhne batchId z Location hlavičky odpovědi POST shipment/order batch.""" location = resp.headers.get("location", "") if not location: log.error("PPL nevrátilo Location hlavičku u batch requestu.") raise UpstreamError("PPL nevrátilo Location hlavičku s batchId.") return location.rstrip("/").split("/")[-1]