soothfast-sdk 0.4.4

# Code generated by `cargo soothfast sdk gen`. DO NOT EDIT.
# mypy: disable-error-code="no-any-return"
"""Shared runtime for soothfast-generated clients.

Everything behavioral lives here: transports with retry/backoff, error
mapping, wire decoding driven by dataclass type hints, and pagination.
Generated modules stay declarative (models and one thin method per route).
"""
from __future__ import annotations

import dataclasses
import random
import time
import types
import typing
from email.utils import parsedate_to_datetime
from urllib.parse import quote

import httpx

T = typing.TypeVar("T")

_RETRY_STATUSES = frozenset({429, 503})
_MAX_RETRY_SLEEP = 60.0
_IDEMPOTENT = frozenset({"GET", "HEAD", "OPTIONS", "PUT", "DELETE"})


class ApiError(Exception):
    """An error response from the API."""

    def __init__(
        self,
        status: int,
        *,
        error: str | None = None,
        message: str | None = None,
        retry_after: float | None = None,
    ) -> None:
        super().__init__(message or error or f"HTTP {status}")
        self.status = status
        self.error = error
        self.message = message
        self.retry_after = retry_after


class BadRequestError(ApiError):
    pass


class NotFoundError(ApiError):
    pass


class RateLimitError(ApiError):
    pass


class ServiceUnavailableError(ApiError):
    pass


class ServerError(ApiError):
    pass


_ERROR_TYPES: dict[int, type[ApiError]] = {
    400: BadRequestError,
    404: NotFoundError,
    429: RateLimitError,
    503: ServiceUnavailableError,
}


def _parse_retry_after(value: str | None) -> float | None:
    if not value:
        return None
    try:
        return float(value)
    except ValueError:
        pass
    try:
        return max(0.0, parsedate_to_datetime(value).timestamp() - time.time())
    except (TypeError, ValueError):
        return None


def _error_for(response: httpx.Response) -> ApiError:
    status = response.status_code
    error = message = None
    retry_after: float | None = None
    try:
        payload = response.json()
    except ValueError:
        payload = None
    if isinstance(payload, dict):
        error = payload.get("error")
        message = payload.get("message")
        seconds = payload.get("retry_after_seconds")
        if isinstance(seconds, (int, float)):
            retry_after = float(seconds)
    if retry_after is None:
        retry_after = _parse_retry_after(response.headers.get("Retry-After"))
    cls = _ERROR_TYPES.get(status)
    if cls is None:
        cls = ServerError if status >= 500 else ApiError
    return cls(status, error=error, message=message, retry_after=retry_after)


def decode(ty: typing.Any, value: typing.Any) -> typing.Any:
    """Decode a wire value into ``ty``, a type or typing annotation."""
    if ty is None or ty is type(None):
        return None
    if ty is typing.Any:
        return value
    origin = typing.get_origin(ty)
    if origin in (typing.Union, types.UnionType):
        return _decode_union(typing.get_args(ty), value)
    if origin is list:
        (item,) = typing.get_args(ty)
        return [decode(item, v) for v in value]
    if origin is dict:
        _, item = typing.get_args(ty)
        return {k: decode(item, v) for k, v in value.items()}
    if origin is typing.Literal:
        return value
    if dataclasses.is_dataclass(ty) and isinstance(ty, type):
        return _decode_dataclass(ty, value)
    return value


def _decode_union(arms: tuple[typing.Any, ...], value: typing.Any) -> typing.Any:
    if value is None and type(None) in arms:
        return None
    candidates = [a for a in arms if a is not type(None)]
    for arm in candidates:
        if _matches(arm, value):
            return decode(arm, value)
    for arm in candidates:
        try:
            return decode(arm, value)
        except (TypeError, ValueError, KeyError):
            continue
    raise ValueError(f"no union arm matched {type(value).__name__}")


def _matches(ty: typing.Any, value: typing.Any) -> bool:
    if dataclasses.is_dataclass(ty) and isinstance(ty, type):
        if not isinstance(value, dict):
            return False
        wire = getattr(ty, "_WIRE", {})
        return all(
            wire.get(f.name, f.name) in value
            for f in dataclasses.fields(ty)
            if f.default is dataclasses.MISSING
        )
    if ty in (str, int, float, bool):
        return isinstance(value, ty)
    if typing.get_origin(ty) is list:
        return isinstance(value, list)
    if typing.get_origin(ty) is dict:
        return isinstance(value, dict)
    return False


