Reference library

Reliability & rate limiting

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

4 matches

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

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.