Reliability & rate limiting
Retries, exponential backoff, circuit breakers, token buckets, and idempotent handlers.
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 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 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 Inject Random Latency for Chaos Testing in Python
Mock unreliable services by wrapping functions with a decorator that adds random network-like delays before execution.
import random
import time
from functools import wraps
def inject_latency(func):
@wraps(func)
def wrapper(*args, **kwargs):
latency = random.uniform(0.1, 0.5)
print(f"Injecting {latency:.3f}s latency...")
time.sleep(latency)
return func(*args, **kwargs)
return wrapper
@inje…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 Mock Fault Injection Percentage in Python
Simulate a service with a 30% failure rate using random.random to test error handling and retries.
import random
class Service:
def call(self):
if random.random() < 0.3: # 30% failure rate
raise ConnectionError("Simulated network fault")
return "ok"
def main():
svc = Service()
random.seed(42) # deterministic for demonstration
results = []
for _ in range(10):
…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 Mock a Slow Startup Probe in Python
Simulate slow service initialization with a configurable mock delay to test readiness probes.
import time
from dataclasses import dataclass, field
@dataclass
class StartupProbe:
name: str
min_wait_sec: float = 0.5
max_wait_sec: float = 2.0
_ready: bool = field(default=False, init=False, repr=False)
def initialize(self) -> None:
"""Simulate slow startup with a fixed mock delay."""…How to Mock a Timeout per HTTP Request in Python
Simulate a per-request HTTP timeout using unittest.mock to test timeout handling without network access.
import time
from unittest.mock import Mock, patch
# Simulate an HTTP client that might time out
def fetch_data(url, timeout=5):
time.sleep(0.5) # Simulate network delay
return f"Response from {url}"
# Mock to test timeout behavior without real network
def test_timeout():
mock_response = Mock(side_effect…How to Mock a Try Confirm Cancel Pattern in Python
Define a simple class with confirm and cancel methods, execute a try confirm with error handling, and print the final state.
class TCC:
def __init__(self):
self.confirmed = False
self.cancelled = False
def confirm(self):
self.confirmed = True
return "confirmed"
def cancel(self):
self.cancelled = True
return "cancelled"
def try_confirm(self):
try:
result =…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 Stop Receiving Requests Until Ready in Python
A mock server that refuses requests until a readiness gate is passed, simulating fail-stop behavior for production reliability.
import random
import time
class MockServer:
def __init__(self):
self.ready = False
self.requests_received = 0
def readiness_check(self):
"""Simulates a readiness probe. Returns True only when ready."""
if not self.ready:
return False
return True
def r…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_…Mock Distributed Rate Limiter with Dict in Python
Simulates a distributed token-bucket rate limiter with a thread-safe dict, useful for testing before moving to Redis.
import time
import threading
from collections import defaultdict
class DistributedRateLimiter:
"""
A mock distributed rate limiter using a dict with thread-safe access.
Implements a token bucket algorithm per user.
"""
def __init__(self, rate_per_second=5, burst_capacity=10):
self.rate_p…Mock a Two-Phase Commit Coordinator in Python
Simulates a two-phase commit protocol where a coordinator asks participants to prepare, then commits or aborts based on unanimous readiness.
import random
import time
from typing import Dict, List
class TwoPhaseCommitCoordinator:
def __init__(self, participants: List[str]):
self.participants = participants
self.participant_state: Dict[str, bool] = {}
def prepare(self) -> bool:
print("[Coordinator] Phase 1: Prepare")
…Rate Limit per User ID in Python with a Dict Mock
Implements a simple sliding window rate limiter using a defaultdict of timestamps per user ID, blocking requests that exceed a max count within a time window.
import time
from collections import defaultdict
class RateLimiter:
def __init__(self, max_requests, window_seconds):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.user_timestamps = defaultdict(list)
def allow_request(self, user_id):
now = time.tim…Saga Compensating Transaction Mock in Python
Simulates a distributed transaction using a saga pattern with compensating actions that roll back steps on failure.
import random
import time
class OrderService:
def __init__(self):
self.orders = {}
def create_order(self, order_id):
print(f"[Order] Creating order {order_id}...")
time.sleep(0.1)
if random.random() < 0.3: # 30% chance of failure
raise RuntimeError(f"Order {order…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.