def _decode_dataclass(ty: type, value: typing.Any) -> typing.Any:
    if not isinstance(value, dict):
        raise TypeError(f"{ty.__name__} expects an object, got {type(value).__name__}")
    hints = typing.get_type_hints(ty)
    wire = getattr(ty, "_WIRE", {})
    kwargs = {}
    for f in dataclasses.fields(ty):
        key = wire.get(f.name, f.name)
        if key in value:
            kwargs[f.name] = decode(hints[f.name], value[key])
    return ty(**kwargs)


def to_wire(value: typing.Any) -> typing.Any:
    """Encode a dataclass (or container of them) into wire-shaped JSON."""
    if dataclasses.is_dataclass(value) and not isinstance(value, type):
        wire = getattr(type(value), "_WIRE", {})
        out = {}
        for f in dataclasses.fields(value):
            v = getattr(value, f.name)
            if v is None:
                continue
            out[wire.get(f.name, f.name)] = to_wire(v)
        return out
    if isinstance(value, (list, tuple)):
        return [to_wire(v) for v in value]
    if isinstance(value, dict):
        return {k: to_wire(v) for k, v in value.items()}
    return value


def query_of(pairs: dict[str, typing.Any]) -> dict[str, typing.Any]:
    """Drop ``None`` values, lowercase booleans, comma-join sequences."""
    out: dict[str, typing.Any] = {}
    for key, value in pairs.items():
        if value is None:
            continue
        if isinstance(value, bool):
            out[key] = "true" if value else "false"
        elif isinstance(value, (list, tuple)):
            out[key] = ",".join(str(v) for v in value)
        else:
            out[key] = value
    return out


def path_seg(value: typing.Any) -> str:
    """Render one path segment, percent-encoded."""
    return quote(str(value), safe="")


def _backoff(err: ApiError | None, attempt: int, base: float) -> float:
    if err is not None and err.retry_after is not None:
        return err.retry_after
    return base * (2**attempt) * (0.5 + random.random() / 2)


def _should_retry(method: str, err: ApiError | None, attempt: int, max_retries: int) -> bool:
    if attempt >= max_retries:
        return False
    if err is None:
        return method in _IDEMPOTENT
    if err.retry_after is not None and err.retry_after > _MAX_RETRY_SLEEP:
        return False
    return err.status in _RETRY_STATUSES


@dataclasses.dataclass(frozen=True)
class PageInfo:
    """Relay-style cursor metadata."""

    end_cursor: str | None
    has_next_page: bool


@dataclasses.dataclass(frozen=True)
class Page(typing.Generic[T]):
    """One page of a paginated listing."""

    items: list[T]
    page_info: PageInfo


def _page_from(data: typing.Any, item_ty: typing.Any) -> Page[typing.Any]:
    if not isinstance(data, dict):
        items = data if isinstance(data, list) else []
        return Page([decode(item_ty, v) for v in items], PageInfo(None, False))
    raw_items = data.get("items")
    if raw_items is None:
        raw_items = next((v for v in data.values() if isinstance(v, list)), [])
    info = data.get("pageInfo") or {}
    page_info = PageInfo(info.get("endCursor"), bool(info.get("hasNextPage")))
    return Page([decode(item_ty, v) for v in raw_items], page_info)


class SyncPager(typing.Generic[T]):
    """Iterates items across pages; ``pages()`` yields whole pages."""

    def __init__(
        self,
        fetch: typing.Callable[[str | None], typing.Any],
        item_ty: typing.Any,
    ) -> None:
        self._fetch = fetch
        self._item_ty = item_ty

    def pages(self) -> typing.Iterator[Page[T]]:
        cursor: str | None = None
        while True:
            page = _page_from(self._fetch(cursor), self._item_ty)
            yield page
            if not page.page_info.has_next_page or page.page_info.end_cursor is None:
                return
            cursor = page.page_info.end_cursor

    def __iter__(self) -> typing.Iterator[T]:
        for page in self.pages():
            yield from page.items


