Reliability & rate limiting
Retries, exponential backoff, circuit breakers, token buckets, and idempotent handlers.
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…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 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 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 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")
…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.