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:
…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 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 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"…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:
…Retry with Exponential Backoff and Jitter in Python
A decorator-style retry wrapper that retries a flaky function with exponential backoff plus random jitter, then raises after the last attempt fails.
import random
import time
def retry_with_backoff(func, max_retries=3, base_delay=0.5, max_jitter=0.1):
for attempt in range(max_retries + 1):
try:
return func()
except Exception as e:
if attempt == max_retries:
raise
delay = base_delay * (2 ** at…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.