Reference library

Reliability & rate limiting

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

5 matches

At Least Once with Idempotent Consumer in Python

Implements a thread-safe idempotent consumer that processes each unique message exactly once, even when a producer sends duplicates under an at-least-once delivery model.

idempotency at-least-once threading

Python

import threading
import time
import uuid
from collections import Counter


class IdempotentConsumer:
    def __init__(self):
        self.processed = set()
        self._lock = threading.Lock()

    def consume(self, message_id, payload):
        with self._lock:
            if message_id in self.processed:
          …

15 0 Open

Reliability & rate limiting easy

Exactly Once Processing Dedupe Mock in Python

Implements a streaming deduplicator using a set and queue to guarantee each item is processed exactly once while preserving insertion order.

deduplication exactly-once streaming

Python

from collections import deque

class DedupeStream:
    def __init__(self):
        self.seen = set()
        self.queue = deque()

    def add(self, item):
        if item not in self.seen:
            self.seen.add(item)
            self.queue.append(item)
            print(f"Processed: {item} (exactly once)")
      …

15 0 Open

Reliability & rate limiting easy

How to Build a Rate Limiter in Python

Implements a simple sliding-window rate limiter that caps the number of calls per period, used to throttle processing of a data list.

rate-limiting time sliding-window

Python

import time

class RateLimiter:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.timestamps = []

    def allow(self):
        now = time.time()
        self.timestamps = [t for t in self.timestamps if now - t < self.period]
        if len(self.tim…

14 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 easy

How to implement an idempotency key store in Python

Build an in-memory idempotency key store with TTL that processes a request once and reuses the cached result for duplicate calls.

idempotency cache ttl

Python

import hashlib
import time
from typing import Dict, Optional


class IdempotencyStore:
    """Simple in-memory idempotency key store with mock processing."""

    def __init__(self, ttl_seconds: int = 3600) -> None:
        self.ttl = ttl_seconds
        self._store: Dict[str, tuple[str, float]] = {}

    def _is_expi…

15 0 Open

Browse by section

Each section groups closely related Python snippets.