http_utils.py

"""Small HTTP helpers shared by both service clients."""

from __future__ import annotations

from typing import Any

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


DEFAULT_TIMEOUT = (10, 90)
MAX_ERROR_BODY = 1200


class ApiError(RuntimeError):
    """Raised for remote API, protocol, or transport failures."""


def create_session(user_agent: str = "SHPD-Translation/2.0") -> requests.Session:
    """Create a session with conservative retries for read-only requests."""

    retry = Retry(
        total=4,
        connect=4,
        read=4,
        status=4,
        backoff_factor=0.6,
        status_forcelist=(429, 500, 502, 503, 504),
        # PUT/PATCH calls in this project set complete values and are
        # idempotent. POST uploads are deliberately excluded to avoid a
        # duplicate async job after an ambiguous connection failure.
        allowed_methods=frozenset({"GET", "HEAD", "OPTIONS", "PUT", "PATCH"}),
        respect_retry_after_header=True,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session = requests.Session()
    session.headers.update({"User-Agent": user_agent})
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session


def request(
    session: requests.Session,
    method: str,
    url: str,
    *,
    service: str,
    expected: tuple[int, ...] = (200,),
    **kwargs: Any,
) -> requests.Response:
    kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
    try:
        response = session.request(method, url, **kwargs)
    except requests.RequestException as exc:
        raise ApiError(f"{service} request failed: {exc}") from exc

    if response.status_code not in expected:
        body = response.text.strip().replace("\x00", "")[:MAX_ERROR_BODY]
        detail = f": {body}" if body else ""
        raise ApiError(
            f"{service} returned HTTP {response.status_code} for "
            f"{method.upper()} {url}{detail}"
        )
    return response


def json_body(response: requests.Response, service: str) -> dict[str, Any]:
    try:
        body = response.json()
    except ValueError as exc:
        raise ApiError(f"{service} returned invalid JSON") from exc
    if not isinstance(body, dict):
        raise ApiError(f"{service} returned an unexpected JSON structure")
    return body