Reliability & rate limiting
Retries, exponential backoff, circuit breakers, token buckets, and idempotent handlers.
At Least Once with Idempotent Consumer in Python
Implements a thread-safe idempotent consumer that processes each unique message exactly once, even when a producer sends duplicates under an at-least-once delivery model.
import threading
import time
import uuid
from collections import Counter
class IdempotentConsumer:
def __init__(self):
self.processed = set()
self._lock = threading.Lock()
def consume(self, message_id, payload):
with self._lock:
if message_id in self.processed:
…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…Build a queue-based admission control system in Python
Implement a simple bounded-queue admission controller that accepts or rejects incoming requests based on current queue capacity.
from collections import deque
import time
class AdmissionControl:
"""Simple admission control using a bounded queue.
Requests arrive at the queue; they are admitted in FIFO order.
If the queue is full, the incoming request is rejected.
"""
def __init__(self, capacity: int):
self.capacit…Chaos Inject Random Failures in Python
Simulate random failures in a Python function to test error handling and resilience, using random thresholds and controllable success rates.
import random
def unreliable_function(success_rate: float = 0.7) -> str:
"""Simulate a function that sometimes fails."""
if random.random() > success_rate:
raise ConnectionError("Simulated network failure")
return "Operation completed successfully"
if __name__ == "__main__":
random.seed(42)…Circuit breaker failure threshold count in Python
Track consecutive or time-windowed failures with a deque to open a circuit breaker and auto-recover to half-open after a cooldown.
from collections import deque
from time import time, sleep
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, recovery_time: float = 10.0):
self.failure_threshold = failure_threshold
self.recovery_time = recovery_time
self.failures: deque[float] = deque()
self.st…Exactly Once Processing Dedupe Mock in Python
Implements a streaming deduplicator using a set and queue to guarantee each item is processed exactly once while preserving insertion order.
from collections import deque
class DedupeStream:
def __init__(self):
self.seen = set()
self.queue = deque()
def add(self, item):
if item not in self.seen:
self.seen.add(item)
self.queue.append(item)
print(f"Processed: {item} (exactly once)")
…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 >=…GCRA generic cell rate algorithm in Python
Mock implementation of the Generic Cell Rate Algorithm (GCRA) for traffic shaping and rate limiting.
from collections import deque
import time
class GCRA:
def __init__(self, rate, burst):
self.tau = burst
self.T = rate
self.t = 0
self.LCT = 0
def add_cell(self, arrival_time):
if arrival_time <= self.t:
return False
arrived_early = (arrival_time - s…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
A beginner-friendly token bucket rate limiter with retry logic for handling API rate limits in Python.
import time
import random
class RateLimiter:
"""Simple token bucket rate limiter for beginners."""
def __init__(self, max_tokens=5, refill_rate=1.0):
self.max_tokens = max_tokens
self.tokens = max_tokens
self.refill_rate = refill_rate # tokens per second
self.last_refill …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 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 Deduplicate Messages in Python by ID
This code consumes a mock inbox of JSON messages and deduplicates them by message ID, keeping either the first or last occurrence.
import json
from collections import OrderedDict
mock_inbox = [
{"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
{"id": 2, "message": "world", "timestamp": "2024-01-01T10:01:00Z"},
{"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
{"id": 3, "message": "test", "times…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 Message Visibility Timeout Renewal in Python
Simulate queue message visibility control with timeout renewal using a simple Python class that tracks received time and visibility state.
import time
import uuid
class Message:
def __init__(self, body, visibility_timeout=30):
self.body = body
self.visibility_timeout = visibility_timeout
self.receipt_handle = str(uuid.uuid4())
self.received_at = time.time()
self.deleted = False
def is_visible(self):
…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 Dead Letter Queue Replay in Python
A mock Dead Letter Queue that stores failed messages with retry attempts and replays them with a simple retry counter.
import json
from collections import deque
class DeadLetterQueue:
def __init__(self):
self.messages = deque()
def add_message(self, message_id, payload, attempts=3):
"""Add a message to the DLQ with retry metadata."""
self.messages.append({
"id": message_id,
…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 a Sliding Window Counter in Python
This code implements an approximate sliding window counter using a deque of time-based buckets to track event counts within a recent time window.
from collections import deque
from time import time
class SlidingWindowCounter:
def __init__(self, window_size, bucket_size=1):
self.window_size = window_size
self.bucket_size = bucket_size
self.buckets = deque()
def _evict_expired(self, now):
while self.buckets and self.buck…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 Temporary Block in Python
Build a reusable PenaltyBox class that temporarily blocks access after a failure and reports remaining lockout time.
class PenaltyBox:
def __init__(self, block_seconds: int = 30):
self.block_seconds = block_seconds
self._blocked_until = 0.0
self._attempts = 0
def try_access(self, current_time: float) -> bool:
if self._blocked_until and current_time < self._blocked_until:
return Fa…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…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.