class AsyncPager(typing.Generic[T]):
    """Async counterpart of :class:`SyncPager`."""

    def __init__(
        self,
        fetch: typing.Callable[[str | None], typing.Awaitable[typing.Any]],
        item_ty: typing.Any,
    ) -> None:
        self._fetch = fetch
        self._item_ty = item_ty

    async def pages(self) -> typing.AsyncIterator[Page[T]]:
        cursor: str | None = None
        while True:
            page = _page_from(await self._fetch(cursor), self._item_ty)
            yield page
            if not page.page_info.has_next_page or page.page_info.end_cursor is None:
                return
            cursor = page.page_info.end_cursor

    async def __aiter__(self) -> typing.AsyncIterator[T]:
        async for page in self.pages():
            for item in page.items:
                yield item


class Transport:
    """Synchronous HTTP transport with retry and rate-limit handling."""

    def __init__(
        self,
        base_url: str,
        *,
        timeout: float = 30.0,
        max_retries: int = 3,
        backoff_base: float = 0.5,
        on_rate_limit: typing.Callable[[RateLimitError], None] | None = None,
        client: httpx.Client | None = None,
    ) -> None:
        self._client = client or httpx.Client(base_url=base_url, timeout=timeout)
        self._max_retries = max_retries
        self._backoff_base = backoff_base
        self._on_rate_limit = on_rate_limit

    def close(self) -> None:
        self._client.close()

    def request(
        self,
        method: str,
        path: str,
        *,
        query: dict[str, typing.Any] | None = None,
        json_body: typing.Any = None,
        into: typing.Any = None,
    ) -> typing.Any:
        attempt = 0
        while True:
            err: ApiError | None = None
            status: int | None = None
            try:
                response = self._client.request(method, path, params=query, json=json_body)
                status = response.status_code
                if status < 400:
                    return _decoded(response, into)
                err = _error_for(response)
            except httpx.TransportError as exc:
                if not _should_retry(method, None, attempt, self._max_retries):
                    raise ApiError(0, message=str(exc)) from exc
                time.sleep(_backoff(None, attempt, self._backoff_base))
                attempt += 1
                continue
            if isinstance(err, RateLimitError) and self._on_rate_limit is not None:
                self._on_rate_limit(err)
            if not _should_retry(method, err, attempt, self._max_retries):
                raise err
            time.sleep(_backoff(err, attempt, self._backoff_base))
            attempt += 1


class AsyncTransport:
    """Asynchronous HTTP transport with retry and rate-limit handling."""

    def __init__(
        self,
        base_url: str,
        *,
        timeout: float = 30.0,
        max_retries: int = 3,
        backoff_base: float = 0.5,
        on_rate_limit: typing.Callable[[RateLimitError], None] | None = None,
        client: httpx.AsyncClient | None = None,
    ) -> None:
        self._client = client or httpx.AsyncClient(base_url=base_url, timeout=timeout)
        self._max_retries = max_retries
        self._backoff_base = backoff_base
        self._on_rate_limit = on_rate_limit

    async def aclose(self) -> None:
        await self._client.aclose()

    async def request(
        self,
        method: str,
        path: str,
        *,
        query: dict[str, typing.Any] | None = None,
        json_body: typing.Any = None,
        into: typing.Any = None,
    ) -> typing.Any:
        import asyncio

        attempt = 0
        while True:
            err: ApiError | None = None
            status: int | None = None
            try:
                response = await self._client.request(method, path, params=query, json=json_body)
                status = response.status_code
                if status < 400:
                    return _decoded(response, into)
                err = _error_for(response)
            except httpx.TransportError as exc:
                if not _should_retry(method, None, attempt, self._max_retries):
                    raise ApiError(0, message=str(exc)) from exc
                await asyncio.sleep(_backoff(None, attempt, self._backoff_base))
                attempt += 1
                continue
            if isinstance(err, RateLimitError) and self._on_rate_limit is not None:
                self._on_rate_limit(err)
            if not _should_retry(method, err, attempt, self._max_retries):
                raise err
            await asyncio.sleep(_backoff(err, attempt, self._backoff_base))
            attempt += 1


def _decoded(response: httpx.Response, into: typing.Any) -> typing.Any:
    if response.status_code == 204 or not response.content:
        return None
    if into is None:
        return response.json()
    return decode(into, response.json())