import json
import logging
import os
import time
from collections import deque
from dataclasses import dataclass
from typing import (
Any,
Callable,
Deque,
Dict,
Iterable,
Iterator,
Mapping,
MutableMapping,
Optional,
Tuple,
Union,
)
from urllib.parse import urljoin
import requests
from requests import Response, Session
from requests.exceptions import ConnectionError, Timeout
@dataclass(frozen=True)
class UnifiedResponse:
"""Unified HTTP response model.
Attributes:
status_code: HTTP status code.
url: Final URL of the response.
method: HTTP method used for the request.
headers: Response headers.
data: Parsed JSON object if available; otherwise, None.
text: Text body if available and not JSON; otherwise, None.
content: Raw bytes content if streaming or non-text; otherwise, None.
elapsed: Elapsed time in seconds for the request.
request_id: Request ID extracted from headers if present.
"""
status_code: int
url: str
method: str
headers: Mapping[str, str]
data: Optional[Any]
text: Optional[str]
content: Optional[bytes]
elapsed: float
request_id: Optional[str]
class RestClientError(Exception):
"""Base exception for REST client errors."""
def __init__(
self,
message: str,
*,
status_code: Optional[int] = None,
response: Optional[UnifiedResponse] = None,
) -> None:
super().__init__(message)
self.status_code = status_code
self.response = response
class BadRequestError(RestClientError):
"""400 Bad Request."""
class UnauthorizedError(RestClientError):
"""401 Unauthorized."""
class ForbiddenError(RestClientError):
"""403 Forbidden."""
class NotFoundError(RestClientError):
"""404 Not Found."""
class ConflictError(RestClientError):
"""409 Conflict."""
class UnprocessableEntityError(RestClientError):
"""422 Unprocessable Entity."""
class TooManyRequestsError(RestClientError):
"""429 Too Many Requests."""
class ServerError(RestClientError):
"""5xx Server Error."""
class HTTPError(RestClientError):
"""Generic HTTP error."""
class RequestSendError(RestClientError):
"""Network or transport-level error when sending a request."""
class RestClient:
"""A generic REST client with retries, rate limiting, and structured logging.
This client provides:
- Session management with a base URL and default timeout.
- Bearer token authentication (static token or provider callable).
- Exponential backoff retries for network and retryable HTTP errors.
- Optional rate limiting (sliding window).
- JSON serialization and unified response parsing.
- Pagination helpers.
- Structured logging with contextual fields.
- File upload and streaming download helpers.
Args:
base_url: Base URL for all requests. Paths are joined via urljoin.
token: Static bearer token for Authorization header. Ignored if
token_provider is provided.
token_provider: Callable returning a bearer token string. Called per
request to populate Authorization header.
timeout: Default timeout in seconds for requests.
max_retries: Maximum number of retry attempts on retryable errors.
Total attempts will be max_retries + 1 including the initial try.
backoff_factor: Base factor for exponential backoff (seconds).
backoff_max: Maximum backoff sleep duration per attempt (seconds).
retry_on_status: Iterable of HTTP status codes to retry on. Defaults to
{429} and 5xx.
rate_limit: Optional rate limiting tuple (max_requests, per_seconds).
Example: (10, 1.0) allows 10 requests per 1 second.
session: Optional pre-configured requests.Session.
default_headers: Default headers added to every request.
logger: Optional logger; defaults to `logging.getLogger(__name__)`.
Attributes:
base_url: Base URL joined with request paths.
timeout: Default timeout in seconds.
max_retries: Configured maximum retry attempts.
backoff_factor: Backoff factor for exponential backoff.
backoff_max: Cap for backoff sleep per attempt.
retry_on_status: Set of status codes that will be retried.
rate_limit: Configured rate limit tuple or None.
session: Underlying requests.Session.
logger: Logger used for structured logs.
"""
def __init__(
self,
base_url: str,
*,
token: Optional[str] = None,
token_provider: Optional[Callable[[], str]] = None,
timeout: float = 30.0,
max_retries: int = 3,
backoff_factor: float = 0.5,
backoff_max: float = 60.0,
retry_on_status: Optional[Iterable[int]] = None,
rate_limit: Optional[Tuple[int, float]] = None,
session: Optional[Session] = None,
default_headers: Optional[Mapping[str, str]] = None,
logger: Optional[logging.Logger] = None,
) -> None:
self.base_url = base_url.rstrip("/") + "/"
self._token = token
self._token_provider = token_provider
self.timeout = timeout
self.max_retries = max(0, int(max_retries))
self.backoff_factor = max(0.0, float(backoff_factor))
self.backoff_max = max(0.0, float(backoff_max))
default_retry = {429}
self.retry_on_status = set(retry_on_status or default_retry)
self.rate_limit = rate_limit
self._rate_window: Deque[float] = deque(maxlen=rate_limit[0]) if rate_limit else deque()
self.session = session or requests.Session()
self.default_headers: Dict[str, str] = dict(default_headers or {})
self.logger = logger or logging.getLogger(__name__)
# Default headers
if "Accept" not in self.default_headers:
self.default_headers["Accept"] = "application/json"
if "User-Agent" not in self.default_headers:
self.default_headers["User-Agent"] = "RestClient/1.0"
# Common JSON header for write methods; GET should not force Content-Type
# but we add per-request if json payload is provided.
# -------------- Public API --------------
def set_token(self, token: Optional[str]) -> None:
"""Set or update the static bearer token.
Args:
token: Bearer token string or None to clear.
"""
self._token = token
def set_token_provider(
self, provider: Optional[Callable[[], str]]
) -> None:
"""Set or update the token provider callable.
Args:
provider: Callable returning a bearer token string, or None to clear.
"""
self._token_provider = provider
def request(
self,
method: str,
path: str,
*,
params: Optional[Mapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
json_body: Optional[Any] = None,
data: Optional[Union[Mapping[str, Any], bytes]] = None,
files: Optional[Mapping[str, Any]] = None,
timeout: Optional[float] = None,
stream: bool = False,
allow_redirects: bool = True,
expected_status: Optional[Iterable[int]] = None,
) -> UnifiedResponse:
"""Perform an HTTP request.
Args:
method: HTTP method (GET, POST, PUT, DELETE, etc.).
path: Relative or absolute URL path.
params: Query string parameters.
headers: Additional request headers.
json_body: JSON-serializable payload set as request body.
data: Raw or form data payload.
files: Files for multipart upload (requests-compatible format).
timeout: Per-request timeout override in seconds.
stream: Whether to stream the response.
allow_redirects: Whether to follow redirects.
expected_status: Optional iterable of acceptable status codes.
If provided, responses not in this set raise an error.
Returns:
UnifiedResponse: Parsed response object.
Raises:
RestClientError: On HTTP or transport errors.
"""
return self._request(
method=method,
path=path,
params=params,
headers=headers,
json_body=json_body,
data=data,
files=files,
timeout=timeout,
stream=stream,
allow_redirects=allow_redirects,
expected_status=set(expected_status) if expected_status else None,
)
def get(
self,
path: str,
*,
params: Optional[Mapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None,
stream: bool = False,
expected_status: Optional[Iterable[int]] = None,
) -> UnifiedResponse:
"""HTTP GET request.
Args:
path: URL path.
params: Query parameters.
headers: Additional headers.
timeout: Timeout override in seconds.
stream: Stream response.
expected_status: Acceptable status codes.
Returns:
UnifiedResponse: Parsed response.
"""
return self.request(
"GET",
path,
params=params,
headers=headers,
timeout=timeout,
stream=stream,
expected_status=expected_status,
)
def post(
self,
path: str,
*,
params: Optional[Mapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
json_body: Optional[Any] = None,
data: Optional[Union[Mapping[str, Any], bytes]] = None,
files: Optional[Mapping[str, Any]] = None,
timeout: Optional[float] = None,
expected_status: Optional[Iterable[int]] = None,
) -> UnifiedResponse:
"""HTTP POST request.
Args:
path: URL path.
params: Query parameters.
headers: Additional headers.
json_body: JSON payload.
data: Raw or form data payload.
files: Files for multipart upload.
timeout: Timeout override in seconds.
expected_status: Acceptable status codes.
Returns:
UnifiedResponse: Parsed response.
"""
return self.request(
"POST",
path,
params=params,
headers=headers,
json_body=json_body,
data=data,
files=files,
timeout=timeout,
expected_status=expected_status,
)
def put(
self,
path: str,
*,
params: Optional[Mapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
json_body: Optional[Any] = None,
data: Optional[Union[Mapping[str, Any], bytes]] = None,
timeout: Optional[float] = None,
expected_status: Optional[Iterable[int]] = None,
) -> UnifiedResponse:
"""HTTP PUT request.
Args:
path: URL path.
params: Query parameters.
headers: Additional headers.
json_body: JSON payload.
data: Raw or form data payload.
timeout: Timeout override in seconds.
expected_status: Acceptable status codes.
Returns:
UnifiedResponse: Parsed response.
"""
return self.request(
"PUT",
path,
params=params,
headers=headers,
json_body=json_body,
data=data,
timeout=timeout,
expected_status=expected_status,
)
def delete(
self,
path: str,
*,
params: Optional[Mapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None,
expected_status: Optional[Iterable[int]] = None,
) -> UnifiedResponse:
"""HTTP DELETE request.
Args:
path: URL path.
params: Query parameters.
headers: Additional headers.
timeout: Timeout override in seconds.
expected_status: Acceptable status codes.
Returns:
UnifiedResponse: Parsed response.
"""
return self.request(
"DELETE",
path,
params=params,
headers=headers,
timeout=timeout,
expected_status=expected_status,
)
def upload(
self,
path: str,
*,
files: Mapping[str, Any],
data: Optional[Mapping[str, Any]] = None,
params: Optional[Mapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None,
expected_status: Optional[Iterable[int]] = None,
) -> UnifiedResponse:
"""Upload files using multipart/form-data.
Args:
path: URL path.
files: Files mapping for multipart upload, compatible with requests.
data: Optional form fields.
params: Query parameters.
headers: Additional headers.
timeout: Timeout override in seconds.
expected_status: Acceptable status codes.
Returns:
UnifiedResponse: Parsed response.
"""
return self.post(
path,
params=params,
headers=headers,
data=data,
files=files,
timeout=timeout,
expected_status=expected_status,
)
def download(
self,
path: str,
*,
dest_path: str,
params: Optional[Mapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None,
chunk_size: int = 8192,
overwrite: bool = False,
expected_status: Optional[Iterable[int]] = None,
) -> str:
"""Download a resource and save to a local file.
Args:
path: URL path.
dest_path: Local file path to save.
params: Query parameters.
headers: Additional headers.
timeout: Timeout override in seconds.
chunk_size: Stream buffer size in bytes.
overwrite: Whether to overwrite an existing file.
expected_status: Acceptable status codes.
Returns:
str: Absolute path to the saved file.
Raises:
RestClientError: On HTTP or IO errors.
"""
if os.path.exists(dest_path) and not overwrite:
raise RestClientError(
f"File exists and overwrite=False: {dest_path}"
)
resp = self.get(
path,
params=params,
headers=headers,
timeout=timeout,
stream=True,
expected_status=expected_status,
)
if resp.status_code >= 400:
self._raise_http_error(resp)
try:
with open(dest_path, "wb") as f, self._stream_response(
path, params, headers, timeout, chunk_size
) as stream_resp:
for chunk in stream_resp.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
except OSError as exc:
raise RestClientError(
f"I/O error while saving to {dest_path}: {exc}"
) from exc
return os.path.abspath(dest_path)
def paginate(
self,
path: str,
*,
method: str = "GET",
params: Optional[MutableMapping[str, Any]] = None,
headers: Optional[Mapping[str, str]] = None,
items_key: Optional[str] = None,
page_param: str = "page",
per_page_param: str = "per_page",
start_page: int = 1,
per_page: int = 100,
next_key: Optional[str] = None,
max_pages: Optional[int] = None,
timeout: Optional[float] = None,
) -> Iterator[Union[UnifiedResponse, Any]]:
"""Paginate through resources.
Supports two common patterns:
1) Page-number based: increments `page_param` until empty or max_pages.
2) Cursor/next-link based: uses `next_key` found in JSON response
(e.g., {"next": "..."}). When provided, this takes precedence.
Args:
path: URL path for the initial request.
method: HTTP method to use, typically GET.
params: Initial query parameters (mutated during pagination).
headers: Additional request headers.
items_key: If provided, yields items from response.data[items_key];
otherwise yields the UnifiedResponse for each page.
page_param: Query string key for page number.
per_page_param: Query string key for page size.
start_page: Starting page number.
per_page: Page size.
next_key: JSON key holding the next cursor or URL; if present, the
paginator will follow it until None/empty.
max_pages: Maximum number of pages to fetch.
timeout: Timeout override per page.
Yields:
Union[UnifiedResponse, Any]: Pages or items, depending on items_key.
Raises:
RestClientError: On HTTP errors.
"""
current_params: MutableMapping[str, Any] = dict(params or {})
if next_key:
# Cursor/next mode
next_value: Optional[str] = None
page_count = 0
while True:
if next_value:
# Prefer cursor in params; if it's a full URL, replace path.
if next_value.startswith("http"):
current_path = next_value
else:
current_params[next_key] = next_value
current_path = path
else:
current_path = path
resp = self.request(
method,
current_path,
params=current_params,
headers=headers,
timeout=timeout,
)
if items_key and isinstance(resp.data, dict):
items = resp.data.get(items_key, []) or []
for item in items:
yield item
else:
yield resp
next_value = None
if isinstance(resp.data, dict):
nval = resp.data.get(next_key) # type: ignore[index]
if isinstance(nval, str) and nval:
next_value = nval
page_count += 1
if not next_value:
break
if max_pages is not None and page_count >= max_pages:
break
else:
# Page-number mode
page = start_page
page_count = 0
current_params[page_param] = page
current_params[per_page_param] = per_page
while True:
resp = self.request(
method,
path,
params=current_params,
headers=headers,
timeout=timeout,
)
count = 0
if items_key and isinstance(resp.data, dict):
items = resp.data.get(items_key, []) or []
for item in items:
count += 1
yield item
else:
yield resp
# If no items_key, we assume presence of content implies
# there may be more pages; this is heuristic.
if isinstance(resp.data, list):
count = len(resp.data)
elif isinstance(resp.data, dict):
# Best-effort heuristic
count = len(resp.data)
page_count += 1
if max_pages is not None and page_count >= max_pages:
break
if count == 0:
break
page += 1
current_params[page_param] = page
def close(self) -> None:
"""Close the underlying session."""
self.session.close()
def __enter__(self) -> "RestClient":
"""Enter context manager."""
return self
def __exit__(self, exc_type, exc, tb) -> None:
"""Exit context manager and close session."""
self.close()
# -------------- Internal Helpers --------------
def _request(
self,
*,
method: str,
path: str,
params: Optional[Mapping[str, Any]],
headers: Optional[Mapping[str, str]],
json_body: Optional[Any],
data: Optional[Union[Mapping[str, Any], bytes]],
files: Optional[Mapping[str, Any]],
timeout: Optional[float],
stream: bool,
allow_redirects: bool,
expected_status: Optional[Iterable[int]],
) -> UnifiedResponse:
url = self._build_url(path)
req_headers = self._merge_headers(headers, json_body=json_body)
req_timeout = timeout if timeout is not None else self.timeout
attempts = self.max_retries + 1
for attempt in range(1, attempts + 1):
self._apply_rate_limit()
start = time.perf_counter()
error: Optional[Exception] = None
try:
resp = self.session.request(
method=method.upper(),
url=url,
params=params,
headers=req_headers,
json=json_body,
data=data,
files=files,
timeout=req_timeout,
stream=stream,
allow_redirects=allow_redirects,
)
elapsed = time.perf_counter() - start
uresp = self._to_unified_response(resp, method)
self._log_request(
method=method,
url=resp.url,
status_code=resp.status_code,
elapsed=elapsed,
attempt=attempt,
params=params,
stream=stream,
)
if expected_status and uresp.status_code not in expected_status:
if self._should_retry_status(uresp.status_code):
self._sleep_backoff(attempt)
continue
self._raise_http_error(uresp)
if 200 <= uresp.status_code < 300:
return uresp
if self._should_retry_status(uresp.status_code):
self._sleep_backoff(attempt)
continue
self._raise_http_error(uresp)
except (Timeout, ConnectionError) as exc:
elapsed = time.perf_counter() - start
error = exc
self._log_request(
method=method,
url=url,
status_code=None,
elapsed=elapsed,
attempt=attempt,
params=params,
stream=stream,
error=repr(exc),
)
if attempt < attempts:
self._sleep_backoff(attempt)
continue
raise RequestSendError(
f"Request failed after {attempts} attempts: {exc}"
) from exc
except requests.RequestException as exc:
elapsed = time.perf_counter() - start
error = exc
self._log_request(
method=method,
url=url,
status_code=None,
elapsed=elapsed,
attempt=attempt,
params=params,
stream=stream,
error=repr(exc),
)
raise RequestSendError(f"Request error: {exc}") from exc
raise RequestSendError("Exhausted retries without a response")
def _stream_response(
self,
path: str,
params: Optional[Mapping[str, Any]],
headers: Optional[Mapping[str, str]],
timeout: Optional[float],
chunk_size: int,
) -> Response:
"""Internal helper to open a streaming response."""
url = self._build_url(path)
req_headers = self._merge_headers(headers)
req_timeout = timeout if timeout is not None else self.timeout
attempts = self.max_retries + 1
for attempt in range(1, attempts + 1):
self._apply_rate_limit()
start = time.perf_counter()
try:
resp = self.session.get(
url,
params=params,
headers=req_headers,
timeout=req_timeout,
stream=True,
)
elapsed = time.perf_counter() - start
self._log_request(
method="GET",
url=resp.url,
status_code=resp.status_code,
elapsed=elapsed,
attempt=attempt,
params=params,
stream=True,
)
if self._should_retry_status(resp.status_code):
self._sleep_backoff(attempt)
continue
resp.raise_for_status()
return resp
except (Timeout, ConnectionError) as exc:
elapsed = time.perf_counter() - start
self._log_request(
method="GET",
url=url,
status_code=None,
elapsed=elapsed,
attempt=attempt,
params=params,
stream=True,
error=repr(exc),
)
if attempt < attempts:
self._sleep_backoff(attempt)
continue
raise RequestSendError(
f"Streaming request failed after {attempts} attempts: {exc}"
) from exc
raise RequestSendError("Exhausted retries for streaming request")
def _build_url(self, path: str) -> str:
if path.startswith("http://") or path.startswith("https://"):
return path
return urljoin(self.base_url, path.lstrip("/"))
def _merge_headers(
self, headers: Optional[Mapping[str, str]], *, json_body: Optional[Any] = None
) -> Dict[str, str]:
merged = dict(self.default_headers)
if headers:
merged.update(headers)
token = self._get_bearer_token()
if token:
merged["Authorization"] = f"Bearer {token}"
if json_body is not None and "Content-Type" not in merged:
merged["Content-Type"] = "application/json"
return merged
def _get_bearer_token(self) -> Optional[str]:
if self._token_provider:
try:
token = self._token_provider()
if token:
return token
except Exception: # noqa: BLE001
# Provider failures should not crash silently; log and fallback
self.logger.exception("Token provider callable raised")
return self._token
def _should_retry_status(self, status_code: int) -> bool:
if status_code in self.retry_on_status:
return True
if 500 <= status_code < 600:
return True
return False
def _sleep_backoff(self, attempt: int) -> None:
if attempt > self.max_retries:
return
# Exponential backoff with jitter (full jitter)
base = self.backoff_factor * (2 ** (attempt - 1))
sleep_for = min(self.backoff_max, base) * (0.5 + 0.5 * (time.time() % 1))
time.sleep(max(0.0, sleep_for))
def _apply_rate_limit(self) -> None:
if not self.rate_limit:
return
max_req, per_seconds = self.rate_limit
now = time.monotonic()
# Purge timestamps outside the window
while self._rate_window and (now - self._rate_window[0]) >= per_seconds:
self._rate_window.popleft()
if len(self._rate_window) >= max_req:
# Sleep until the oldest timestamp exits the window
wait_for = per_seconds - (now - self._rate_window[0])
if wait_for > 0:
time.sleep(wait_for)
# After sleep, purge again
now = time.monotonic()
while self._rate_window and (now - self._rate_window[0]) >= per_seconds:
self._rate_window.popleft()
self._rate_window.append(time.monotonic())
def _to_unified_response(self, resp: Response, method: str) -> UnifiedResponse:
request_id = (
resp.headers.get("X-Request-Id")
or resp.headers.get("X-Request-ID")
or resp.headers.get("Request-Id")
)
data: Optional[Any] = None
text: Optional[str] = None
content: Optional[bytes] = None
content_type = resp.headers.get("Content-Type", "")
if "application/json" in content_type.lower():
try:
data = resp.json()
except ValueError:
text = resp.text
else:
# For streamed responses, content may be large; leave None here.
if resp.request and resp.request.headers.get("Accept") == "application/json":
# If server ignored Accept and returned non-JSON, capture text
text = resp.text
else:
# Non-JSON: capture bytes for convenience (non-streamed)
if not resp.raw.closed and not resp.request.kwargs.get( # type: ignore[attr-defined]
"stream", False
):
content = resp.content
else:
text = resp.text
return UnifiedResponse(
status_code=resp.status_code,
url=str(resp.url),
method=method.upper(),
headers=dict(resp.headers),
data=data,
text=text,
content=content,
elapsed=resp.elapsed.total_seconds() if resp.elapsed else 0.0,
request_id=request_id,
)
def _raise_http_error(self, resp: UnifiedResponse) -> None:
status = resp.status_code
message = self._build_error_message(resp)
if status == 400:
raise BadRequestError(message, status_code=status, response=resp)
if status == 401:
raise UnauthorizedError(message, status_code=status, response=resp)
if status == 403:
raise ForbiddenError(message, status_code=status, response=resp)
if status == 404:
raise NotFoundError(message, status_code=status, response=resp)
if status == 409:
raise ConflictError(message, status_code=status, response=resp)
if status == 422:
raise UnprocessableEntityError(
message, status_code=status, response=resp
)
if status == 429:
raise TooManyRequestsError(message, status_code=status, response=resp)
if 500 <= status < 600:
raise ServerError(message, status_code=status, response=resp)
raise HTTPError(message, status_code=status, response=resp)
def _build_error_message(self, resp: UnifiedResponse) -> str:
# Prefer server-provided message from JSON body
if isinstance(resp.data, dict):
msg = (
resp.data.get("message")
or resp.data.get("error")
or resp.data.get("detail")
)
if isinstance(msg, str):
return f"{resp.status_code} {msg}"
# Fallback to text snippet
if resp.text:
snippet = resp.text.strip()
if len(snippet) > 300:
snippet = snippet[:300] + "..."
return f"{resp.status_code} {snippet}"
return f"HTTP {resp.status_code} at {resp.url}"
def _log_request(
self,
*,
method: str,
url: str,
status_code: Optional[int],
elapsed: float,
attempt: int,
params: Optional[Mapping[str, Any]],
stream: bool,
error: Optional[str] = None,
) -> None:
extra = {
"event": "http_request",
"method": method.upper(),
"url": url,
"status_code": status_code,
"elapsed_ms": round(elapsed * 1000, 2),
"attempt": attempt,
"retry": attempt - 1,
"stream": stream,
"error": error,
"params": dict(params or {}),
}
if error:
self.logger.warning("HTTP request error", extra=extra)
elif status_code and status_code >= 400:
self.logger.error("HTTP request failed", extra=extra)
else:
self.logger.info("HTTP request succeeded", extra=extra)
