"""HTTP API client with retry, rate limiting, and structured output."""
import os
import time
import base64
import json
import logging
from dataclasses import dataclass, field
from typing import Optional, Any
from urllib.parse import urljoin, urlencode
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
logger = logging.getLogger(__name__)
@dataclass
class APIResponse:
"""Structured API response."""
success: bool
status_code: int
headers: dict
body: Any
duration_ms: int
retries: int
error: Optional[str] = None
@dataclass
class APIClientConfig:
"""Configuration for the API client."""
base_url: str
auth_method: str = "none"
timeout: int = 30
max_retries: int = 3
backoff_factor: float = 0.5
rate_limit_buffer: float = 0.1
max_pages: int = 100
class APIClient:
"""HTTP API client with safety controls."""
def __init__(self, config: APIClientConfig):
self.config = config
self.session = self._build_session()
self._apply_auth()
self._last_request_time = 0.0
self._rate_limit_remaining = None
self._rate_limit_reset = None
def _build_session(self) -> requests.Session:
"""Create a session with retry configuration."""
session = requests.Session()
retry = Retry(
total=self.config.max_retries,
backoff_factor=self.config.backoff_factor,
status_forcelist=[500, 502, 503, 504],
allowed_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
def _apply_auth(self) -> None:
"""Apply authentication to the session."""
method = self.config.auth_method
if method == "bearer":
token = os.environ.get("API_BEARER_TOKEN", "")
self.session.headers["Authorization"] = f"Bearer {token}"
elif method == "api_key_header":
key = os.environ.get("API_KEY", "")
self.session.headers["X-API-Key"] = key
elif method == "basic":
user = os.environ.get("API_USER", "")
passwd = os.environ.get("API_PASS", "")
encoded = base64.b64encode(f"{user}:{passwd}".encode()).decode()
self.session.headers["Authorization"] = f"Basic {encoded}"
elif method == "oauth2":
self._refresh_oauth_token()
def _refresh_oauth_token(self) -> None:
"""Obtain OAuth2 token using client credentials."""
token_url = os.environ.get("OAUTH_TOKEN_URL", "")
client_id = os.environ.get("OAUTH_CLIENT_ID", "")
client_secret = os.environ.get("OAUTH_CLIENT_SECRET", "")
resp = requests.post(
token_url,
data={"grant_type": "client_credentials"},
auth=(client_id, client_secret),
timeout=self.config.timeout,
)
resp.raise_for_status()
token = resp.json()["access_token"]
self.session.headers["Authorization"] = f"Bearer {token}"
def _respect_rate_limit(self, response: requests.Response) -> None:
"""Track and respect rate limit headers."""
remaining = response.headers.get("X-RateLimit-Remaining")
reset = response.headers.get("X-RateLimit-Reset")
if remaining is not None:
self._rate_limit_remaining = int(remaining)
if reset is not None:
self._rate_limit_reset = float(reset)
if self._rate_limit_remaining is not None and self._rate_limit_remaining <= 1:
if self._rate_limit_reset:
wait = max(0, self._rate_limit_reset - time.time())
logger.info(f"Rate limit approaching, waiting {wait:.1f}s")
time.sleep(wait)
def request(
self,
method: str,
path: str,
params: Optional[dict] = None,
json_body: Optional[dict] = None,
headers: Optional[dict] = None,
) -> APIResponse:
"""
Make an API request.
Args:
method: HTTP method (GET, POST, PUT, PATCH, DELETE).
path: URL path (appended to base_url).
params: Query parameters.
json_body: JSON request body.
headers: Additional headers.
Returns:
Structured APIResponse.
"""
url = urljoin(self.config.base_url, path)
start = time.monotonic()
retries = 0
try:
resp = self.session.request(
method=method.upper(),
url=url,
params=params,
json=json_body,
headers=headers,
timeout=self.config.timeout,
)
self._respect_rate_limit(resp)
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 5))
logger.warning(f"Rate limited, waiting {retry_after}s")
time.sleep(retry_after)
return self.request(method, path, params, json_body, headers)
body = None
content_type = resp.headers.get("Content-Type", "")
if "application/json" in content_type:
body = resp.json()
elif "text/" in content_type:
body = resp.text
else:
body = resp.content.decode("utf-8", errors="replace")
duration_ms = int((time.monotonic() - start) * 1000)
return APIResponse(
success=resp.ok,
status_code=resp.status_code,
headers=dict(resp.headers),
body=body,
duration_ms=duration_ms,
retries=retries,
)
except requests.exceptions.Timeout:
duration_ms = int((time.monotonic() - start) * 1000)
return APIResponse(
success=False, status_code=0, headers={},
body=None, duration_ms=duration_ms, retries=retries,
error="Request timed out",
)
except requests.exceptions.ConnectionError as e:
duration_ms = int((time.monotonic() - start) * 1000)
return APIResponse(
success=False, status_code=0, headers={},
body=None, duration_ms=duration_ms, retries=retries,
error=f"Connection failed: {e}",
)
def get(self, path: str, **kwargs) -> APIResponse:
return self.request("GET", path, **kwargs)
def post(self, path: str, **kwargs) -> APIResponse:
return self.request("POST", path, **kwargs)
def put(self, path: str, **kwargs) -> APIResponse:
return self.request("PUT", path, **kwargs)
def delete(self, path: str, **kwargs) -> APIResponse:
return self.request("DELETE", path, **kwargs)
def paginate(
self,
path: str,
params: Optional[dict] = None,
page_param: str = "page",
per_page_param: str = "per_page",
per_page: int = 100,
results_key: Optional[str] = None,
) -> list:
"""
Fetch all pages of a paginated endpoint.
Args:
path: API endpoint path.
params: Base query parameters.
page_param: Name of the page parameter.
per_page_param: Name of the per-page parameter.
per_page: Items per page.
results_key: JSON key containing the results array.
Returns:
Aggregated list of all results.
"""
all_results = []
page = 1
params = params or {}
while page <= self.config.max_pages:
params[page_param] = page
params[per_page_param] = per_page
resp = self.get(path, params=params)
if not resp.success:
break
results = resp.body
if results_key and isinstance(results, dict):
results = results.get(results_key, [])
if not results:
break
all_results.extend(results)
page += 1
return all_results
