Reference library

Reliability & rate limiting

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

4 matches

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.

circuit-breaker resilience deque

Python

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…

16 0 Open

Reliability & rate limiting easy

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.

health-check reliability traffic-management

Python

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…

13 0 Open

Reliability & rate limiting easy

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.

quota rate-limiting class

Python

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…

17 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

Browse by section

Each section groups closely related Python snippets.