http.py

"""Shared httpx client factory with tenacity retry logic."""

import json
from collections.abc import Callable
from typing import Any

import httpx
from tenacity import (
    retry,
    retry_if_exception,
    stop_after_attempt,
    wait_exponential,
)


def decode_json(response: httpx.Response, url: str = "") -> Any:
    """Parse a JSON body, raising ``httpx.DecodingError`` rather than a ValueError.

    ``json.JSONDecodeError`` subclasses ``ValueError``, so an upstream that
    answers HTTP 200 with an HTML error page would otherwise be caught by the
    ``except ValueError -> INVALID_INPUT`` arms in saskatchewan and manitoba and
    blamed on the caller. ``httpx.DecodingError`` is an ``httpx.HTTPError`` but
    NOT a ``ValueError``, so it bypasses those arms and reaches the catch-all
    every tool now has.

    Every shared portal client must decode through this — Phase 20.2 guarded
    only ``api_get``, which left the ArcGIS Hub, OGC WFS and Socrata paths
    exposed (Phase 20.3).
    """
    try:
        return response.json()
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        # UnicodeDecodeError, not JSONDecodeError, is what a body that is not
        # valid UTF-8/16/32 raises (b"\xff", a truncated multi-byte sequence).
        # It subclasses ValueError too, so guarding only JSONDecodeError left
        # the same masking in place for a mangled body.
        where = f" from {url}" if url else ""
        raise httpx.DecodingError(
            f"upstream returned an undecodable body{where}: "
            f"{type(exc).__name__}: {exc}"
        ) from exc


def decode_json_bytes(content: bytes, url: str = "") -> Any:
    """``decode_json`` for callers holding raw bytes (OGC WFS reads .content)."""
    try:
        return json.loads(content)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        # See decode_json — UnicodeDecodeError is also a ValueError subclass.
        where = f" from {url}" if url else ""
        raise httpx.DecodingError(
            f"upstream returned an undecodable body{where}: "
            f"{type(exc).__name__}: {exc}"
        ) from exc


def is_retryable(exc: BaseException) -> bool:
    """Return True if the exception warrants a retry."""
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in {429, 500, 502, 503, 504}
    if isinstance(exc, (httpx.ConnectError, httpx.TimeoutException)):
        return True
    return False


def with_retry(func: Callable) -> Callable:
    """Decorator that applies tenacity retry with exponential backoff."""
    return retry(
        retry=retry_if_exception(is_retryable),
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        reraise=True,
    )(func)


async def api_get(
    url: str,
    params: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    timeout: float = 30.0,
) -> Any:
    """Shared GET helper with retry, optional headers.

    Args:
        url: Full URL to fetch.
        params: Query parameters.
        headers: Optional HTTP headers (e.g. {"Accept": "application/json"}).
        timeout: Request timeout in seconds.

    Returns:
        Parsed JSON response.
    """
    @with_retry
    async def _fetch() -> Any:
        async with httpx.AsyncClient(timeout=timeout) as http:
            response = await http.get(url, params=params, headers=headers)
            response.raise_for_status()
            return decode_json(response, url)

    return await _fetch()