Reference library

System design patterns

Sharding, load balancing, CAP tradeoffs, and scaling patterns — interview and production ready.

4 matches

Idempotent Consumer: Store Processed IDs in Python

Implement an idempotent consumer that persists processed message IDs to a JSON file, skipping duplicates on restart.

idempotency duplicate-detection state-persistence

import json
from pathlib import Path


class IdempotentStore:
    def __init__(self, storage_path: str = "processed_ids.json"):
        self.storage_path = Path(storage_path)
        self.processed_ids = self._load()

    def _load(self) -> set:
        if self.storage_path.exists():
            with self.storage_path…

15 0 Open

System design patterns medium

Inbox pattern consumer dedupe mock in Python

Implements a mock inbox consumer that deduplicates incoming messages by ID, with automatic eviction of old seen IDs to prevent unbounded memory growth.

deduplication inbox-pattern dataclasses

Python

import json
from collections import deque
from dataclasses import dataclass, field
from hashlib import sha256
from typing import Any


@dataclass
class InboxConsumer:
    max_seen: int = 1000
    seen_ids: set = field(default_factory=set)
    seen_history: deque = field(default_factory=deque)

    def _mark_seen(self,…

13 0 Open

System design patterns medium

Outbox pattern reliable publish in Python with SQLite

Implements a transactional outbox with SQLite, ensuring reliable message publishing by storing events in the same DB transaction as business changes.

outbox sqlite transaction

Python

import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone


class Outbox:
    def __init__(self, db_path=":memory:"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS outbox (
                id INTEGER PRIMARY KEY AUTO…

11 0 Open

System design patterns easy

Route Messages to Handlers with a Python Dict

This code demonstrates a simple message routing pattern using a dictionary to map topic keys to handler functions, with a default handler for unmatched topics.

routing dictionary message-broker

Python

def route_message(message, routing_table):
    """Route a message to the correct handler based on the topic key."""
    topic = message.get("topic", "default")
    handler = routing_table.get(topic, routing_table.get("default"))
    return handler(message)


def handle_orders(message):
    return f"Orders handler proc…

12 0 Open

Browse by section

Each section groups closely related Python snippets.