17 — Low-level design

High-level design decides what the services are. Low-level design decides what happens inside one: the classes, interfaces, data structures and concurrency. Poor LLD turns a good architecture into something fragile and slow to change.

Core principles

Depend on interfaces. If domain logic imports a storage SDK directly, it can’t be tested without that infrastructure, and changing vendors means rewriting logic.

Prefer composition over inheritance. Deep inheritance trees mean a base-class change ripples everywhere. Assembling small components is more flexible.

Keep one reason to change. A class doing several jobs changes for several reasons, and each change risks the others.

SOLID

These are guidelines, not laws. Applied too hard, they produce layers of indirection around nothing.

Hexagonal architecture

Hexagonal architecture, also called ports and adapters, keeps business rules in the centre, and treats databases and HTTP as replaceable details.

[HTTP handler] --> [ Domain logic ] --> [Repository interface]
                                                 |
                                       [Postgres] or [In-memory]

Dependencies point inward: the domain knows nothing about SQL. So domain logic can be tested in memory, in milliseconds.

Separate models

A database row, a domain object and an API response often start identical, then drift apart. Keeping them as separate types stops a schema change leaking into a public API.

Design patterns

Patterns are named solutions to common problems. A few do most of the work in backend systems.

State machines deserve more use than they get. An order moving through CREATED, PAID, SHIPPED and CANCELLED has rules, and an explicit table of allowed transitions prevents impossible states.

Strategy, Factory and Observer together

A payment system could be one function full of if method == "CARD". Instead:

from abc import ABC, abstractmethod
from dataclasses import dataclass
from decimal import Decimal
from typing import Callable

# --- 1. Domain Models (Immutable Value Objects) ---
@dataclass(frozen=True)
class PaymentRequest:
    order_id: str
    user_id: str
    amount: Decimal
    currency: str
    method: str

@dataclass(frozen=True)
class PaymentReceipt:
    order_id: str
    transaction_id: str
    status: str

# --- 2. Strategy Pattern: Interchangeable Payment Gateways ---
class PaymentStrategy(ABC):
    @abstractmethod
    def charge(self, request: PaymentRequest) -> PaymentReceipt:
        # Executes the charge against the underlying provider.
        pass

class StripeCardStrategy(PaymentStrategy):
    def charge(self, request: PaymentRequest) -> PaymentReceipt:
        # In production: call Stripe SDK with idempotency key = request.order_id
        return PaymentReceipt(
            order_id=request.order_id,
            transaction_id=f"ch_stripe_{request.order_id}",
            status="SUCCEEDED",
        )

class BankWireStrategy(PaymentStrategy):
    def charge(self, request: PaymentRequest) -> PaymentReceipt:
        # In production: initiate ISO-20022 wire transfer
        return PaymentReceipt(
            order_id=request.order_id,
            transaction_id=f"wire_{request.order_id}",
            status="PENDING_SETTLEMENT",
        )

# --- 3. Registry / Factory Pattern: Eliminating Conditional Sprawl ---
class PaymentStrategyFactory:
    def __init__(self):
        self._strategies: dict[str, PaymentStrategy] = {}

    def register(self, method_name: str, strategy: PaymentStrategy) -> None:
        self._strategies[method_name.upper()] = strategy

    def get_strategy(self, method_name: str) -> PaymentStrategy:
        strategy = self._strategies.get(method_name.upper())
        if not strategy:
            raise ValueError(f"Unsupported payment method: {method_name}")
        return strategy

# --- 4. Observer Pattern: Decoupling Side Effects ---
class CheckoutService:
    def __init__(self, factory: PaymentStrategyFactory):
        self._factory = factory
        self._listeners: list[Callable[[PaymentRequest, PaymentReceipt], None]] = []

    def subscribe(self, listener: Callable[[PaymentRequest, PaymentReceipt], None]) -> None:
        self._listeners.append(listener)

    def checkout(self, request: PaymentRequest) -> PaymentReceipt:
        strategy = self._factory.get_strategy(request.method)
        receipt = strategy.charge(request)
        
        # Notify all registered observers without coupling CheckoutService to their internals
        for listener in self._listeners:
            listener(request, receipt)
            
        return receipt

# --- Wiring It Together ---
if __name__ == "__main__":
    factory = PaymentStrategyFactory()
    factory.register("CARD", StripeCardStrategy())
    factory.register("WIRE", BankWireStrategy())

    checkout_service = CheckoutService(factory)

    # Attach decoupled observers for audit logging and ledger entries
    checkout_service.subscribe(
        lambda req, rec: print(f"[AUDIT] Order {req.order_id} charged {req.amount} {req.currency} -> {rec.status}")
    )
    checkout_service.subscribe(
        lambda req, rec: print(f"[LEDGER] Recorded txn {rec.transaction_id} for user {req.user_id}")
    )

    req = PaymentRequest(order_id="ORD-9921", user_id="USR-42", amount=Decimal("149.99"), currency="USD", method="CARD")
    checkout_service.checkout(req)

Output:

