Reliability & rate limiting
Retries, exponential backoff, circuit breakers, token buckets, and idempotent handlers.
Build a Rate Limiter Decorator in Python
This code defines a reusable rate limiter decorator that caps function calls within a sliding time window using a deque and monotonic time.
import time
from collections import deque
def rate_limiter(max_calls: int, period: float):
calls = deque()
def decorator(func):
def wrapper(*args, **kwargs):
now = time.monotonic()
while calls and now - calls[0] >= period:
calls.popleft()
if len(ca…Fixed Window Counter Rate Limiting in Python
A simple fixed window counter rate limiter that allows a maximum number of requests per 60-second window, with a mock time simulation.
from collections import deque
from time import time
class FixedWindowCounter:
def __init__(self, max_requests):
self.max_requests = max_requests
self.window_start = int(time())
self.window_count = 0
def allow_request(self):
current_time = int(time())
if current_time >=…Health Check Mark Unhealthy Stop Traffic Mock in Python
Simulates a health check with a 20% failure rate and automatically stops traffic when the service is unhealthy.
import time
import random
class HealthCheck:
def __init__(self):
self.is_healthy = True
self.stop_traffic = False
def check_health(self):
# Simulate health check with random failure rate (20% chance unhealthy)
self.is_healthy = random.random() > 0.2
return self.is_heal…How to Build a Rate Limiter in Python
Implements a simple sliding-window rate limiter that caps the number of calls per period, used to throttle processing of a data list.
import time
class RateLimiter:
def __init__(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.timestamps = []
def allow(self):
now = time.time()
self.timestamps = [t for t in self.timestamps if now - t < self.period]
if len(self.tim…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 Rate Limiter in Python
A beginner-friendly Python class that tracks call timestamps with a deque to allow or block calls based on a max rate per time period.
import time
from collections import deque
class RateLimiter:
"""Simple rate limiter for beginners."""
def __init__(self, max_calls: int, period_seconds: float):
self.max_calls = max_calls
self.period = period_seconds
self.calls = deque()
def allow(self) -> bool:
"""Retur…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 Daily and Monthly Quota Counters in Python
Track daily and monthly API call usage with automatic resets, quota checks, and limits using a Python class.
import random
from datetime import datetime, timedelta
class QuotaCounter:
def __init__(self, daily_limit=1000, monthly_limit=20000):
self.daily_limit = daily_limit
self.monthly_limit = monthly_limit
self.daily_usage = 0
self.monthly_usage = 0
self.current_day = datetime.n…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 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 an idempotency key store in Python
Build an in-memory idempotency key store with TTL that processes a request once and reuses the cached result for duplicate calls.
import hashlib
import time
from typing import Dict, Optional
class IdempotencyStore:
"""Simple in-memory idempotency key store with mock processing."""
def __init__(self, ttl_seconds: int = 3600) -> None:
self.ttl = ttl_seconds
self._store: Dict[str, tuple[str, float]] = {}
def _is_expi…How to implement rate limiting in Python
Build a simple sliding-window rate limiter in Python that enforces a max number of calls per time period and formats data with timestamps.
import time
class RateLimiter:
def __init__(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.calls = []
def allow(self):
now = time.time()
# Remove calls older than the period window
self.calls = [t for t in self.calls if now -…How to implement rate limiting in Python
A beginner-friendly Python rate limiter that throttles API calls and retries parsing tasks with exponential backoff.
import time
import random
class RateLimiter:
def __init__(self, max_calls, per_seconds):
self.max_calls = max_calls
self.per_seconds = per_seconds
self.timestamps = []
def allow(self):
now = time.time()
self.timestamps = [t for t in self.timestamps if now - t < sel…How to mock a fallback return value in Python
Test a function that returns a default value on failure by mocking requests.get and its side effects.
from unittest.mock import Mock, patch
import requests
def fetch_data(url, default=None):
try:
response = requests.get(url)
response.raise_for_status()
return response.json()
except (requests.RequestException, ValueError):
return default
with patch("requests.get") as mock_get:
…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_…Implement a Circuit Breaker Pattern in Python
This code implements a simple circuit breaker that opens after a threshold of consecutive failures, causing subsequent calls to fail fast without invoking the underlying function.
class CircuitBreaker:
def __init__(self, failure_threshold=3):
self.failure_threshold = failure_threshold
self.failure_count = 0
self.open = False
def call(self, func, *args, **kwargs):
if self.open:
raise RuntimeError("Circuit is open - failing fast")
try:
…Implementing Fallback with Cached Stale Data in Python
This code demonstrates a resilient data-fetching pattern that caches successful responses, falls back to cached data when the external API fails, and returns stale data as a last-resort fallback.
import random
import time
# Simulated cache dictionary: key -> (value, timestamp)
_cache = {}
_CACHE_TTL = 3 # seconds
# Mock data source (simulates an unreliable external API)
def fetch_mock_data(key):
failure = random.random() < 0.4 # 40% chance of failure
if failure:
raise ConnectionError("Mock …Rate Limiting in Python with a Sliding Window
A beginner-friendly dataclass-based sliding window rate limiter that controls how many calls are allowed per time window.
import time
from dataclasses import dataclass
@dataclass
class RateLimiter:
max_calls: int
window_seconds: float = 1.0
def __post_init__(self):
self.calls = []
self._start = time.monotonic()
def _update(self, now):
self.calls = [t for t in self.calls if now - t < self.window…Rate Limiting with a Simple Python RateLimiter Class
A beginner-friendly Python rate limiter that tracks call timestamps and enforces a maximum number of calls within a rolling time window, with a helper to validate positive integers.
import time
class RateLimiter:
def __init__(self, max_calls, period_seconds):
self.max_calls = max_calls
self.period_seconds = period_seconds
self.calls = []
def is_allowed(self):
now = time.time()
while self.calls and now - self.calls[0] >= self.period_seconds:
…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.