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")

Tags

api requests http-client rest

Related Snippets

javascript

Express Global Error Handler

python

Web Scraper with BeautifulSoup

python

CSV Data Processor

python

Email Sender with Templates