python
API
REST API Client with Retry
Reusable API client class with session management, automatic retries, token refresh, and response caching.
Apex Logic
0 copies
python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time
from typing import Optional, Dict, Any
class APIClient:
def __init__(self, base_url: str, api_key: Optional[str] = None, timeout: int = 30):
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.session = requests.Session()
# Configure retry strategy
retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503])
adapter = HTTPAdapter(max_retries=retry)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
if api_key:
self.session.headers.update({"Authorization": f"Bearer {api_key}"})
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json",
})
def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
response = self.session.request(method, url, timeout=self.timeout, **kwargs)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 5))
time.sleep(retry_after)
response = self.session.request(method, url, timeout=self.timeout, **kwargs)
response.raise_for_status()
return response.json() if response.content else {}
def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
return self._request("GET", endpoint, params=params)
def post(self, endpoint: str, data: Optional[Dict] = None) -> Dict:
return self._request("POST", endpoint, json=data)
def put(self, endpoint: str, data: Optional[Dict] = None) -> Dict:
return self._request("PUT", endpoint, json=data)
def delete(self, endpoint: str) -> Dict:
return self._request("DELETE", endpoint)
if __name__ == "__main__":
client = APIClient("https://api.example.com", api_key="your-key")
users = client.get("/users", params={"page": 1, "limit": 10})
print(f"Found {len(users.get('data', []))} users")