Reference library

Reliability & rate limiting

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

4 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:
          …

16 0 Open

Reliability & rate limiting easy

How to Deduplicate Messages in Python by ID

This code consumes a mock inbox of JSON messages and deduplicates them by message ID, keeping either the first or last occurrence.

deduplication inbox json

Python

import json
from collections import OrderedDict

mock_inbox = [
    {"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
    {"id": 2, "message": "world", "timestamp": "2024-01-01T10:01:00Z"},
    {"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
    {"id": 3, "message": "test", "times…

15 0 Open

Reliability & rate limiting medium

How to Implement Hedged Requests in Python

This code demonstrates a hedged request pattern using threading, which sends duplicate calls and returns the first result that arrives within a timeout.

hedged-requests threading timeout

Python

import time
from unittest.mock import Mock

def hedged_request(call, timeout=0.05):
    """Execute two duplicate calls, return first result within timeout."""
    result_container = {}

    def run_and_store():
        result_container['result'] = call()
        result_container['done'] = True

    # Simulate slow cal…

16 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.