Reference library

Reliability & rate limiting

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

3 matches

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

16 0 Open

Reliability & rate limiting easy

How to Stop Receiving Requests Until Ready in Python

A mock server that refuses requests until a readiness gate is passed, simulating fail-stop behavior for production reliability.

readiness fail-stop mock-server

Python

import random
import time


class MockServer:
    def __init__(self):
        self.ready = False
        self.requests_received = 0

    def readiness_check(self):
        """Simulates a readiness probe. Returns True only when ready."""
        if not self.ready:
            return False
        return True

    def r…

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.