252 lines
9.4 KiB
Python
252 lines
9.4 KiB
Python
"""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)
|