pplcplapi/app/cpl_client.py

"""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]