Reference library

Reliability & rate limiting

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

8 matches
Reliability & rate limiting easy

Build a queue-based admission control system in Python

Implement a simple bounded-queue admission controller that accepts or rejects incoming requests based on current queue capacity.

admission-control queue rate-limiting
Python
from collections import deque
import time


class AdmissionControl:
    """Simple admission control using a bounded queue.

    Requests arrive at the queue; they are admitted in FIFO order.
    If the queue is full, the incoming request is rejected.
    """

    def __init__(self, capacity: int):
        self.capacit…
16 0 Open
Reliability & rate limiting easy

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.

rate-limiting fixed-window time
Python
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 >=…
13 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 Implement Message Visibility Timeout Renewal in Python

Simulate queue message visibility control with timeout renewal using a simple Python class that tracks received time and visibility state.

visibility-timeout queue sqs
Python
import time
import uuid

class Message:
    def __init__(self, body, visibility_timeout=30):
        self.body = body
        self.visibility_timeout = visibility_timeout
        self.receipt_handle = str(uuid.uuid4())
        self.received_at = time.time()
        self.deleted = False

    def is_visible(self):
     …
13 0 Open
Reliability & rate limiting medium

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.

liveness restart mock
Python
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()
   …
14 0 Open
Reliability & rate limiting medium

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.

dlq message queue retries
Python
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…
15 0 Open
Reliability & rate limiting medium

Leaky Bucket Rate Limiter in Python: Smooth Burst Traffic

Implements a token-bucket-style leaky bucket rate limiter that smooths bursty traffic by draining at a fixed rate and dropping excess packets.

rate-limiting traffic-shaping simulation
Python
import time
import random


class LeakyBucket:
    def __init__(self, capacity, drain_rate):
        self.capacity = capacity
        self.drain_rate = drain_rate
        self.water = 0.0
        self.last_time = time.time()

    def allow(self, packet_size=1.0):
        now = time.time()
        elapsed = now - self.…
14 0 Open
Reliability & rate limiting medium

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.

two-phase commit distributed systems transactions
Python
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")
     …
12 0 Open

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.