Reliability & rate limiting
Retries, exponential backoff, circuit breakers, token buckets, and idempotent handlers.
How to Cap Retry Attempts in Python with a Decorator
Build a reusable retry decorator that caps attempts, adds delays, and lets flaky services fail fast instead of hanging.
import random
from functools import wraps
from time import sleep
def retry(max_attempts, delay=0.1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kw…How to Implement Graceful Degradation with Feature Disabling in Python
A pattern that disables enhanced features and falls back to basic functionality when a dependency fails, with mock-based testing.
import random
from unittest.mock import patch
class EnhancedFeature:
"""A feature that can gracefully degrade when a dependency is unavailable."""
def __init__(self):
self.feature_enabled = True
def get_enhanced_data(self):
"""Simulate an enhanced feature that depends on external data."…How to Implement Hedged Requests in Python
This code demonstrates a hedged request pattern using threading, which sends duplicate calls and returns the first result that arrives within a timeout.
import time
from unittest.mock import Mock
def hedged_request(call, timeout=0.05):
"""Execute two duplicate calls, return first result within timeout."""
result_container = {}
def run_and_store():
result_container['result'] = call()
result_container['done'] = True
# Simulate slow cal…How to Implement a Bulkhead Pattern with Threading in Python
Implement a bulkhead pattern in Python that isolates concurrent tasks with a bounded semaphore, limiting active workers to prevent resource exhaustion.
import threading
import time
import random
class Bulkhead:
def __init__(self, workers: int):
self._semaphore = threading.BoundedSemaphore(workers)
self._lock = threading.Lock()
self._active = 0
def run(self, task):
with self._semaphore:
with self._lock:
…How to Implement a Circuit Breaker in Python
A Python dataclass that provides circuit breaker logic with closed, open, and half-open states to fail fast on repeated errors.
from dataclasses import dataclass
from datetime import datetime, timedelta
import time
@dataclass
class CircuitBreaker:
failure_threshold: int = 3
timeout_seconds: float = 5.0
failures: int = 0
state: str = "closed"
last_failure: datetime = None
def call(self, func):
if self.state ==…How to Implement a Sliding Window Log Rate Limiter in Python
Implements a sliding window log rate limiter in Python using a deque of timestamps to enforce a maximum request count within a rolling time window.
from collections import deque
from datetime import datetime, timedelta
from time import sleep
class SlidingWindowLog:
def __init__(self, window_seconds: int, max_requests: int):
self.window_seconds = window_seconds
self.max_requests = max_requests
self.timestamps = deque()
def allow_…How to Implement a Token Bucket Rate Limiter per Client IP in Python
Implements a simple sliding-window rate limiter using a dictionary of timestamp lists per client IP to limit requests per window.
from time import time
from collections import defaultdict
class RateLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.clients = defaultdict(list)
def allow(self, ip: str) -> bool:
now…How to Implement an Adaptive Rate Limiter in Python
Build an adaptive rate limiter that adjusts request intervals dynamically based on recent error rates, slowing down when failures spike.
import time
import random
class AdaptiveRateLimiter:
"""Simple adaptive rate limiter that reduces requests when error rate is high."""
def __init__(self, min_interval=0.1, max_interval=2.0, error_threshold=0.3):
self.min_interval = min_interval
self.max_interval = max_interval
sel…How to Mock a Circuit Breaker Reset Timeout in Python
This code implements a simple circuit breaker with a reset timeout test, simulating a flaky service to show half-open state transitions.
import time
import random
class CircuitBreaker:
def __init__(self, failure_threshold=3, reset_timeout=5):
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED (nor…How to Mock a Liveness Check and Restart a Process in Python
Simulate a failing process and restart it after a liveness check fails, using a mock class and a liveness loop.
import subprocess
import sys
import time
import os
class ProcessMock:
def __init__(self, name, fail_after_seconds=3):
self.name = name
self.fail_after = fail_after_seconds
self.start_time = None
self.is_running = False
def start(self):
self.start_time = time.time()
…How to Propagate Context Variables with asyncio in Python
Use Python's ContextVar with asyncio to carry deadline information across concurrent tasks and propagate context automatically.
import asyncio
from contextvars import ContextVar
from datetime import datetime
deadline = ContextVar("deadline", default=None)
async def worker(name):
current = deadline.get()
if current:
print(f"{name} sees deadline: {current}")
else:
print(f"{name} sees no deadline")
await asyncio.…How to Send Messages to a Dead Letter Queue in Python
Simulates a poison message queue that retries failed messages up to a limit before moving them to a dead letter queue.
import json
class PoisonMessageQueue:
def __init__(self, max_retries=3):
self.dlq = []
self.max_retries = max_retries
self.processed_count = 0
self.failed_count = 0
def process_message(self, message_body):
if "poison" in message_body:
self.failed_count += 1…How to Simulate an Outbox Pattern with Reliable Retry in Python
This code implements a mock outbox pattern with records, delivery attempts, and retries to simulate reliable message publishing.
import time
import itertools
class Outbox:
def __init__(self):
self._records = []
self._seq = itertools.count(1)
def publish(self, topic, payload):
record = {
"id": next(self._seq),
"topic": topic,
"payload": payload,
"status": "pending"…How to implement a rate-limited shared counter in Python
Implements a thread-safe global counter that allows a maximum number of increments per second using a lock and time-based refill.
import threading
import time
import random
counter = 0
lock = threading.Lock()
MAX_CALLS_PER_SECOND = 3
last_refill = time.time()
def rate_limited_increment():
global counter, last_refill
with lock:
now = time.time()
if now - last_refill >= 1.0:
last_refill = now
count…How to implement rate limiting per API key in Python
A simple sliding-window rate limiter that tracks request timestamps per API key and rejects requests exceeding the configured limit.
import time
API_RATE_LIMITS = {"api_key_1": 5, "api_key_2": 3} # max requests per window
WINDOW_SECONDS = 10
class RateLimiter:
def __init__(self, limits, window):
self.limits = limits
self.window = window
self.requests = {key: [] for key in limits}
def allow(self, api_key):
…How to retry idempotent operations with a mock in Python
Wrap a flaky idempotent operation in a retry loop with exponential backoff, and use unittest.mock to deterministically test the str's behavior.
import random
import time
from unittest.mock import Mock
def idempotent_operation(value):
"""Simulate an idempotent operation that sometimes fails."""
if random.random() < 0.6: # 60% failure rate
raise ConnectionError("Temporary failure")
return value * 2
def retry_with_backoff(operation, max_…Browse by section
Each section groups closely related Python snippets.
Reliability & rate limiting — Python code examples
What you will find here
This page collects reliability & rate limiting snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.