Reference library

Reliability & rate limiting

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

3 matches

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

How to Implement a Dead Letter Queue Replay in Python

A mock Dead Letter Queue that stores failed messages with retry attempts and replays them with a simple retry counter.

dead-letter-queue queue retry

Python

import json
from collections import deque

class DeadLetterQueue:
    def __init__(self):
        self.messages = deque()
    
    def add_message(self, message_id, payload, attempts=3):
        """Add a message to the DLQ with retry metadata."""
        self.messages.append({
            "id": message_id,
           …

13 0 Open

Browse by section

Each section groups closely related Python snippets.