Reference library

System design patterns

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

6 matches

Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States

Implement a circuit breaker with closed, open, and half-open states to prevent repeated calls to failing services and allow recovery after a timeout.

circuit-breaker resilience fault-tolerance

Python

class CircuitBreaker:
    def __init__(self, failure_threshold=3, timeout_seconds=5):
        self.failure_threshold = failure_threshold
        self.timeout_seconds = timeout_seconds
        self.state = "closed"
        self.failure_count = 0
        self.last_failure_time = None

    def record_success(self):
     …

14 0 Open

System design patterns easy

How to Build an Append-Only Event Store in Python

Implement a simple append-only event store class that stores events in a list and supports retrieval by index range.

event-sourcing append-only event-store

Python

class EventStore:
    def __init__(self):
        self._events = []

    def append(self, event):
        """Append an event to the store."""
        self._events.append(event)

    def get_events(self, start=0, end=None):
        """Return events from start index to end (exclusive)."""
        return self._events[sta…

14 0 Open

System design patterns easy

How to Implement a Simple Event Bus in Python

Create a publish-subscribe event bus using dataclasses and defaultdict to decouple event producers from consumers.

event-bus publish-subscribe design-patterns

Python

from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Set


@dataclass
class EventBus:
    _subscribers: Dict[str, List[Callable]] = field(
        default_factory=lambda: defaultdict(list)
    )

    def subscribe(self, event_type: str, handler: Callable…

15 0 Open

System design patterns medium

How to Limit Concurrent Requests with a Semaphore in Python

Use threading.Semaphore with a ThreadPoolExecutor to cap how many worker threads run simultaneously, preventing resource overload.

concurrency semaphore threading

Python

import threading
import time
from concurrent.futures import ThreadPoolExecutor

def worker(name, semaphore, results):
    with semaphore:
        results.append(f"start {name}")
        time.sleep(0.5)  # simulate async work
        results.append(f"done {name}")

def main():
    sem = threading.Semaphore(2)  # max 2 …

14 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

Browse by section

Each section groups closely related Python snippets.