Reference library

Reliability & rate limiting

Retries, exponential backoff, circuit breakers, token buckets, and idempotent handlers.

6 matches

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.

chaos-engineering random resilience

Python

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)…

15 0 Open

Reliability & rate limiting medium

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.

circuit-breaker resilience fault-tolerance

Python

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 ==…

15 0 Open

Reliability & rate limiting easy

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.

chaos-engineering decorators latency

Python

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…

12 0 Open

Reliability & rate limiting medium

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.

contextvars asyncio concurrency

Python

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.…

13 0 Open

Reliability & rate limiting easy

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.

rate-limiting time api

Python

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:
      …

13 0 Open

Reliability & rate limiting medium

Saga Compensating Transaction Mock in Python

Simulates a distributed transaction using a saga pattern with compensating actions that roll back steps on failure.

saga transaction compensation

Python

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…

12 0 Open

Browse by section

Each section groups closely related Python snippets.