Python Snippets

RESTful API Client with Automatic Retry and Circuit Breaker Pattern

import requests
import time
import threading
from typing import Optional, Dict, Any
from enum import Enum
from dataclasses import dataclass

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    timeout: int = 60  # seconds
    half_open_timeout: int = 30  # seconds
    retry_attempts: int = 3
    retry_delay: float = 1.0  # seconds

class CircuitBreaker:
    def __init__(self, config: CircuitBreakerConfig = None):
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_start_time = None
        self._lock = threading.Lock()
    
    def can_execute(self) -> bool:
        with self._lock:
            if self.state == CircuitState.CLOSED:
                return True
            elif self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time >= self.config.timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_start_time = time.time()
                    return True
                return False
            elif self.state == CircuitState.HALF_OPEN:
                if time.time() - self.half_open_start_time >= self.config.half_open_timeout:
                    self.state = CircuitState.HALF_OPEN
                    return True
                return False
    
    def on_success(self):
        with self._lock:
            self.failure_count = 0
            self.state = CircuitState.CLOSED
    
    def on_failure(self):
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.config.failure_threshold:
                self.state = CircuitState.OPEN
    
    def is_half_open_test(self) -> bool:
        return self.state == CircuitState.HALF_OPEN

class APIClient:
    def __init__(self, base_url: str, circuit_breaker_config: CircuitBreakerConfig = None):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.circuit_breaker = CircuitBreaker(circuit_breaker_config)
    
    def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        for attempt in range(self.circuit_breaker.config.retry_attempts):
            if not self.circuit_breaker.can_execute():
                raise Exception("Circuit breaker is OPEN. Service unavailable.")
            
            try:
                response = self.session.request(method, url, **kwargs)
                response.raise_for_status()
                
                # If we're in HALF_OPEN state and this is a test request, 
                # only count success if it's the test request
                if self.circuit_breaker.is_half_open_test():
                    self.circuit_breaker.on_success()
                elif not self.circuit_breaker.is_half_open_test():
                    self.circuit_breaker.on_success()
                
                return response
                
            except requests.exceptions.RequestException as e:
                # If we're in HALF_OPEN state, any failure means we go back to OPEN
                if self.circuit_breaker.is_half_open_test():
                    self.circuit_breaker.on_failure()
                    raise Exception("Service test failed. Circuit breaker remains OPEN.")
                else:
                    self.circuit_breaker.on_failure()
                
                if attempt < self.circuit_breaker.config.retry_attempts - 1:
                    time.sleep(self.circuit_breaker.config.retry_delay * (2 ** attempt))  # Exponential backoff
                else:
                    raise e
    
    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict[str, Any]:
        response = self._make_request("GET", endpoint, params=params)
        return response.json()
    
    def post(self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None) -> Dict[str, Any]:
        response = self._make_request("POST", endpoint, data=data, json=json)
        return response.json()
    
    def put(self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None) -> Dict[str, Any]:
        response = self._make_request("PUT", endpoint, data=data, json=json)
        return response.json()
    
    def delete(self, endpoint: str) -> bool:
        response = self._make_request("DELETE", endpoint)
        return response.status_code == 204

# Example usage
if __name__ == "__main__":
    # Configure the circuit breaker
    config = CircuitBreakerConfig(
        failure_threshold=3,
        timeout=30,
        retry_attempts=2,
        retry_delay=0.5
    )
    
    # Create API client
    client = APIClient("https://jsonplaceholder.typicode.com", config)
    
    try:
        # Make a successful GET request
        posts = client.get("/posts", params={"userId": 1})
        print(f"Retrieved {len(posts)} posts")
        
        # Create a new post
        new_post = client.post("/posts", json={
            "title": "Test Post",
            "body": "This is a test post",
            "userId": 1
        })
        print(f"Created post with ID: {new_post.get('id')}")
        
    except Exception as e:
        print(f"API request failed: {e}")

What This Code Does

This snippet implements a robust RESTful API client that incorporates two important resilience patterns:

Automatic Retry with Exponential Backoff: When a request fails, the client automatically retries it up to a configurable number of times, with increasing delays between attempts.
Circuit Breaker Pattern: Prevents the client from continuously making requests to a failing service. After a threshold of failures, the circuit “opens” and blocks further requests for a timeout period, allowing the service to recover.

Key Features

Configurable Circuit Breaker: Adjust failure thresholds, timeouts, and retry parameters
Thread-Safe Implementation: Uses locks to ensure safe concurrent access
Multiple HTTP Methods: Support for GET, POST, PUT, and DELETE operations
Automatic JSON Handling: Automatically parses JSON responses
Exponential Backoff: Increases delay between retry attempts to avoid overwhelming services
Half-Open State: After timeout, allows one test request to check if service has recovered

Why This is Useful

Modern applications often depend on multiple external services. Network issues, service outages, and rate limiting can cause requests to fail temporarily. This client helps your application:

Handle transient failures gracefully instead of crashing
Prevent cascade failures by stopping requests to consistently failing services
Reduce load on struggling services by backing off during outages
Maintain better user experience through intelligent retry logic

How to Run

Install the required dependency:
```
pip install requests
```
Run the script directly:
```
python api_client.py
```
To use in your own code: ```python from api_client import APIClient, CircuitBreakerConfig

config = CircuitBreakerConfig(failure_threshold=5, timeout=60) client = APIClient(“https://your-api.com”, config)

Make requests

data = client.get(“/endpoint”) result = client.post(“/endpoint”, json={“key”: “value”}) ```

The client will automatically handle retries and circuit breaking logic, making your API interactions more resilient to temporary failures.