Reliability Patterns: Circuit Breakers, Retries, Fallbacks, and Graceful Degradation

import httpx

# Set reasonable timeouts
response = httpx.get(
    "https://api.example.com/data",
    timeout=httpx.Timeout(5.0, connect=1.0)
)

# Don't do this - no timeout means infinite wait!
# response = requests.get(url)  # Bad!

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except RetryableError as e:
            if attempt == max_retries - 1:
                raise
            
            # Exponential backoff with jitter
            delay = base_delay * (2 ** attempt)
            jitter = random.uniform(0, delay / 2)
            time.sleep(delay + jitter)

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.state = "closed"
        self.last_failure_time = None
    
    def call(self, func):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half_open"
            else:
                raise CircuitOpenError("Circuit is open")
        
        try:
            result = func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        self.failure_count = 0
        self.state = "closed"
    
    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = "open"

def get_product_recommendations(user_id):
    try:
        # Try primary service
        recommendations = ml_service.get_recommendations(user_id)
        return recommendations
    except MLServiceError:
        # Fallback 1: Cached recommendations
        cached = cache.get(f"recs:{user_id}")
        if cached:
            return cached
        
        # Fallback 2: Popular products
        return popular_products()

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health():
    checks = {
        "database": check_database(),
        "cache": check_cache(),
        "external_api": check_external_api(),
    }
    
    all_healthy = all(checks.values())
    
    return {
        "status": "healthy" if all_healthy else "unhealthy",
        "checks": checks
    }, 200 if all_healthy else 503