"""The transport underneath the client: one request, retried sensibly. Standard library only. "No dependencies" is the difference between a service adopting this client and copy-pasting a ``requests`` call it then maintains badly. If ``httpx`` is installed it is used instead, purely for connection pooling -- a caller pulling a thousand elements pays a TLS handshake per request on urllib. Nothing else differs and nothing requires it. """ from __future__ import annotations import json as _json import random import time import urllib.error import urllib.parse import urllib.request from typing import Any from .errors import ( HTTPError, NotFound, RateLimited, ServiceUnavailable, TransportError, Unauthorized, ) #: 429 and 503 are "not now"; 502/504 are a proxy in front. Not 500: that means #: ODMS took the request and broke on it, and repeating it breaks it again. RETRY_STATUSES = frozenset({429, 502, 503, 504}) DEFAULT_TIMEOUT_S = 30.0 DEFAULT_MAX_RETRIES = 3 #: First backoff step, doubled per attempt and jittered. DEFAULT_BACKOFF_S = 0.5 #: Past this, fail fast and let the caller's scheduler decide rather than #: holding a worker for an unbounded stretch. MAX_RETRY_AFTER_S = 120.0 class Response: """What came back: status, headers, and the body as bytes.""" __slots__ = ("status", "headers", "body", "url") def __init__(self, status: int, headers: dict, body: bytes, url: str): self.status = status # Header lookup is case-insensitive on the wire; normalise once. self.headers = {k.lower(): v for k, v in headers.items()} self.body = body self.url = url @property def text(self) -> str: charset = "utf-8" content_type = self.headers.get("content-type", "") if "charset=" in content_type: charset = content_type.split("charset=", 1)[1].split(";")[0].strip() return self.body.decode(charset, "replace") def json(self) -> Any: try: return _json.loads(self.text) except ValueError as exc: raise HTTPError( self.status, self.url, self.text, f"expected JSON, got {self.headers.get('content-type', 'no content type')}", ) from exc def _httpx_session(): """A pooled ``httpx.Client``, or ``None`` when httpx is not installed.""" try: import httpx except ImportError: return None return httpx.Client(follow_redirects=True) def _retry_after(headers: dict) -> float | None: """Retry-After in seconds, delta-seconds form only. The HTTP-date form is legal but rare here, and mis-parsing it into a very long sleep is worse than falling back to our own backoff. """ raw = headers.get("retry-after") if not raw: return None try: return max(0.0, float(raw)) except (TypeError, ValueError): return None def _raise_for_status(response: Response) -> None: if response.status < 400: return args = (response.status, response.url, response.text) if response.status == 404: raise NotFound(*args) if response.status in (401, 403): raise Unauthorized(*args) if response.status == 429: raise RateLimited(*args, retry_after=_retry_after(response.headers)) if response.status in (502, 503, 504): raise ServiceUnavailable(*args) raise HTTPError(*args) class Transport: """Issues requests, retries the ones worth retrying, and raises the rest. The policy is asymmetric on purpose. A GET that died mid-flight can be repeated; a POST cannot -- ``messages/push/`` that timed out may already have stored the message. So writes retry only on statuses that *prove* the request was not acted on, never on a transport failure of unknown outcome. """ def __init__(self, base_url: str, *, token: str | None = None, timeout_s: float = DEFAULT_TIMEOUT_S, max_retries: int = DEFAULT_MAX_RETRIES, backoff_s: float = DEFAULT_BACKOFF_S, user_agent: str = "odms-client/1.0", session: Any = None): self.base_url = base_url.rstrip("/") self.token = token self.timeout_s = timeout_s self.max_retries = max_retries self.backoff_s = backoff_s self.user_agent = user_agent self._session = session if session is not None else _httpx_session() # -- lifecycle ---------------------------------------------------------- def close(self) -> None: if self._session is not None and hasattr(self._session, "close"): self._session.close() def __enter__(self): return self def __exit__(self, *_exc): self.close() # -- request ------------------------------------------------------------ def request(self, method: str, path: str, *, params: dict | None = None, json: Any = None, data: bytes | None = None, content_type: str | None = None, headers: dict | None = None, timeout_s: float | None = None) -> Response: url = self._url(path, params) body, request_headers = self._prepare(json, data, content_type, headers) idempotent = method.upper() in ("GET", "HEAD") timeout = self.timeout_s if timeout_s is None else timeout_s attempt = 0 while True: try: response = self._send(method, url, body, request_headers, timeout) except TransportError: # Unknown outcome. Safe to repeat only if the call has no effect. if idempotent and attempt < self.max_retries: self._sleep(attempt, None) attempt += 1 continue raise if response.status in RETRY_STATUSES and attempt < self.max_retries: self._sleep(attempt, _retry_after(response.headers)) attempt += 1 continue _raise_for_status(response) return response def _url(self, path: str, params: dict | None) -> str: url = f"{self.base_url}/{path.lstrip('/')}" if not params: return url # Drop unset filters rather than sending `?source=None`, and expand a # list into repeated keys, which is how ODMS reads multi-valued filters. pairs: list[tuple[str, str]] = [] for key, value in params.items(): if value is None: continue if isinstance(value, (list, tuple)): pairs.extend((key, _stringify(v)) for v in value if v is not None) else: pairs.append((key, _stringify(value))) return f"{url}?{urllib.parse.urlencode(pairs)}" if pairs else url def _prepare(self, json_body, data, content_type, extra_headers): headers = { "User-Agent": self.user_agent, "Accept": "application/json, text/plain, */*", } if self.token: headers["Authorization"] = f"Bearer {self.token}" body = data if json_body is not None: body = _json.dumps(json_body).encode("utf-8") headers["Content-Type"] = "application/json" if content_type: headers["Content-Type"] = content_type if extra_headers: headers.update(extra_headers) return body, headers def _send(self, method, url, body, headers, timeout) -> Response: if self._session is not None: return self._send_httpx(method, url, body, headers, timeout) return self._send_urllib(method, url, body, headers, timeout) def _send_httpx(self, method, url, body, headers, timeout) -> Response: import httpx try: reply = self._session.request( method, url, content=body, headers=headers, timeout=timeout, ) except httpx.HTTPError as exc: raise TransportError(f"{method} {url} failed: {exc}") from exc return Response(reply.status_code, dict(reply.headers), reply.content, url) def _send_urllib(self, method, url, body, headers, timeout) -> Response: request = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=timeout) as reply: return Response(reply.status, dict(reply.headers), reply.read(), url) except urllib.error.HTTPError as exc: # urllib raises on 4xx/5xx. That is a real response with a real body, # so hand it back and let the retry/raise logic classify it. return Response(exc.code, dict(exc.headers or {}), exc.read(), url) except (urllib.error.URLError, OSError) as exc: raise TransportError(f"{method} {url} failed: {exc}") from exc def _sleep(self, attempt: int, retry_after: float | None) -> None: if retry_after is not None: time.sleep(min(retry_after, MAX_RETRY_AFTER_S)) return # Jittered: a fleet that hits the limit together must not wake together. delay = self.backoff_s * (2 ** attempt) time.sleep(delay * (0.5 + random.random())) def _stringify(value) -> str: if isinstance(value, bool): # Django reads "True"/"False", not Python's lowercased str(bool). return "True" if value else "False" if hasattr(value, "isoformat"): return value.isoformat() return str(value)