[AUDIT] Order ORD-9921 charged 149.99 USD -> SUCCEEDED
[LEDGER] Recorded txn ch_stripe_ORD-9921 for user USR-42

Adding Apple Pay means one new class and one registration, with no change to CheckoutService.

Patterns have a cost. Each one adds indirection, so a two-branch if that never grows doesn’t need a strategy.

Errors

Expected failures such as “insufficient funds” belong in the domain, as return values or typed errors. Unexpected failures such as a lost database connection should propagate to a level that can handle them. Errors should keep their context, and never be silently swallowed.

Immutability

An object that can’t change can be shared across threads without locks, and it can’t be altered behind a caller’s back. frozen=True on the dataclasses above does this.

Testing

Good LLD makes testing easy.

Most tests should sit at the fast end. Time is best injected as a dependency, so tests don’t have to sleep.

Concurrency

The concurrency guide covers this in depth. In short, a race condition occurs when the result depends on timing. count += 1 is a read, an add and a write, so two threads can both read 5 and both write 6.

This program makes that visible by yielding between the read and the write:

import threading, time

count = 0

def add(times, lock=None):
    global count
    for _ in range(times):
        if lock:
            lock.acquire()
        value = count
        time.sleep(0)        # another thread may run here
        count = value + 1
        if lock:
            lock.release()

def run(lock=None):
    global count
    count = 0
    threads = [threading.Thread(target=add, args=(2000, lock)) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return count

print("without a lock:", run(), "of 8000")
print("with a lock:   ", run(threading.Lock()), "of 8000")

Output:

without a lock: 2000 of 8000
with a lock:    8000 of 8000

Three quarters of the updates vanished without a lock.

A thread-safe LRU cache with TTL

This combines the ideas above. To get O(1) get and put, it uses a dict for lookup, and a doubly linked list for recency. Sentinel head and tail nodes remove the empty-list edge cases.

import threading
import time
from typing import Any, Optional

class _Node:
    # Internal Doubly Linked List node storing key, value, and expiration timestamp.
    __slots__ = ("key", "value", "expires_at", "prev", "next")

    def __init__(self, key: str, value: Any, expires_at: float):
        self.key = key
        self.value = value
        self.expires_at = expires_at
        self.prev: Optional["_Node"] = None
        self.next: Optional["_Node"] = None

class ThreadSafeLRUCache:
    # O(1) Thread-Safe Least Recently Used (LRU) Cache with per-key TTL expiration.
    # Uses a Hash Map + Doubly Linked List with sentinel head/tail nodes.
    def __init__(self, capacity: int):
        if capacity <= 0:
            raise ValueError("Capacity must be greater than 0")
        self.capacity = capacity
        self.cache: dict[str, _Node] = {}
        self.lock = threading.RLock()

        # Dummy sentinel head (Most Recently Used) and tail (Least Recently Used)
        # Sentinel nodes eliminate null-pointer edge cases during node insertion/removal
        self.head = _Node("", None, 0.0)
        self.tail = _Node("", None, 0.0)
        self.head.next = self.tail
        self.tail.prev = self.head

    def _remove_node(self, node: _Node) -> None:
        # Unlinks a node from the doubly linked list in O(1) time. Must hold lock.
        prev_node = node.prev
        next_node = node.next
        prev_node.next = next_node
        next_node.prev = prev_node

    def _insert_at_head(self, node: _Node) -> None:
        # Inserts a node right after the sentinel head (marks as Most Recently Used).
        node.next = self.head.next
        node.prev = self.head
        self.head.next.prev = node
        self.head.next = node

    def get(self, key: str) -> Optional[Any]:
        with self.lock:
            node = self.cache.get(key)
            if not node:
                return None

            # Lazy TTL Eviction: check if the entry has expired
            if time.monotonic() > node.expires_at:
                self._remove_node(node)
                del self.cache[key]
                return None

            # Move accessed node to the head (Most Recently Used)
            self._remove_node(node)
            self._insert_at_head(node)
            return node.value

    def put(self, key: str, value: Any, ttl_seconds: float = 60.0) -> None:
        expires_at = time.monotonic() + ttl_seconds
        with self.lock:
            if key in self.cache:
                node = self.cache[key]
                node.value = value
                node.expires_at = expires_at
                self._remove_node(node)
                self._insert_at_head(node)
                return

            # Evict Least Recently Used node (right before sentinel tail) if at capacity
            if len(self.cache) >= self.capacity:
                lru_node = self.tail.prev
                self._remove_node(lru_node)
                del self.cache[lru_node.key]

            new_node = _Node(key, value, expires_at)
            self.cache[key] = new_node
            self._insert_at_head(new_node)

Test results: with capacity 2, after inserting a and b, reading a and inserting c, b was evicted. An entry with a 0.05 s TTL was gone after 0.1 s. Eight threads running 20,000 rounds each left the cache at capacity with a matching list.

Nothing here re-enters the lock, so a plain threading.Lock would work as well as RLock.

The takeaway

Keep domain logic independent of infrastructure, and use patterns only when they earn their cost. Make invalid states impossible with types and state machines. Protect shared state, or better, avoid sharing it.