Low-Level Design (LLD)

The low-level design round (also called object-oriented design, OOD, or machine coding) asks you to turn a fuzzy prompt like "design a parking lot" or "build a rate limiter" into clean, extensible, correct code in under an hour. It is graded less on cleverness than on judgement: did you ask the right questions, find the right entities, put behaviour in the right place, and leave seams for the extensions the interviewer is about to ask for? This page is meant to be enough on its own. It covers what the round is and how it's graded, a minute-by-minute approach, OOP/SOLID with bad-to-good code, the design patterns that actually come up, UML, concurrency, ten classic problems worked in depth, eight AI-flavoured problems that AI-engineering loops increasingly use, testing, machine-coding tactics, and a question bank. Code is typed Python, with notes on the Java idioms interviewers often expect. Every Python snippet on this page was executed before publishing; the Java snippets were not compiled (see the note in §3).

1. What the LLD round is and how it's graded

The three variants

LLD vs HLD

The two overlap. A rate limiter is an LLD problem in one process and an HLD problem when it's shared across 200 API servers (Redis, Lua scripts, clock skew). Say which you are solving, and mention the other briefly: "In a distributed setting I'd move the bucket state to Redis with an atomic script, but the class design stays the same." The HLD side is covered in B6 · AI System Design and C2.

What's graded

How expectations change with seniority

2. A step-by-step approach, with timelines

The same eight steps work for every LLD prompt. What changes with the clock is how long you spend on each, and whether the code has to run.

1. Clarify requirements. Functional: what operations, which actors, what inputs and outputs. Non-functional: single process or distributed? concurrent callers? persistence? approximate scale (10 spots or 10,000)? latency targets? Scope: explicitly list what you will not build (payments, auth, UI).
2. List use cases. Three to six verbs from the actor's point of view: park(vehicle), unpark(ticket), pay(ticket), display_free_spots(). These become your public API.
3. Find entities and relationships. Underline nouns in the requirements (candidate classes) and verbs (candidate methods). Discard nouns that are just attributes (colour, name). Decide has-a vs is-a, and multiplicity (a Floor has many Spots).
4. Define interfaces before implementations. Where will behaviour vary? Allocation policy, pricing, notification channel, storage. Those get an ABC/Protocol. Everything else can be a concrete class.
5. Draw the class diagram. Five to ten boxes, with the main relationships. Don't draw every getter.
6. Walk the key flows as sequence diagrams. For the two most important use cases, trace which object calls which. This is where you find misplaced responsibilities ("who computes the fee: the Ticket, the Lot, or a PricingStrategy?").
7. Implement the core happy path first. Enums and value objects, then entities, then the service/orchestrator, then a tiny main or test that runs it.
8. Harden and extend. Edge cases and errors, then extensibility (add the second strategy), then concurrency, then tests. Finish by talking through trade-offs and what you'd do next.

Timeline by round length

These are guides, not rules. If the interviewer cuts requirements short ("assume whatever is reasonable"), say your assumptions out loud in one sentence each and move on.

Clarifying questions that work for almost any prompt

Fill-in checklist template

Copy this into the top of your file (or the corner of the whiteboard) in the first five minutes, and fill it in as you go. It keeps you and the interviewer aligned and gives you a script for the wrap-up.

3. Foundations: OOP, SOLID and friends

The four pillars, in practice

Composition over inheritance

Inheritance fixes behaviour at class-definition time and couples the child to the parent's internals. Composition (an object has collaborators behind interfaces) lets behaviour vary per instance and at runtime, and stops the class count from multiplying: with two independent dimensions of variation (fly × quack) inheritance needs a class per combination, while composition needs one class per behaviour. GoF puts the guideline as favouring object composition over class inheritance. GoF 1994

from typing import Protocol


# Inheritance explosion: FlyingDuck, SwimmingDuck, FlyingSwimmingQuackingDuck, ...
# Composition: a Duck HAS behaviours that can be swapped, even at runtime.
class FlyBehavior(Protocol):
    def fly(self) -> str: ...


class Wings:
    def fly(self) -> str: return "flap flap"


class NoFly:
    def fly(self) -> str: return "can't fly"


class Duck:
    def __init__(self, name: str, flyer: FlyBehavior) -> None:
        self.name, self._flyer = name, flyer

    def fly(self) -> str:
        return f"{self.name}: {self._flyer.fly()}"

    def set_flyer(self, flyer: FlyBehavior) -> None:   # behaviour change at runtime
        self._flyer = flyer


rubber = Duck("rubber", NoFly())
assert rubber.fly() == "rubber: can't fly"
rubber.set_flyer(Wings())
assert rubber.fly() == "rubber: flap flap"

Inheritance is still the right tool when (a) there's a real is-a relationship, (b) the parent's contract is stable, and (c) subclasses don't need to override things in ways that break it. Abstract base classes that define an interface (with maybe a template method) are the safe kind of inheritance.

SOLID, with bad → good code

Robert C. Martin collected these five principles; the acronym came later. Wikipedia: SOLID He has written about their continued relevance on his blog. Martin 2020

S: Single Responsibility Principle

A class should have one reason to change, meaning one stakeholder or concern whose requests would make you edit it. Martin 2014 An invoice that computes totals, renders PDFs, saves itself and sends email changes whenever finance, design, the DBA or the email team asks for something.

class Invoice:
    def __init__(self, items: list[tuple[str, int]]) -> None:
        self.items = items

    def total(self) -> int:
        return sum(price for _, price in self.items)

    def to_pdf(self) -> bytes:            # presentation concern
        ...

    def save(self, db) -> None:           # persistence concern
        db.execute("INSERT ...")

    def email(self, smtp) -> None:        # delivery concern
        smtp.send(...)

from dataclasses import dataclass, field
from typing import Protocol


@dataclass
class Invoice:
    items: list[tuple[str, int]] = field(default_factory=list)

    def total(self) -> int:               # domain logic only
        return sum(price for _, price in self.items)


class InvoiceRenderer(Protocol):
    def render(self, invoice: Invoice) -> bytes: ...


class InvoiceRepository(Protocol):
    def save(self, invoice: Invoice) -> None: ...


class InvoiceSender(Protocol):
    def send(self, invoice: Invoice, to: str) -> None: ...

"One reason to change" doesn't mean "one method". Invoice can have ten methods as long as they're all domain logic about invoices.

O: Open/Closed Principle

Open for extension, closed for modification: you add behaviour by adding code, not by editing working code. The tell-tale violation is an if/elif on a type string that grows with every feature.

def shipping_cost(order, method: str) -> float:
    if method == "standard":
        return 5.0
    elif method == "express":
        return 5.0 + order.weight * 1.5
    elif method == "drone":          # every new method edits this function
        return 20.0
    raise ValueError(method)

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass(frozen=True)
class Order:
    weight: float


class ShippingMethod(ABC):
    @abstractmethod
    def cost(self, order: Order) -> float: ...


class Standard(ShippingMethod):
    def cost(self, order: Order) -> float:
        return 5.0


class Express(ShippingMethod):
    def cost(self, order: Order) -> float:
        return 5.0 + order.weight * 1.5


class Drone(ShippingMethod):          # new behaviour = new class, no edits elsewhere
    def cost(self, order: Order) -> float:
        return 20.0


def shipping_cost(order: Order, method: ShippingMethod) -> float:
    return method.cost(order)


assert shipping_cost(Order(2.0), Express()) == 8.0

Pair this with a registry or factory so that adding Drone is one new class plus one registration line.

L: Liskov Substitution Principle

Anything that works with the parent type must keep working with a subtype: no stronger preconditions, no weaker postconditions, no surprising side effects. Wikipedia: LSP The classic counter-example is a mutable Square extending Rectangle.

class Rectangle:
    def __init__(self, w: int, h: int) -> None:
        self.w, self.h = w, h

    def set_width(self, w: int) -> None:
        self.w = w

    def area(self) -> int:
        return self.w * self.h


class Square(Rectangle):
    def set_width(self, w: int) -> None:   # silently changes height too
        self.w = self.h = w


def stretch(r: Rectangle) -> None:
    r.set_width(10)
    assert r.area() == 10 * r.h            # holds for Rectangle; Square breaks the caller's expectations

from abc import ABC, abstractmethod
from dataclasses import dataclass


class Shape(ABC):
    @abstractmethod
    def area(self) -> int: ...


@dataclass(frozen=True)
class Rectangle(Shape):
    w: int
    h: int

    def area(self) -> int:
        return self.w * self.h

    def with_width(self, w: int) -> "Rectangle":
        return Rectangle(w, self.h)


@dataclass(frozen=True)
class Square(Shape):
    side: int

    def area(self) -> int:
        return self.side * self.side


assert Rectangle(2, 3).with_width(10).area() == 30
assert Square(3).area() == 9

Other LSP smells to spot in your own designs: a subclass that raises NotImplementedError for an inherited method, a ReadOnlyList extending list, or an override that silently ignores an argument.

I: Interface Segregation Principle

Clients shouldn't depend on methods they don't use. Prefer several small role interfaces to one wide one.

class Worker(ABC):
    @abstractmethod
    def work(self) -> None: ...
    @abstractmethod
    def eat(self) -> None: ...
    @abstractmethod
    def sleep(self) -> None: ...


class Robot(Worker):
    def work(self) -> None: ...
    def eat(self) -> None:
        raise NotImplementedError      # forced to implement what it doesn't need
    def sleep(self) -> None:
        raise NotImplementedError

from typing import Protocol


class Workable(Protocol):
    def work(self) -> None: ...


class Feedable(Protocol):
    def eat(self) -> None: ...


class Human:
    def work(self) -> None: print("typing")
    def eat(self) -> None: print("lunch")


class Robot:
    def work(self) -> None: print("welding")


def run_shift(workers: list[Workable]) -> None:   # depends only on what it uses
    for w in workers:
        w.work()


run_shift([Human(), Robot()])

In Python, typing.Protocol gives structural ("duck") typing that a type checker can verify, which fits ISP well: a class satisfies Workable just by having work(). PEP 544

D: Dependency Inversion Principle

High-level policy shouldn't depend on low-level details; both should depend on abstractions. Practically: constructors take interfaces, and one "composition root" (your main) decides which concrete classes to wire in.

import smtplib


class OrderService:
    def __init__(self) -> None:
        self.mailer = smtplib.SMTP("smtp.internal")   # hard-wired concrete dependency

    def place(self, order) -> None:
        ...
        self.mailer.sendmail("shop@x.com", order.email, "Thanks!")

from typing import Protocol


class Notifier(Protocol):
    def notify(self, to: str, msg: str) -> None: ...


class OrderService:
    def __init__(self, notifier: Notifier) -> None:   # injected abstraction
        self._notifier = notifier

    def place(self, email: str) -> None:
        # ... persist order ...
        self._notifier.notify(email, "Thanks!")


class FakeNotifier:                                     # trivial test double
    def __init__(self) -> None:
        self.sent: list[tuple[str, str]] = []

    def notify(self, to: str, msg: str) -> None:
        self.sent.append((to, msg))


fake = FakeNotifier()
OrderService(fake).place("a@b.com")
assert fake.sent == [("a@b.com", "Thanks!")]

DRY, KISS, YAGNI

• DRY (Don't Repeat Yourself) is about knowledge, not text. Two identical-looking blocks that change for different reasons are not duplication; merging them couples unrelated things. Two places that encode the same business rule (the fee formula) are duplication even if they look different.
• KISS (Keep It Simple): the simplest design that meets the agreed requirements. In a 60-minute round, one lock around a dict often beats lock striping.
• YAGNI (You Aren't Gonna Need It): don't build extension points for requirements nobody mentioned. The balance in LLD is to leave a seam (an interface with one implementation) for the extensions you can name, without building the extensions themselves.

Law of Demeter

"Talk only to your immediate friends": a method should call methods on itself, its fields, its parameters and objects it creates, not on objects returned by those. Wikipedia: LoD Chains like a.b().c().d() couple you to the whole object graph.

# Violates Law of Demeter: caller reaches through three objects
total = order.customer.wallet.balance - order.total()

# Better: ask the object that owns the data to do the work
order.customer.can_afford(order.total())

Fluent builders (builder.method().header().build()) are not violations: every call returns the same object.

Cohesion and coupling

High cohesion: the things in a class belong together and change together. Low coupling: classes know little about each other, ideally only an interface. They tend to move together: when you split a low-cohesion class along its responsibilities, the pieces usually need fewer dependencies. A quick test in the round: "if the pricing rules change, how many classes do I touch?" One is the right answer.

Designing for testability

The code is testable if you can control its inputs, including the hidden ones: time, randomness, IDs, and anything over the network. The techniques are simple:

• Constructor injection of collaborators (gateway, repository, notifier) behind interfaces.
• Inject a clock (Callable[[], float] or a Clock object) instead of calling time.time() in domain code. All the rate-limiter, booking and parking-lot code on this page does this.
• Inject an RNG or ID generator where outputs depend on them (dice, shuffles, UUIDs in assertions).
• Pure functions for calculations (fee, split, score), which can be tested without any setup.
• Keep I/O at the edges: parse input → call domain → format output. The domain never prints or reads files.

Immutability and value objects

A value object is defined by its value, not its identity: Money(10, USD) equals any other Money(10, USD). Make them immutable (@dataclass(frozen=True)), validate in __post_init__, and return new instances from operations. They can be shared across threads with no locks, used as dict keys, and never end up half-updated. Fowler: ValueObject Entities (a Booking, a User) have identity and a lifecycle; they're usually mutable and need protecting.

from dataclasses import dataclass
from decimal import Decimal
from enum import Enum


class Currency(Enum):
    USD = "USD"
    INR = "INR"


@dataclass(frozen=True, slots=True)
class Money:
    amount: Decimal
    currency: Currency

    def __post_init__(self) -> None:
        if self.amount < 0:
            raise ValueError("Money cannot be negative")

    def __add__(self, other: "Money") -> "Money":
        if other.currency != self.currency:
            raise ValueError("currency mismatch")
        return Money(self.amount + other.amount, self.currency)


a = Money(Decimal("10.50"), Currency.USD)
b = Money(Decimal("2.25"), Currency.USD)
assert a + b == Money(Decimal("12.75"), Currency.USD)   # equality by value
assert len({a, Money(Decimal("10.50"), Currency.USD)}) == 1   # hashable

Error handling and exceptions

• Define a small hierarchy: one base DomainError for expected business failures (seat taken, insufficient funds), separate from programming errors (TypeError, failed assertions).
• Fail fast on invalid input at the boundary. Raise specific types that carry data (seat_id), not bare Exception("error").
• Don't use exceptions for normal control flow in hot paths. allow() on a rate limiter returns a bool; that's an expected outcome, not an error.
• Never swallow exceptions silently. If you catch broadly (a subscriber callback, a logging sink), log it and say why you're isolating it.
• Leave state consistent when you raise: validate everything first, then mutate (see make_change in the vending machine, which rolls back).

class DomainError(Exception):
    """Base for all expected, business-rule failures."""


class SeatUnavailable(DomainError):
    def __init__(self, seat_id: str) -> None:
        super().__init__(f"seat {seat_id} is not available")
        self.seat_id = seat_id


class PaymentDeclined(DomainError):
    pass


def book(seat_id: str, available: set[str]) -> str:
    if seat_id not in available:
        raise SeatUnavailable(seat_id)        # fail fast, specific type
    available.remove(seat_id)
    return f"booking-for-{seat_id}"


try:
    book("A1", {"A2"})
except SeatUnavailable as e:
    assert e.seat_id == "A1"

Enums and explicit state

Closed sets (vehicle types, order statuses, directions) should be Enums, not strings: typos become errors, and the IDE can list every value. For lifecycle state, combine an enum with an explicit transition table. That gives you a state machine you can draw, test and explain. When behaviour per state grows large, move to the State pattern (§4).

from enum import Enum, auto


class OrderStatus(Enum):
    CREATED = auto()
    PAID = auto()
    SHIPPED = auto()
    DELIVERED = auto()
    CANCELLED = auto()


ALLOWED: dict[OrderStatus, set[OrderStatus]] = {
    OrderStatus.CREATED: {OrderStatus.PAID, OrderStatus.CANCELLED},
    OrderStatus.PAID: {OrderStatus.SHIPPED, OrderStatus.CANCELLED},
    OrderStatus.SHIPPED: {OrderStatus.DELIVERED},
    OrderStatus.DELIVERED: set(),
    OrderStatus.CANCELLED: set(),
}


class Order:
    def __init__(self) -> None:
        self.status = OrderStatus.CREATED

    def transition(self, to: OrderStatus) -> None:
        if to not in ALLOWED[self.status]:
            raise ValueError(f"illegal transition {self.status.name} -> {to.name}")
        self.status = to


o = Order()
o.transition(OrderStatus.PAID)
try:
    o.transition(OrderStatus.DELIVERED)
except ValueError:
    pass

Mapping to Java

Many interviewers, especially for backend roles, expect Java-flavoured OOD vocabulary even if you code in Python. The mapping:

import java.util.EnumMap;
import java.util.Map;

// Interface = Python ABC/Protocol; enum with fields; constructor injection.
interface PricingStrategy {
    long fee(VehicleType type, long minutes);
}

enum VehicleType { MOTORCYCLE, CAR, TRUCK }

final class HourlyPricing implements PricingStrategy {
    private final Map<VehicleType, Long> rates = new EnumMap<>(VehicleType.class);

    HourlyPricing() {
        rates.put(VehicleType.MOTORCYCLE, 10L);
        rates.put(VehicleType.CAR, 20L);
        rates.put(VehicleType.TRUCK, 40L);
    }

    @Override
    public long fee(VehicleType type, long minutes) {
        long hours = Math.max(1, (minutes + 59) / 60);
        return hours * rates.get(type);
    }
}

// Java 16+ record = Python @dataclass(frozen=True)
record Ticket(long id, String plate, String spotId) {}

public class C1Strategy {
    public static void main(String[] args) {
        PricingStrategy p = new HourlyPricing();
        if (p.fee(VehicleType.CAR, 125) != 60) throw new AssertionError();
        System.out.println(new Ticket(1, "KA-01", "F0-M1"));
    }
}

4. The design patterns that actually come up

The 1994 "Gang of Four" book catalogued 23 patterns in three groups: creational, structural, behavioural. GoF 1994 Around a dozen of them, plus two from enterprise design (Repository, Specification), account for nearly everything you'll use in an LLD round. Each entry below gives the intent, a minimal runnable Python example, and the problems where it shows up. refactoring.guru has an illustrated page for each GoF pattern. refactoring.guru

Strategy

Intent: define a family of interchangeable algorithms behind one interface and choose one at runtime. refactoring.guru

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass(frozen=True)
class Cart:
    subtotal: float
    is_member: bool = False


class PricingStrategy(ABC):
    @abstractmethod
    def price(self, cart: Cart) -> float: ...


class RegularPricing(PricingStrategy):
    def price(self, cart: Cart) -> float:
        return cart.subtotal


class PercentOff(PricingStrategy):
    def __init__(self, pct: float) -> None:
        self.pct = pct

    def price(self, cart: Cart) -> float:
        return round(cart.subtotal * (1 - self.pct / 100), 2)


class Checkout:
    def __init__(self, strategy: PricingStrategy) -> None:
        self.strategy = strategy            # injected; swappable

    def total(self, cart: Cart) -> float:
        return self.strategy.price(cart)


assert Checkout(PercentOff(10)).total(Cart(200)) == 180.0

Shows up in: parking-lot pricing and spot allocation, elevator dispatch, rate-limiting algorithms, split types in Splitwise, cache eviction policy, payment methods, LLM provider selection, memory strategies. In Python a strategy can just be a function (Callable[[Cart], float]); use a class when it has configuration or several methods.

Factory Method and Abstract Factory

Intent: centralise object creation so callers ask for "a notifier for SMS" without naming the concrete class. A simple factory / registry (below) is what you'll usually write. The textbook Factory Method lets subclasses decide which class to instantiate. refactoring.guru

from abc import ABC, abstractmethod
from enum import Enum


class Channel(Enum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"


class Notifier(ABC):
    @abstractmethod
    def send(self, to: str, body: str) -> str: ...


class EmailNotifier(Notifier):
    def send(self, to: str, body: str) -> str: return f"email to {to}: {body}"


class SmsNotifier(Notifier):
    def send(self, to: str, body: str) -> str: return f"sms to {to}: {body}"


class PushNotifier(Notifier):
    def send(self, to: str, body: str) -> str: return f"push to {to}: {body}"


class NotifierFactory:
    _registry: dict[Channel, type[Notifier]] = {
        Channel.EMAIL: EmailNotifier,
        Channel.SMS: SmsNotifier,
        Channel.PUSH: PushNotifier,
    }

    @classmethod
    def register(cls, channel: Channel, impl: type[Notifier]) -> None:
        cls._registry[channel] = impl       # open for extension without editing create()

    @classmethod
    def create(cls, channel: Channel) -> Notifier:
        try:
            return cls._registry[channel]()
        except KeyError:
            raise ValueError(f"unsupported channel {channel}") from None


assert NotifierFactory.create(Channel.SMS).send("+1", "hi").startswith("sms")

Abstract Factory creates families of related objects that must be used together (dark-theme button + dark-theme checkbox; AWS queue + AWS blob store). refactoring.guru

from abc import ABC, abstractmethod


class Button(ABC):
    @abstractmethod
    def render(self) -> str: ...


class Checkbox(ABC):
    @abstractmethod
    def render(self) -> str: ...


class UIFactory(ABC):                       # creates a *family* of related products
    @abstractmethod
    def button(self) -> Button: ...
    @abstractmethod
    def checkbox(self) -> Checkbox: ...


class DarkButton(Button):
    def render(self) -> str: return "[dark button]"


class DarkCheckbox(Checkbox):
    def render(self) -> str: return "[dark checkbox]"


class DarkFactory(UIFactory):
    def button(self) -> Button: return DarkButton()
    def checkbox(self) -> Checkbox: return DarkCheckbox()


def build_form(f: UIFactory) -> list[str]:  # client never names a concrete class
    return [f.button().render(), f.checkbox().render()]


assert build_form(DarkFactory()) == ["[dark button]", "[dark checkbox]"]

Shows up in: notification service (channel → notifier), vehicle/spot creation, parsing commands in machine-coding rounds (command string → Command object), choosing an LLM provider adapter from config, chess piece creation.

Builder

Intent: construct a complex object step by step, validate once at build(), and produce an immutable result. refactoring.guru

from dataclasses import dataclass, field


@dataclass(frozen=True)
class HttpRequest:
    method: str
    url: str
    headers: dict[str, str]
    body: bytes | None
    timeout_s: float


class HttpRequestBuilder:
    def __init__(self, url: str) -> None:
        self._url = url
        self._method = "GET"
        self._headers: dict[str, str] = {}
        self._body: bytes | None = None
        self._timeout = 30.0

    def method(self, m: str) -> "HttpRequestBuilder":
        self._method = m.upper()
        return self

    def header(self, k: str, v: str) -> "HttpRequestBuilder":
        self._headers[k] = v
        return self

    def json(self, payload: str) -> "HttpRequestBuilder":
        self._body = payload.encode()
        return self.header("Content-Type", "application/json")

    def timeout(self, s: float) -> "HttpRequestBuilder":
        self._timeout = s
        return self

    def build(self) -> HttpRequest:
        if self._method == "GET" and self._body is not None:
            raise ValueError("GET with body")   # validate the whole object once
        return HttpRequest(self._method, self._url, dict(self._headers), self._body, self._timeout)


req = HttpRequestBuilder("https://api.example.com/v1").method("post").json('{"a":1}').timeout(5).build()
assert req.method == "POST" and req.headers["Content-Type"] == "application/json"

Shows up in: HTTP/LLM request objects, query builders, pizza/burger ordering prompts, configuring a game board. In Python, keyword arguments with defaults often make Builder unnecessary. Use it when construction has ordering or cross-field validation rules, or when the interviewer expects Java style.

Singleton (and why it's usually a smell)

Intent: ensure one instance and provide global access to it. refactoring.guru

import threading


class Config:
    _instance: "Config | None" = None
    _lock = threading.Lock()

    def __new__(cls) -> "Config":
        if cls._instance is None:               # fast path, no lock
            with cls._lock:
                if cls._instance is None:       # double-checked inside the lock
                    inst = super().__new__(cls)
                    inst.settings = {}
                    cls._instance = inst
        return cls._instance


assert Config() is Config()

# Usually better in Python: a module-level instance (modules are imported once),
# or simply construct one object at the composition root and inject it.

Why it's often a smell: it's global mutable state with a nicer name. It hides dependencies (nothing in a constructor signature says the class reads config), it makes tests share state and order-dependent, and it bakes "exactly one" into the class when "one per process" is really a wiring decision. The usual better answer is to create one instance at the composition root and inject it.

When it's acceptable: genuinely process-wide, stateless or read-mostly resources such as a logger, a metrics registry, or a connection pool, and even then prefer a module-level instance in Python.

Thread-safe variants: double-checked locking as above, an eagerly created module-level instance (Python modules are initialised once, under the import lock), and in Java the enum singleton or the initialization-on-demand holder idiom, which rely on the JVM's guarantees about class initialisation.

Observer / pub-sub

Intent: when one object changes, notify a list of dependents without coupling to them. refactoring.guru Pub-sub is the decoupled version, where publishers and subscribers only share a topic name on a bus.

from collections import defaultdict
from typing import Callable

Handler = Callable[[dict], None]


class EventBus:
    def __init__(self) -> None:
        self._subs: dict[str, list[Handler]] = defaultdict(list)

    def subscribe(self, topic: str, handler: Handler) -> Callable[[], None]:
        self._subs[topic].append(handler)
        return lambda: self._subs[topic].remove(handler)   # unsubscribe handle

    def publish(self, topic: str, event: dict) -> None:
        for h in list(self._subs[topic]):                   # copy: handlers may unsubscribe
            try:
                h(event)
            except Exception as exc:                        # one bad subscriber must not break others
                print(f"handler failed: {exc!r}")


bus = EventBus()
seen: list[dict] = []
unsub = bus.subscribe("order.paid", seen.append)
bus.publish("order.paid", {"id": 1})
unsub()
bus.publish("order.paid", {"id": 2})
assert seen == [{"id": 1}]

Shows up in: parking-lot display boards, stock tickers, notification on booking confirmed, auction bids, elevator floor displays, in-memory pub-sub (§7), agent event streams. Watch for: iterating a list that handlers modify, one failing handler breaking the rest, memory leaks from never unsubscribing, and synchronous handlers slowing the publisher (offload to a queue).

State

Intent: an object changes behaviour when its internal state changes. Each state is a class implementing the same interface, and transitions replace if state == … in every method. refactoring.guru

from abc import ABC, abstractmethod


class DocState(ABC):
    @abstractmethod
    def publish(self, doc: "Document") -> None: ...
    @abstractmethod
    def edit(self, doc: "Document", text: str) -> None: ...


class Draft(DocState):
    def publish(self, doc: "Document") -> None:
        doc.state = InReview()

    def edit(self, doc: "Document", text: str) -> None:
        doc.text = text


class InReview(DocState):
    def publish(self, doc: "Document") -> None:
        doc.state = Published()

    def edit(self, doc: "Document", text: str) -> None:
        raise PermissionError("cannot edit while in review")


class Published(DocState):
    def publish(self, doc: "Document") -> None:
        pass                                    # idempotent

    def edit(self, doc: "Document", text: str) -> None:
        doc.text, doc.state = text, Draft()     # editing a published doc reopens it


class Document:
    def __init__(self) -> None:
        self.text = ""
        self.state: DocState = Draft()

    def publish(self) -> None: self.state.publish(self)
    def edit(self, text: str) -> None: self.state.edit(self, text)


d = Document()
d.edit("v1"); d.publish()
assert isinstance(d.state, InReview)

Shows up in: vending machine, ATM, elevator (moving/idle/maintenance), order lifecycle, traffic light, document workflow, TCP connection. If there are few states and little per-state behaviour, an enum plus a transition table (§3) is simpler. Switch to State classes when every method has a big switch on the state.

Command (with undo)

Intent: wrap a request as an object, so you can queue it, log it, retry it, or undo it. refactoring.guru

from abc import ABC, abstractmethod


class Command(ABC):
    @abstractmethod
    def execute(self) -> None: ...
    @abstractmethod
    def undo(self) -> None: ...


class TextBuffer:
    def __init__(self) -> None:
        self.text = ""


class Insert(Command):
    def __init__(self, buf: TextBuffer, pos: int, s: str) -> None:
        self.buf, self.pos, self.s = buf, pos, s

    def execute(self) -> None:
        t = self.buf.text
        self.buf.text = t[: self.pos] + self.s + t[self.pos:]

    def undo(self) -> None:
        t = self.buf.text
        self.buf.text = t[: self.pos] + t[self.pos + len(self.s):]


class Editor:
    def __init__(self) -> None:
        self.buf = TextBuffer()
        self._undo: list[Command] = []
        self._redo: list[Command] = []

    def run(self, cmd: Command) -> None:
        cmd.execute()
        self._undo.append(cmd)
        self._redo.clear()                     # a new action invalidates the redo history

    def undo(self) -> None:
        if self._undo:
            cmd = self._undo.pop(); cmd.undo(); self._redo.append(cmd)

    def redo(self) -> None:
        if self._redo:
            cmd = self._redo.pop(); cmd.execute(); self._undo.append(cmd)


e = Editor()
e.run(Insert(e.buf, 0, "hello"))
e.run(Insert(e.buf, 5, " world"))
e.undo()
assert e.buf.text == "hello"
e.redo()
assert e.buf.text == "hello world"

Shows up in: text editor undo/redo, machine-coding CLIs (each input line → a Command), job queues, transaction logs, remote controls, LLM tool calls (§8). The undo stack plus "clear redo on a new action" is the detail interviewers check.

Decorator

Intent: add behaviour to an object by wrapping it in another object with the same interface. Decorators stack, and each does one thing. refactoring.guru

from abc import ABC, abstractmethod


class DataSource(ABC):
    @abstractmethod
    def read(self, key: str) -> str: ...


class Database(DataSource):
    def __init__(self) -> None:
        self.calls = 0

    def read(self, key: str) -> str:
        self.calls += 1
        return f"value:{key}"


class CachingSource(DataSource):              # same interface, wraps another DataSource
    def __init__(self, inner: DataSource) -> None:
        self._inner, self._cache = inner, {}

    def read(self, key: str) -> str:
        if key not in self._cache:
            self._cache[key] = self._inner.read(key)
        return self._cache[key]


class LoggingSource(DataSource):
    def __init__(self, inner: DataSource) -> None:
        self._inner = inner

    def read(self, key: str) -> str:
        print(f"read {key}")
        return self._inner.read(key)


db = Database()
src = LoggingSource(CachingSource(db))         # stack behaviours in any order
src.read("a"); src.read("a")
assert db.calls == 1

Shows up in: caching, logging, metrics, retries, rate limiting around a client; pizza/coffee toppings pricing; I/O streams. The LLM client in §8 is a stack of decorators. Don't confuse this with Python's @decorator syntax, which wraps functions. The idea is the same, but the GoF pattern wraps objects behind an interface.

Adapter

Intent: make an existing class with the wrong interface usable where another interface is expected. refactoring.guru

from typing import Protocol


class PaymentGateway(Protocol):               # the interface our code expects
    def charge(self, user_id: str, amount_cents: int) -> str: ...


class LegacyStripeLikeClient:                 # third-party shape we can't change
    def create_charge(self, payload: dict) -> dict:
        return {"id": "ch_123", "status": "succeeded", **payload}


class LegacyClientAdapter:
    def __init__(self, client: LegacyStripeLikeClient) -> None:
        self._c = client

    def charge(self, user_id: str, amount_cents: int) -> str:
        resp = self._c.create_charge({"customer": user_id, "amount": amount_cents})
        if resp["status"] != "succeeded":
            raise RuntimeError("charge failed")
        return resp["id"]


gw: PaymentGateway = LegacyClientAdapter(LegacyStripeLikeClient())
assert gw.charge("u1", 500) == "ch_123"

Shows up in: payment gateways, third-party SMS/email providers, legacy systems, LLM vendor SDKs behind a common LLMProvider, vector-database backends.

Facade

Intent: one simple interface over a complicated subsystem. refactoring.guru

class Inventory:
    def reserve(self, sku: str) -> bool: return True


class Payments:
    def charge(self, user: str, cents: int) -> str: return "pay_1"


class Shipping:
    def schedule(self, sku: str, user: str) -> str: return "ship_1"


class OrderFacade:                            # one simple entry point over a subsystem
    def __init__(self, inv: Inventory, pay: Payments, ship: Shipping) -> None:
        self.inv, self.pay, self.ship = inv, pay, ship

    def place_order(self, user: str, sku: str, cents: int) -> dict[str, str]:
        if not self.inv.reserve(sku):
            raise RuntimeError("out of stock")
        return {"payment": self.pay.charge(user, cents), "shipment": self.ship.schedule(sku, user)}


assert OrderFacade(Inventory(), Payments(), Shipping()).place_order("u", "sku", 100)["payment"] == "pay_1"

Shows up in: the top-level service of almost every LLD answer (BookingService, ParkingLot, Retriever). Adapter changes an interface; Facade simplifies many.

Chain of Responsibility

Intent: pass a request along a chain of handlers; each either handles it, rejects it, or passes it on. refactoring.guru

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass
class Request:
    user: str | None
    body: str
    ip: str


class Handler(ABC):
    def __init__(self) -> None:
        self._next: "Handler | None" = None

    def then(self, nxt: "Handler") -> "Handler":
        self._next = nxt
        return nxt                            # allows a.then(b).then(c)

    def handle(self, req: Request) -> str:
        return self._next.handle(req) if self._next else "OK"


class AuthHandler(Handler):
    def handle(self, req: Request) -> str:
        return "401" if req.user is None else super().handle(req)


class SizeLimitHandler(Handler):
    def handle(self, req: Request) -> str:
        return "413" if len(req.body) > 1000 else super().handle(req)


class BlocklistHandler(Handler):
    def __init__(self, blocked: set[str]) -> None:
        super().__init__()
        self.blocked = blocked

    def handle(self, req: Request) -> str:
        return "403" if req.ip in self.blocked else super().handle(req)


head = BlocklistHandler({"6.6.6.6"})
head.then(AuthHandler()).then(SizeLimitHandler())
assert head.handle(Request("u", "x", "1.1.1.1")) == "OK"
assert head.handle(Request(None, "x", "1.1.1.1")) == "401"

Shows up in: middleware pipelines (auth → rate limit → validation), ATM cash dispensing (500s, then 200s, then 100s), logger level handling, approval workflows (manager → director → VP), LLM provider fallback, guardrail pipelines.

Template Method

Intent: a base class fixes the skeleton of an algorithm; subclasses fill in the steps. refactoring.guru

from abc import ABC, abstractmethod


class DataExporter(ABC):
    def export(self, rows: list[dict]) -> str:      # the fixed algorithm skeleton
        rows = self.filter(rows)
        out = self.header(rows) + "".join(self.format_row(r) for r in rows)
        return out

    def filter(self, rows: list[dict]) -> list[dict]:   # hook with a default
        return rows

    @abstractmethod
    def header(self, rows: list[dict]) -> str: ...

    @abstractmethod
    def format_row(self, row: dict) -> str: ...


class CsvExporter(DataExporter):
    def header(self, rows: list[dict]) -> str:
        return ",".join(rows[0]) + "\n" if rows else ""

    def format_row(self, row: dict) -> str:
        return ",".join(str(v) for v in row.values()) + "\n"


assert CsvExporter().export([{"a": 1, "b": 2}]) == "a,b\n1,2\n"

Shows up in: game loops (init → while not over: take_turn → announce_winner), data exporters and parsers, test fixtures, report generators. It's inheritance-based, so prefer Strategy when the steps vary independently.

Composite

Intent: treat individual objects and groups of objects uniformly through a tree. refactoring.guru

from abc import ABC, abstractmethod


class FsNode(ABC):
    def __init__(self, name: str) -> None:
        self.name = name

    @abstractmethod
    def size(self) -> int: ...


class File(FsNode):
    def __init__(self, name: str, size: int) -> None:
        super().__init__(name)
        self._size = size

    def size(self) -> int:
        return self._size


class Directory(FsNode):
    def __init__(self, name: str) -> None:
        super().__init__(name)
        self.children: list[FsNode] = []

    def add(self, node: FsNode) -> "Directory":
        self.children.append(node)
        return self

    def size(self) -> int:                    # treats leaves and composites uniformly
        return sum(c.size() for c in self.children)


root = Directory("/").add(File("a.txt", 10)).add(Directory("src").add(File("m.py", 32)))
assert root.size() == 42

Shows up in: file systems, org charts, menus, UI component trees, nested expense groups, boolean filter expressions (the Specification below is a composite).

Iterator

Intent: traverse a collection without exposing how it's stored, possibly in several orders. refactoring.guru In Python this is built in: implement __iter__ or write a generator.

from collections.abc import Iterator


class Playlist:
    def __init__(self, songs: list[str]) -> None:
        self._songs = songs

    def __iter__(self) -> Iterator[str]:      # default order
        return iter(self._songs)

    def shuffled(self, seed: int) -> Iterator[str]:   # alternative traversal, same collection
        import random
        order = list(range(len(self._songs)))
        random.Random(seed).shuffle(order)
        for i in order:
            yield self._songs[i]


p = Playlist(["a", "b", "c"])
assert list(p) == ["a", "b", "c"]
assert sorted(p.shuffled(seed=1)) == ["a", "b", "c"]

Shows up in: playlists, paginated APIs, tree traversal (BFS/DFS iterators), streaming LLM tokens.

Proxy

Intent: a stand-in with the same interface that controls access to the real object: lazy creation (virtual proxy), permission checks (protection proxy), remote calls (remote proxy), or caching. refactoring.guru

from typing import Protocol


class Image(Protocol):
    def display(self) -> str: ...


class RealImage:
    def __init__(self, path: str) -> None:
        self.path = path
        self.pixels = f"<decoded {path}>"     # imagine an expensive load here

    def display(self) -> str:
        return self.pixels


class LazyImageProxy:                         # virtual proxy: same interface, defers creation
    def __init__(self, path: str, user_role: str) -> None:
        self.path, self.role = path, user_role
        self._real: RealImage | None = None

    def display(self) -> str:
        if self.role != "viewer" and self.role != "admin":
            raise PermissionError("protection proxy: access denied")
        if self._real is None:
            self._real = RealImage(self.path)
        return self._real.display()


img = LazyImageProxy("cat.png", "viewer")
assert img._real is None
assert img.display() == "<decoded cat.png>"

Proxy vs Decorator: structurally identical. A proxy controls access to the subject and often manages its lifecycle; a decorator adds behaviour and is designed to stack.

Repository

Intent: a collection-like interface for loading and saving domain objects, hiding the storage technology. Fowler: Repository

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass
class User:
    id: str
    email: str


class UserRepository(ABC):                    # domain-facing collection-like interface
    @abstractmethod
    def get(self, user_id: str) -> User | None: ...
    @abstractmethod
    def add(self, user: User) -> None: ...
    @abstractmethod
    def find_by_email(self, email: str) -> User | None: ...


class InMemoryUserRepository(UserRepository):
    def __init__(self) -> None:
        self._rows: dict[str, User] = {}

    def get(self, user_id: str) -> User | None:
        return self._rows.get(user_id)

    def add(self, user: User) -> None:
        self._rows[user.id] = user

    def find_by_email(self, email: str) -> User | None:
        return next((u for u in self._rows.values() if u.email == email), None)


# A PostgresUserRepository would implement the same interface; services never know which.
repo = InMemoryUserRepository()
repo.add(User("1", "a@x.com"))
assert repo.find_by_email("a@x.com").id == "1"

Shows up in: any problem where the interviewer says "assume a database". Write an in-memory repository behind an interface, mention that a SQL one would implement the same interface, and move on. It also gives you a fake for tests for free.

Specification

Intent: encapsulate a business rule as a composable predicate object (and/or/not), so filters can be combined without writing a method per combination. Described by Evans and Fowler. Evans & Fowler

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass(frozen=True)
class Product:
    name: str
    price: float
    color: str
    in_stock: bool


class Spec(ABC):
    @abstractmethod
    def ok(self, p: Product) -> bool: ...

    def __and__(self, other: "Spec") -> "Spec": return _And(self, other)
    def __or__(self, other: "Spec") -> "Spec": return _Or(self, other)
    def __invert__(self) -> "Spec": return _Not(self)


class _And(Spec):
    def __init__(self, a: Spec, b: Spec) -> None: self.a, self.b = a, b
    def ok(self, p: Product) -> bool: return self.a.ok(p) and self.b.ok(p)


class _Or(Spec):
    def __init__(self, a: Spec, b: Spec) -> None: self.a, self.b = a, b
    def ok(self, p: Product) -> bool: return self.a.ok(p) or self.b.ok(p)


class _Not(Spec):
    def __init__(self, a: Spec) -> None: self.a = a
    def ok(self, p: Product) -> bool: return not self.a.ok(p)


class ColorIs(Spec):
    def __init__(self, c: str) -> None: self.c = c
    def ok(self, p: Product) -> bool: return p.color == self.c


class PriceBelow(Spec):
    def __init__(self, x: float) -> None: self.x = x
    def ok(self, p: Product) -> bool: return p.price < self.x


class InStock(Spec):
    def ok(self, p: Product) -> bool: return p.in_stock


catalog = [Product("a", 10, "red", True), Product("b", 50, "red", True), Product("c", 5, "blue", False)]
spec = ColorIs("red") & PriceBelow(20) & InStock()
assert [p.name for p in catalog if spec.ok(p)] == ["a"]

Shows up in: product search filters, eligibility rules (who gets a discount), vector-store metadata filters (§8), parking-spot matching rules, alert conditions.

Pattern → problem cheat sheet

5. UML essentials

You need about ten pieces of UML notation, and interviewers rarely care whether your arrowheads match the spec exactly. What they want is to see that you distinguish "owns" from "uses" from "is-a". uml-diagrams.org The formal specification is published by the OMG. OMG UML

Class diagram notation

Sequence diagrams

A sequence diagram shows objects as columns (lifelines) and messages as arrows going down in time order. Solid arrows are calls, dashed arrows are returns, and boxes labelled alt/opt/loop show branches and repetition. uml-diagrams.org In an interview, ASCII is fine. Here is the "book seats" flow from §7:

6. Concurrency for LLD

Almost every LLD problem ends with "now make it thread-safe". The method is always the same: (1) list the shared mutable state, (2) write down the invariants that span it ("a seat is in at most one hold", "free + occupied = total"), (3) protect each invariant with one lock, or remove the sharing (immutability, confinement, queues), (4) check for deadlock and for slow work inside locks.

Race conditions and locks

A race condition happens when the result depends on how threads interleave. The classic case is read-modify-write: two threads read value = 5, both write 6, and one increment is lost. Check-then-act is the other common shape: if seat.free: seat.book() lets two threads both see "free". The fix is to make the whole compound operation a critical section under a mutex. Python docs: threading

import threading


class Counter:
    def __init__(self) -> None:
        self.value = 0
        self._lock = threading.Lock()

    def unsafe_incr(self) -> None:
        v = self.value          # read
        v += 1                  # modify
        self.value = v          # write: another thread may have written in between

    def safe_incr(self) -> None:
        with self._lock:        # read-modify-write is now one critical section
            self.value += 1


c = Counter()
threads = [threading.Thread(target=lambda: [c.safe_incr() for _ in range(10_000)]) for _ in range(8)]
for t in threads: t.start()
for t in threads: t.join()
assert c.value == 80_000

RLock (re-entrant lock)

A plain Lock deadlocks if the thread holding it tries to take it again. An RLock counts acquisitions by the owning thread. It's useful when a locked public method calls another locked public method. It's often a sign you should split out a private, unlocked helper, but it's the pragmatic answer in an interview.

import threading


class Account:
    def __init__(self, balance: int) -> None:
        self.balance = balance
        self._lock = threading.RLock()     # re-entrant: the same thread may acquire it again

    def withdraw(self, amt: int) -> None:
        with self._lock:
            if amt > self.balance:
                raise ValueError("insufficient funds")
            self.balance -= amt

    def withdraw_with_fee(self, amt: int, fee: int) -> None:
        with self._lock:                   # holds the lock...
            self.withdraw(amt)             # ...and calls a method that takes it again
            self.withdraw(fee)             # with a plain Lock this would deadlock


a = Account(100)
a.withdraw_with_fee(50, 1)
assert a.balance == 49

Condition variables and producer–consumer

A condition variable lets threads sleep until some predicate on shared state becomes true, releasing the lock while they wait. Two rules: always wait in a while loop that re-checks the predicate (spurious wakeups, and another thread may get there first), and notify after changing the state the predicate depends on.

import threading
from collections import deque
from typing import Generic, TypeVar

T = TypeVar("T")


class BoundedBuffer(Generic[T]):
    """Classic producer-consumer with one lock and two condition variables."""

    def __init__(self, capacity: int) -> None:
        self._items: deque[T] = deque()
        self._cap = capacity
        lock = threading.Lock()
        self._not_full = threading.Condition(lock)
        self._not_empty = threading.Condition(lock)

    def put(self, item: T) -> None:
        with self._not_full:
            while len(self._items) >= self._cap:     # always re-check in a while loop
                self._not_full.wait()
            self._items.append(item)
            self._not_empty.notify()

    def get(self) -> T:
        with self._not_empty:
            while not self._items:
                self._not_empty.wait()
            item = self._items.popleft()
            self._not_full.notify()
            return item


buf: BoundedBuffer[int] = BoundedBuffer(2)
out: list[int] = []
consumer = threading.Thread(target=lambda: [out.append(buf.get()) for _ in range(5)])
consumer.start()
for i in range(5):
    buf.put(i)
consumer.join()
assert out == [0, 1, 2, 3, 4]

In real Python code you'd use queue.Queue, which is exactly this bounded buffer, already thread-safe. A bounded queue gives you back-pressure: producers block when consumers fall behind instead of using unbounded memory. Python docs: queue

import queue
import threading

SENTINEL = object()


def worker(q: "queue.Queue[object]", results: list[int], lock: threading.Lock) -> None:
    while True:
        job = q.get()
        try:
            if job is SENTINEL:
                return
            with lock:
                results.append(job * job)      # list.append is atomic in CPython, but don't rely on it
        finally:
            q.task_done()


q: "queue.Queue[object]" = queue.Queue(maxsize=100)   # bounded = back-pressure on producers
results: list[int] = []
lock = threading.Lock()
pool = [threading.Thread(target=worker, args=(q, results, lock)) for _ in range(4)]
for t in pool: t.start()
for n in range(20): q.put(n)
for _ in pool: q.put(SENTINEL)                       # one poison pill per worker
q.join()
for t in pool: t.join()
assert sorted(results) == [n * n for n in range(20)]

Semaphores

A semaphore is a counter of permits: acquire takes one (blocking at zero), release returns one. It's the tool for "at most N concurrent X": connection pools, concurrent LLM calls per provider, parking-lot gate capacity. BoundedSemaphore raises if you release more than you acquired, which catches bugs.

import threading
import time


class ConnectionPool:
    def __init__(self, size: int) -> None:
        self._sem = threading.BoundedSemaphore(size)   # at most `size` concurrent holders
        self.max_seen = 0
        self._active = 0
        self._lock = threading.Lock()

    def query(self) -> None:
        with self._sem:
            with self._lock:
                self._active += 1
                self.max_seen = max(self.max_seen, self._active)
            time.sleep(0.01)
            with self._lock:
                self._active -= 1


pool = ConnectionPool(3)
ts = [threading.Thread(target=pool.query) for _ in range(10)]
for t in ts: t.start()
for t in ts: t.join()
assert pool.max_seen <= 3

Read-write locks

When reads vastly outnumber writes (a config cache, a catalogue), a read-write lock lets many readers in at once but gives writers exclusive access. Python's standard library doesn't include one; here's a writer-preferring version built on a condition variable. Java has ReentrantReadWriteLock. Java docs

import threading
from contextlib import contextmanager
from collections.abc import Iterator


class RWLock:
    """Many readers OR one writer. Writer-preferring so writers don't starve."""

    def __init__(self) -> None:
        self._cond = threading.Condition()
        self._readers = 0
        self._writer = False
        self._waiting_writers = 0

    @contextmanager
    def read(self) -> Iterator[None]:
        with self._cond:
            while self._writer or self._waiting_writers:
                self._cond.wait()
            self._readers += 1
        try:
            yield
        finally:
            with self._cond:
                self._readers -= 1
                if self._readers == 0:
                    self._cond.notify_all()

    @contextmanager
    def write(self) -> Iterator[None]:
        with self._cond:
            self._waiting_writers += 1
            while self._writer or self._readers:
                self._cond.wait()
            self._waiting_writers -= 1
            self._writer = True
        try:
            yield
        finally:
            with self._cond:
                self._writer = False
                self._cond.notify_all()


class Catalog:
    def __init__(self) -> None:
        self._data: dict[str, int] = {}
        self._rw = RWLock()

    def get(self, k: str) -> int | None:
        with self._rw.read():
            return self._data.get(k)

    def put(self, k: str, v: int) -> None:
        with self._rw.write():
            self._data[k] = v


cat = Catalog()
cat.put("x", 1)
assert cat.get("x") == 1

Starvation: a reader-preferring lock can starve writers under constant reads. Writer preference (above) fixes that but can starve readers under constant writes. Say which you chose and why.

Atomic operations and thread-safe collections

• Java has real atomics (AtomicInteger.incrementAndGet(), compareAndSet) and concurrent collections (ConcurrentHashMap with atomic computeIfAbsent/merge, ConcurrentLinkedQueue, BlockingQueue). Java docs: ConcurrentHashMap
• Python has no general atomic-integer type in the standard library. Individual operations on built-in types (such as list.append or dict[k] = v) happen to be atomic in CPython because of the GIL, but compound operations (d[k] += 1, check-then-set) are not, and relying on the implementation detail is fragile. Use a lock, or queue.Queue to hand data between threads. collections.deque's append/popleft are documented as thread-safe. Python docs: collections
• Lock striping: instead of one lock for a whole map, use N locks and pick one by hash(key) % N. This is roughly how older versions of ConcurrentHashMap worked. Mention it as the next step when one global lock becomes a bottleneck.

Deadlock

Deadlock needs all four Coffman conditions at once: mutual exclusion (resources can't be shared), hold and wait (holding one lock while waiting for another), no preemption (locks can't be taken away), and circular wait (A waits for B, B waits for A). Wikipedia: Deadlock Break any one to prevent it. In practice:

• Lock ordering (breaks circular wait): always acquire locks in a global order, such as by account ID. This is the standard answer for bank transfers and multi-seat booking across shows.
• Try-lock with timeout (breaks hold-and-wait): lock.acquire(timeout=…); on failure, release what you hold, back off, retry.
• One coarser lock: one lock per show instead of one per seat removes multi-lock acquisition entirely.
• Never call out while holding a lock: callbacks, network calls and other objects' locked methods are where hidden cycles come from.

import threading
from dataclasses import dataclass, field


@dataclass
class Account:
    id: int
    balance: int
    lock: threading.Lock = field(default_factory=threading.Lock, repr=False)


def transfer(a: Account, b: Account, amt: int) -> None:
    # Always lock in a global order (by id) so two opposite transfers can't deadlock.
    first, second = (a, b) if a.id < b.id else (b, a)
    with first.lock, second.lock:
        if a.balance < amt:
            raise ValueError("insufficient funds")
        a.balance -= amt
        b.balance += amt


x, y = Account(1, 1000), Account(2, 1000)
ts = [threading.Thread(target=transfer, args=(x, y, 1)) for _ in range(200)]
ts += [threading.Thread(target=transfer, args=(y, x, 1)) for _ in range(200)]
for t in ts: t.start()
for t in ts: t.join()
assert x.balance + y.balance == 2000

Optimistic vs pessimistic locking

import threading
from dataclasses import dataclass


class ConcurrentModification(Exception):
    pass


@dataclass
class Row:
    value: int
    version: int


class VersionedStore:
    """Simulates `UPDATE t SET value=?, version=version+1 WHERE id=? AND version=?`."""

    def __init__(self) -> None:
        self._rows: dict[str, Row] = {}
        self._lock = threading.Lock()          # stands in for the DB's atomic row update

    def read(self, key: str) -> Row:
        with self._lock:
            r = self._rows.setdefault(key, Row(0, 0))
            return Row(r.value, r.version)     # snapshot copy

    def compare_and_set(self, key: str, expected_version: int, new_value: int) -> None:
        with self._lock:
            r = self._rows[key]
            if r.version != expected_version:
                raise ConcurrentModification(key)
            r.value, r.version = new_value, r.version + 1


def increment_with_retry(store: VersionedStore, key: str, retries: int = 50) -> None:
    for _ in range(retries):
        snap = store.read(key)
        try:
            store.compare_and_set(key, snap.version, snap.value + 1)
            return
        except ConcurrentModification:
            continue                           # someone else won; re-read and retry
    raise RuntimeError("too much contention")


s = VersionedStore()
ts = [threading.Thread(target=increment_with_retry, args=(s, "k", 1000)) for _ in range(20)]
for t in ts: t.start()
for t in ts: t.join()
assert s.read("k").value == 20

Idempotency

Networks fail after the server has done the work but before the client sees the response, so clients retry. An operation is idempotent if doing it twice has the same effect as doing it once. For non-idempotent operations (charge a card, create a booking), the client sends an idempotency key; the server stores the result under that key and returns the stored result on a replay. Stripe's API is the well-known public example. Stripe docs

import threading
import uuid


class PaymentService:
    def __init__(self) -> None:
        self._results: dict[str, str] = {}
        self._lock = threading.Lock()
        self.charges = 0

    def charge(self, idempotency_key: str, user: str, cents: int) -> str:
        with self._lock:                       # in production: unique constraint on the key
            if idempotency_key in self._results:
                return self._results[idempotency_key]   # replay returns the original result
            self.charges += 1
            receipt = f"rcpt-{uuid.uuid4().hex[:8]}"
            self._results[idempotency_key] = receipt
            return receipt


svc = PaymentService()
key = "order-42-attempt"
assert svc.charge(key, "u", 500) == svc.charge(key, "u", 500)   # client retry is safe
assert svc.charges == 1

Python's GIL, threads, processes and asyncio

In the standard CPython build, the Global Interpreter Lock lets only one thread execute Python bytecode at a time. Threads still help with I/O-bound work (the GIL is released while waiting on sockets and files) but not with CPU-bound work, where you'd use multiprocessing or concurrent.futures.ProcessPoolExecutor. Python docs: concurrent.futures The GIL does not make your code thread-safe: a thread can be switched out between any two bytecodes, so compound operations still race.

PEP 703 introduced an optional free-threaded build without the GIL, available from Python 3.13 as a separate build. PEP 703 Python docs: free threading On it, data races that the GIL used to hide become much more likely, which is another reason to lock properly.

asyncio is single-threaded cooperative concurrency: coroutines yield control only at await. That removes most data races (no preemption between awaits), but you still need asyncio.Lock when a critical section contains an await. asyncio primitives are not thread-safe, and a blocking call inside a coroutine stalls the whole event loop. Python docs: asyncio

import asyncio


async def fetch(i: int, sem: asyncio.Semaphore) -> int:
    async with sem:                     # cap concurrency, e.g. per-provider connection limit
        await asyncio.sleep(0.01)       # I/O wait yields to the event loop
        return i * 2


async def main() -> list[int]:
    sem = asyncio.Semaphore(5)
    lock = asyncio.Lock()               # asyncio primitives are NOT thread-safe; use within one loop
    total = 0

    async def add(x: int) -> None:
        nonlocal total
        async with lock:                # needed only if there is an await inside the critical section
            total += x

    results = await asyncio.gather(*(fetch(i, sem) for i in range(20)))
    await asyncio.gather(*(add(r) for r in results))
    assert total == sum(results)
    return results


assert asyncio.run(main())[:3] == [0, 2, 4]

Java equivalents, briefly

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class C1Concurrency {
    // 1. synchronized method: intrinsic lock on `this`
    static class TokenBucket {
        private final double capacity, ratePerSec;
        private double tokens;
        private long lastNanos = System.nanoTime();

        TokenBucket(double capacity, double ratePerSec) {
            this.capacity = capacity; this.ratePerSec = ratePerSec; this.tokens = capacity;
        }

        synchronized boolean tryAcquire() {
            long now = System.nanoTime();
            tokens = Math.min(capacity, tokens + (now - lastNanos) / 1e9 * ratePerSec);
            lastNanos = now;
            if (tokens >= 1) { tokens -= 1; return true; }
            return false;
        }
    }

    // 2. ReentrantLock: explicit lock, tryLock with timeout, always unlock in finally
    static class Seat {
        private final ReentrantLock lock = new ReentrantLock();
        private String heldBy;

        boolean hold(String user) throws InterruptedException {
            if (!lock.tryLock(100, TimeUnit.MILLISECONDS)) return false;
            try {
                if (heldBy != null) return false;
                heldBy = user;
                return true;
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 3. ConcurrentHashMap + AtomicInteger: per-key counters without a global lock
        Map<String, AtomicInteger> hits = new ConcurrentHashMap<>();
        // 4. ExecutorService instead of raw threads
        ExecutorService pool = Executors.newFixedThreadPool(8);
        Seat seat = new Seat();
        AtomicInteger winners = new AtomicInteger();
        TokenBucket bucket = new TokenBucket(5, 0);
        AtomicInteger allowed = new AtomicInteger();
        for (int i = 0; i < 100; i++) {
            final String user = "u" + i;
            pool.submit(() -> {
                hits.computeIfAbsent("/api", k -> new AtomicInteger()).incrementAndGet();
                if (bucket.tryAcquire()) allowed.incrementAndGet();
                try { if (seat.hold(user)) winners.incrementAndGet(); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        }
        pool.shutdown();
        pool.awaitTermination(5, TimeUnit.SECONDS);
        if (hits.get("/api").get() != 100 || winners.get() != 1 || allowed.get() != 5)
            throw new AssertionError();
        System.out.println("ok");
    }
}

One Java-specific point interviewers like: without volatile or a lock, one thread's write may never become visible to another (the Java Memory Model's happens-before rules). Locks give you both mutual exclusion and visibility. Oracle Java tutorial: Concurrency

7. Worked problems

Each problem follows the same outline: requirements and clarifying questions, entities, a class diagram, the core code (runnable, with its own small test at the bottom), patterns used, concurrency, the extensions interviewers ask for, and common mistakes. The first four (parking lot, LRU, rate limiter, booking) get the most depth because they come up most often and contain the most traps.

7.1 Parking lot

Prompt: "Design a parking lot system."

Entities: ParkingLot (facade), Spot, Vehicle (value object), Ticket (value object), enums VehicleType and SpotSize, strategies SpotAllocationStrategy and PricingStrategy, and an injected Clock. Floor and Gate are often drawn too; here the floor is an attribute of the spot, which is enough for allocation.

from __future__ import annotations

import itertools
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from math import ceil


class VehicleType(Enum):
    MOTORCYCLE = 1
    CAR = 2
    TRUCK = 3


class SpotSize(Enum):
    SMALL = 1
    MEDIUM = 2
    LARGE = 3


# Which spot sizes each vehicle may use, smallest first (prefer the tightest fit).
FITS: dict[VehicleType, list[SpotSize]] = {
    VehicleType.MOTORCYCLE: [SpotSize.SMALL, SpotSize.MEDIUM, SpotSize.LARGE],
    VehicleType.CAR: [SpotSize.MEDIUM, SpotSize.LARGE],
    VehicleType.TRUCK: [SpotSize.LARGE],
}


@dataclass(frozen=True)
class Vehicle:
    plate: str
    type: VehicleType


@dataclass
class Spot:
    id: str
    floor: int
    size: SpotSize
    vehicle: Vehicle | None = None

    @property
    def free(self) -> bool:
        return self.vehicle is None


@dataclass(frozen=True)
class Ticket:
    id: int
    plate: str
    spot_id: str
    entered_at: datetime


class NoSpotAvailable(Exception):
    pass


# ---- Strategies -------------------------------------------------------------
class SpotAllocationStrategy(ABC):
    @abstractmethod
    def choose(self, candidates: list[Spot]) -> Spot | None: ...


class LowestFloorFirst(SpotAllocationStrategy):
    def choose(self, candidates: list[Spot]) -> Spot | None:
        return min(candidates, key=lambda s: (s.floor, s.id), default=None)


class PricingStrategy(ABC):
    @abstractmethod
    def fee(self, vehicle_type: VehicleType, duration: timedelta) -> int: ...


class HourlyPricing(PricingStrategy):
    RATES = {VehicleType.MOTORCYCLE: 10, VehicleType.CAR: 20, VehicleType.TRUCK: 40}

    def fee(self, vehicle_type: VehicleType, duration: timedelta) -> int:
        hours = max(1, ceil(duration.total_seconds() / 3600))   # first hour always charged
        return hours * self.RATES[vehicle_type]


class Clock(ABC):
    @abstractmethod
    def now(self) -> datetime: ...


class SystemClock(Clock):
    def now(self) -> datetime:
        return datetime.now()


# ---- The lot -------------------------------------------------------------------
class ParkingLot:
    def __init__(self, spots: list[Spot], allocator: SpotAllocationStrategy,
                 pricing: PricingStrategy, clock: Clock | None = None) -> None:
        # index free spots by size so allocation doesn't scan the whole lot
        self._free: dict[SpotSize, dict[str, Spot]] = {s: {} for s in SpotSize}
        self._spots = {s.id: s for s in spots}
        for s in spots:
            self._free[s.size][s.id] = s
        self._active: dict[int, Ticket] = {}
        self._plates: set[str] = set()
        self._allocator, self._pricing = allocator, pricing
        self._clock = clock or SystemClock()
        self._ids = itertools.count(1)
        self._lock = threading.Lock()          # one lock: simple and correct for one process

    def park(self, v: Vehicle) -> Ticket:
        with self._lock:
            if v.plate in self._plates:
                raise ValueError(f"{v.plate} is already parked")
            for size in FITS[v.type]:
                spot = self._allocator.choose(list(self._free[size].values()))
                if spot:
                    del self._free[size][spot.id]
                    spot.vehicle = v
                    t = Ticket(next(self._ids), v.plate, spot.id, self._clock.now())
                    self._active[t.id] = t
                    self._plates.add(v.plate)
                    return t
            raise NoSpotAvailable(v.type.name)

    def unpark(self, ticket_id: int) -> int:
        with self._lock:
            t = self._active.pop(ticket_id, None)
            if t is None:
                raise KeyError(f"unknown or already-used ticket {ticket_id}")
            spot = self._spots[t.spot_id]
            assert spot.vehicle is not None
            fee = self._pricing.fee(spot.vehicle.type, self._clock.now() - t.entered_at)
            self._plates.discard(spot.vehicle.plate)
            spot.vehicle = None
            self._free[spot.size][spot.id] = spot
            return fee

    def free_count(self, size: SpotSize) -> int:
        with self._lock:
            return len(self._free[size])


class FakeClock(Clock):
    def __init__(self, t: datetime) -> None:
        self.t = t

    def now(self) -> datetime:
        return self.t


clock = FakeClock(datetime(2026, 1, 1, 9, 0))
spots = [Spot("F0-S1", 0, SpotSize.SMALL), Spot("F0-M1", 0, SpotSize.MEDIUM), Spot("F1-L1", 1, SpotSize.LARGE)]
lot = ParkingLot(spots, LowestFloorFirst(), HourlyPricing(), clock)
t1 = lot.park(Vehicle("KA-01", VehicleType.CAR))
assert t1.spot_id == "F0-M1"
t2 = lot.park(Vehicle("KA-02", VehicleType.CAR))      # medium full -> falls back to large
assert t2.spot_id == "F1-L1"
clock.t += timedelta(hours=2, minutes=5)
assert lot.unpark(t1.id) == 60                         # 3 started hours x 20
try:
    lot.park(Vehicle("TR-1", VehicleType.TRUCK))
except NoSpotAvailable:
    pass

Design decisions worth saying out loud:

• Free spots are indexed by size, so park doesn't scan every spot. The allocator gets only the candidates. For thousands of spots per size, keep a heap keyed by (floor, id) per size to make allocation O(log n).
• The FITS table is data, not code: a motorcycle can take any spot, smallest first. A new vehicle type is an enum value plus a row.
• Pricing and allocation are strategies, so "weekend pricing" or "nearest to the exit" are new classes with no edits to ParkingLot.
• A duplicate-plate check stops the same car parking twice. A popped ticket stops a double unpark.
• The clock is injected, so fee tests are deterministic.

Patterns: Strategy (allocation, pricing), Facade (ParkingLot), value objects, optional Singleton (resist it), Observer for display boards, Factory for creating spots from a layout config.

Concurrency: several gates call park at once. The invariant spans the free index, the spot and the ticket map, so one lock around each operation is correct and simple. Allocation is pure in-memory work, so the critical section is microseconds long. If asked to scale: one lock per SpotSize (cars and trucks don't contend), but then park may need to try several sizes, and the duplicate-plate set needs its own lock. Take locks in a fixed order (size enum order) to avoid deadlock.

Extensions interviewers ask for:

• EV charging spots: add SpotFeature.EV_CHARGER (or a features: frozenset on Spot) and make the allocator filter by a Specification (needs_charger & size_fits). Add a charging fee to pricing via a Decorator: ChargingPricing(HourlyPricing()).
• Display boards per floor: Observer. ParkingLot publishes spot_freed/spot_taken; boards subscribe.
• Reservations: a spot gets a RESERVED state with a TTL (the same pattern as the seat hold in 7.5).
• Multiple lots / persistence: a SpotRepository interface; with a DB, use conditional updates (UPDATE spot SET vehicle=? WHERE id=? AND vehicle IS NULL).
• Payment: a PaymentProcessor interface with cash/card/UPI implementations; the exit gate only opens after payment succeeds.

7.2 LRU cache (and LFU)

Prompt: "Implement an LRU cache with O(1) get and put." This is half data-structures question, half LLD: the interviewer watches for clean encapsulation, generics, edge cases and thread safety as much as the algorithm.

from __future__ import annotations

import threading
from typing import Generic, Hashable, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")


class _Node(Generic[K, V]):
    __slots__ = ("key", "val", "prev", "next")

    def __init__(self, key: K | None = None, val: V | None = None) -> None:
        self.key, self.val = key, val
        self.prev: _Node[K, V] | None = None
        self.next: _Node[K, V] | None = None


class LRUCache(Generic[K, V]):
    """O(1) get/put: hash map for lookup + doubly linked list for recency.

    head <-> [most recent] <-> ... <-> [least recent] <-> tail   (sentinels)
    """

    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            raise ValueError("capacity must be positive")
        self._cap = capacity
        self._map: dict[K, _Node[K, V]] = {}
        self._head: _Node[K, V] = _Node()
        self._tail: _Node[K, V] = _Node()
        self._head.next, self._tail.prev = self._tail, self._head
        self._lock = threading.Lock()
        self.hits = self.misses = 0

    # -- linked-list helpers (caller holds the lock) --
    def _unlink(self, n: _Node[K, V]) -> None:
        assert n.prev and n.next
        n.prev.next, n.next.prev = n.next, n.prev

    def _push_front(self, n: _Node[K, V]) -> None:
        first = self._head.next
        assert first
        n.prev, n.next = self._head, first
        self._head.next = first.prev = n

    # -- public API --
    def get(self, key: K) -> V | None:
        with self._lock:
            n = self._map.get(key)
            if n is None:
                self.misses += 1
                return None
            self._unlink(n)
            self._push_front(n)            # a read counts as "use"
            self.hits += 1
            return n.val

    def put(self, key: K, val: V) -> None:
        with self._lock:
            n = self._map.get(key)
            if n:
                n.val = val
                self._unlink(n)
                self._push_front(n)
                return
            if len(self._map) >= self._cap:
                lru = self._tail.prev
                assert lru and lru is not self._head
                self._unlink(lru)
                del self._map[lru.key]      # why the node stores its key
            n = _Node(key, val)
            self._map[key] = n
            self._push_front(n)

    def __len__(self) -> int:
        return len(self._map)


c: LRUCache[str, int] = LRUCache(2)
c.put("a", 1); c.put("b", 2)
assert c.get("a") == 1          # a is now most recent
c.put("c", 3)                   # evicts b
assert c.get("b") is None and c.get("c") == 3 and len(c) == 2

If the interviewer allows library structures, OrderedDict gives you the same thing in a few lines: move_to_end and popitem(last=False) are both O(1). Python docs: collections Ask first, and be ready to write the linked-list version. (The standard library's functools.lru_cache memoises function calls; it isn't a general key-value cache object. Python docs: functools)

from collections import OrderedDict
from typing import Hashable


class LRU:
    def __init__(self, cap: int) -> None:
        self.cap, self.d = cap, OrderedDict()

    def get(self, k: Hashable):
        if k not in self.d:
            return None
        self.d.move_to_end(k)            # mark most-recent
        return self.d[k]

    def put(self, k: Hashable, v) -> None:
        self.d[k] = v
        self.d.move_to_end(k)
        if len(self.d) > self.cap:
            self.d.popitem(last=False)   # pop least-recent


c = LRU(1); c.put(1, 1); c.put(2, 2)
assert c.get(1) is None and c.get(2) == 2

In Java, the equivalent shortcut is LinkedHashMap with accessOrder=true and an overridden removeEldestEntry:

import java.util.LinkedHashMap;
import java.util.Map;

// accessOrder=true makes LinkedHashMap move entries to the tail on get();
// overriding removeEldestEntry turns it into an LRU cache.
class LruCache<K, V> {
    private final Map<K, V> map;

    LruCache(int capacity) {
        this.map = new LinkedHashMap<>(16, 0.75f, true) {
            @Override
            protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
                return size() > capacity;
            }
        };
    }

    public synchronized V get(K key) { return map.get(key); }   // get mutates order: needs the lock too
    public synchronized void put(K key, V value) { map.put(key, value); }
}

public class C1Lru {
    public static void main(String[] args) {
        LruCache<String, Integer> c = new LruCache<>(2);
        c.put("a", 1); c.put("b", 2); c.get("a"); c.put("c", 3);
        if (c.get("b") != null || c.get("a") != 1) throw new AssertionError();
        System.out.println("ok");
    }
}

Extension: LFU

Least Frequently Used evicts the key with the lowest access count, breaking ties by least recent. The O(1) design keeps key → value, key → freq, and freq → ordered set of keys, plus a min_freq pointer. On access, move the key from bucket f to f+1 and bump min_freq if bucket f was the minimum and is now empty. On insert, the new key has frequency 1, so min_freq resets to 1. This construction is described in a short paper by Shah, Mitra and Matani. Shah, Mitra, Matani 2010

from collections import OrderedDict, defaultdict
from typing import Hashable


class LFUCache:
    """O(1) LFU with LRU tie-break: key->(val,freq) and freq->OrderedDict of keys."""

    def __init__(self, capacity: int) -> None:
        self.cap = capacity
        self.vals: dict[Hashable, object] = {}
        self.freq: dict[Hashable, int] = {}
        self.buckets: defaultdict[int, OrderedDict[Hashable, None]] = defaultdict(OrderedDict)
        self.min_freq = 0

    def _touch(self, k: Hashable) -> None:
        f = self.freq[k]
        del self.buckets[f][k]
        if not self.buckets[f]:
            del self.buckets[f]
            if self.min_freq == f:
                self.min_freq = f + 1
        self.freq[k] = f + 1
        self.buckets[f + 1][k] = None

    def get(self, k: Hashable):
        if k not in self.vals:
            return None
        self._touch(k)
        return self.vals[k]

    def put(self, k: Hashable, v: object) -> None:
        if self.cap <= 0:
            return
        if k in self.vals:
            self.vals[k] = v
            self._touch(k)
            return
        if len(self.vals) >= self.cap:
            victim, _ = self.buckets[self.min_freq].popitem(last=False)   # least freq, oldest
            if not self.buckets[self.min_freq]:
                del self.buckets[self.min_freq]
            del self.vals[victim], self.freq[victim]
        self.vals[k], self.freq[k] = v, 1
        self.buckets[1][k] = None
        self.min_freq = 1                 # a brand-new key always has the minimum frequency


c = LFUCache(2)
c.put("a", 1); c.put("b", 2)
c.get("a")                 # a: freq 2, b: freq 1
c.put("c", 3)              # evicts b
assert c.get("b") is None and c.get("a") == 1 and c.get("c") == 3

Patterns: Strategy for the eviction policy if asked to support both (EvictionPolicy with on_access, on_insert, victim), Decorator for adding TTL, metrics or write-through behaviour to a cache.

Concurrency: get mutates the list, so a read-write lock doesn't help: every operation is a write. One lock per cache is the standard answer. To scale, shard into N independent LRU caches by hash(key) % N, each with its own lock (approximate global LRU, much less contention).

Extensions: TTL (store expires_at in the node, check lazily on get, optionally sweep), size-in-bytes capacity (evict in a loop until under budget), eviction listeners (Observer), write-through or write-back to a backing store (Decorator plus Repository), distributed cache (consistent hashing, which is HLD territory).

7.3 Rate limiter

Prompt: "Design a rate limiter: each client may make N requests per time window."

from __future__ import annotations

import threading
import time
from abc import ABC, abstractmethod
from collections import deque
from dataclasses import dataclass
from typing import Callable

Clock = Callable[[], float]


class RateLimiter(ABC):
    @abstractmethod
    def allow(self, key: str, cost: float = 1.0) -> bool: ...


# ---------------- Token bucket ----------------
@dataclass
class _Bucket:
    tokens: float
    updated: float


class TokenBucketLimiter(RateLimiter):
    """Capacity = burst size; refill_rate = sustained tokens/second. Lazy refill, no timer thread."""

    def __init__(self, capacity: float, refill_rate: float, clock: Clock = time.monotonic) -> None:
        self.capacity, self.rate, self.clock = capacity, refill_rate, clock
        self._buckets: dict[str, _Bucket] = {}
        self._lock = threading.Lock()

    def allow(self, key: str, cost: float = 1.0) -> bool:
        now = self.clock()
        with self._lock:
            b = self._buckets.get(key)
            if b is None:
                b = self._buckets[key] = _Bucket(self.capacity, now)
            elapsed = max(0.0, now - b.updated)     # clock read outside the lock may be slightly stale
            b.tokens = min(self.capacity, b.tokens + elapsed * self.rate)
            b.updated = max(b.updated, now)
            if b.tokens >= cost:
                b.tokens -= cost
                return True
            return False


# ---------------- Sliding window log ----------------
class SlidingWindowLogLimiter(RateLimiter):
    """Exact: keep a timestamp per accepted request. Memory O(limit) per key."""

    def __init__(self, limit: int, window_s: float, clock: Clock = time.monotonic) -> None:
        self.limit, self.window, self.clock = limit, window_s, clock
        self._logs: dict[str, deque[float]] = {}
        self._lock = threading.Lock()

    def allow(self, key: str, cost: float = 1.0) -> bool:
        now = self.clock()
        with self._lock:
            log = self._logs.setdefault(key, deque())
            while log and log[0] <= now - self.window:
                log.popleft()
            if len(log) + cost <= self.limit:
                log.extend([now] * int(cost))
                return True
            return False


# ---------------- Sliding window counter ----------------
class SlidingWindowCounterLimiter(RateLimiter):
    """Approximate: weight the previous fixed window by how much of it still overlaps. O(1) memory."""

    def __init__(self, limit: int, window_s: float, clock: Clock = time.monotonic) -> None:
        self.limit, self.window, self.clock = limit, window_s, clock
        self._state: dict[str, tuple[int, float, float]] = {}   # key -> (window_idx, prev, curr)
        self._lock = threading.Lock()

    def allow(self, key: str, cost: float = 1.0) -> bool:
        now = self.clock()
        idx = int(now // self.window)
        with self._lock:
            w, prev, curr = self._state.get(key, (idx, 0.0, 0.0))
            if idx == w + 1:
                prev, curr = curr, 0.0
            elif idx > w + 1:
                prev, curr = 0.0, 0.0
            elapsed = (now % self.window) / self.window
            estimated = prev * (1 - elapsed) + curr
            if estimated + cost <= self.limit:
                self._state[key] = (idx, prev, curr + cost)
                return True
            self._state[key] = (idx, prev, curr)
            return False


# ---------------- tests with a fake clock ----------------
class FakeClock:
    def __init__(self) -> None:
        self.t = 1000.0

    def __call__(self) -> float:
        return self.t


clk = FakeClock()
tb = TokenBucketLimiter(capacity=3, refill_rate=1.0, clock=clk)
assert [tb.allow("u") for _ in range(4)] == [True, True, True, False]   # burst of 3
clk.t += 1.0
assert tb.allow("u") and not tb.allow("u")                                # 1 token refilled
assert tb.allow("other")                                                  # per-key isolation

clk2 = FakeClock()
sw = SlidingWindowLogLimiter(limit=2, window_s=10, clock=clk2)
assert sw.allow("u") and sw.allow("u") and not sw.allow("u")
clk2.t += 10
assert sw.allow("u")

clk3 = FakeClock()
sc = SlidingWindowCounterLimiter(limit=10, window_s=60, clock=clk3)
assert sum(sc.allow("u") for _ in range(15)) == 10

# thread-safety smoke test: 50 threads x 10 calls against capacity 100 -> exactly 100 allowed
frozen = FakeClock()
tb2 = TokenBucketLimiter(capacity=100, refill_rate=0.0, clock=frozen)
allowed: list[bool] = []
lk = threading.Lock()


def hammer() -> None:
    for _ in range(10):
        r = tb2.allow("k")
        with lk:
            allowed.append(r)


ts = [threading.Thread(target=hammer) for _ in range(50)]
for t in ts: t.start()
for t in ts: t.join()
assert sum(allowed) == 100

Design decisions worth saying out loud:

• Lazy refill. No background thread tops up buckets. Each call computes elapsed × rate. This scales to millions of keys and has no timer to manage.
• Monotonic clock, injected. time.monotonic() doesn't jump when the wall clock is adjusted. Injection makes the tests deterministic, including the "exactly 100 of 500 allowed" concurrency test.
• Strategy interface. The middleware depends on RateLimiter; switching algorithms or tiers is configuration.
• Memory. Idle keys accumulate. Mention eviction: an LRU of buckets, or drop buckets that are full and idle (a full bucket is the same as no bucket).

Patterns: Strategy (algorithm), Chain of Responsibility or Decorator (middleware around a handler or client), Factory (limiter per tier from config).

Concurrency: refill-then-consume is a read-modify-write, so it must be atomic per key. One global lock is fine for an interview; per-key locks (a dict of locks guarded by a lock, or striping) reduce contention. The code reads the clock outside the lock to keep the critical section short. Two threads can then apply their timestamps out of order, so it clamps negative elapsed time to zero and never moves updated backwards.

Extensions:

• Distributed: move state to Redis and make refill-and-consume atomic with a server-side script, or accept small inaccuracy with local buckets plus periodic sync. Clock skew between servers is why the server-side script should use the store's own time.
• Multiple limits: per-second and per-day; a composite limiter that checks all, and only consumes if all allow (otherwise you leak tokens from the ones that passed).
• Weighted cost: already supported via cost. For LLM APIs, cost is tokens (see §8d).
• Retry-After: return (allowed, wait_seconds), where for a token bucket wait = (cost − tokens) / rate.

7.4 Elevator system

Prompt: "Design an elevator system for a building with N floors and M elevators."

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum


class Direction(Enum):
    UP = 1
    DOWN = -1
    IDLE = 0


@dataclass
class Elevator:
    id: int
    floor: int = 0
    direction: Direction = Direction.IDLE
    stops: set[int] = field(default_factory=set)

    def add_stop(self, floor: int) -> None:
        if floor != self.floor:
            self.stops.add(floor)
        if self.direction is Direction.IDLE and self.stops:
            self.direction = Direction.UP if floor > self.floor else Direction.DOWN

    def step(self) -> None:
        """Advance one floor using SCAN/LOOK: keep going while stops remain ahead, else reverse."""
        if not self.stops:
            self.direction = Direction.IDLE
            return
        ahead = [s for s in self.stops if (s - self.floor) * self.direction.value > 0]
        if not ahead:
            self.direction = Direction.DOWN if self.direction is Direction.UP else Direction.UP
        self.floor += self.direction.value
        self.stops.discard(self.floor)          # open doors here
        if not self.stops:
            self.direction = Direction.IDLE


@dataclass(frozen=True)
class HallCall:
    floor: int
    direction: Direction


class DispatchStrategy(ABC):
    @abstractmethod
    def pick(self, elevators: list[Elevator], call: HallCall) -> Elevator: ...


class NearestCar(DispatchStrategy):
    """Prefer an idle car or one already moving toward the call in the same direction."""

    def pick(self, elevators: list[Elevator], call: HallCall) -> Elevator:
        def cost(e: Elevator) -> tuple[int, int]:
            dist = abs(e.floor - call.floor)
            if e.direction is Direction.IDLE:
                return (0, dist)
            moving_toward = (call.floor - e.floor) * e.direction.value >= 0
            if moving_toward and e.direction is call.direction:
                return (0, dist)
            return (1, dist + len(e.stops))     # penalise cars that must turn around first
        return min(elevators, key=cost)


class ElevatorController:
    def __init__(self, n: int, strategy: DispatchStrategy) -> None:
        self.cars = [Elevator(i) for i in range(n)]
        self.strategy = strategy

    def hall_call(self, floor: int, direction: Direction) -> int:
        car = self.strategy.pick(self.cars, HallCall(floor, direction))
        car.add_stop(floor)
        return car.id

    def car_call(self, car_id: int, floor: int) -> None:    # button pressed inside the car
        self.cars[car_id].add_stop(floor)

    def tick(self) -> None:
        for c in self.cars:
            c.step()


ctl = ElevatorController(2, NearestCar())
ctl.cars[1].floor = 9
assert ctl.hall_call(2, Direction.UP) == 0      # car 0 at floor 0 is nearest
ctl.car_call(0, 5)
for _ in range(5):
    ctl.tick()
assert ctl.cars[0].floor == 5 and ctl.cars[0].direction is Direction.IDLE

Scheduling strategies to discuss: FCFS (simple, terrible travel), SSTF / nearest (good average, can starve far floors), SCAN/LOOK (no starvation, the standard answer), and zoning (cars assigned to floor bands in tall buildings). Destination dispatch (passengers enter their target floor in the lobby) lets the controller group riders. Mention it as an extension.

Patterns: Strategy (dispatch), State (car states, especially door and maintenance handling), Observer (floor displays, the controller listening to car arrivals), Command (requests as objects in a queue).

Concurrency: in real time, button presses arrive on many threads while each car runs its own loop. Give each car a thread-safe request inbox (a queue.Queue or a lock around stops), and have the controller read car positions as snapshots. Dispatch decisions can be slightly stale; that's acceptable.

Extensions: capacity (skip hall calls when full), maintenance mode (State; dispatcher excludes the car), VIP/fire-service override (a priority command that clears stops), energy-saving idle parking (strategy picks a home floor).

7.5 Movie ticket booking (BookMyShow)

Prompt: "Design a movie ticket booking system." The interviewer is mostly interested in one thing: two users must never get the same seat, and seats must not stay blocked forever when users abandon checkout.

from __future__ import annotations

import threading
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Callable


class SeatState(Enum):
    AVAILABLE = "available"
    HELD = "held"
    BOOKED = "booked"


@dataclass
class ShowSeat:
    seat_id: str
    price: int
    state: SeatState = SeatState.AVAILABLE
    hold_id: str | None = None
    hold_expires: datetime | None = None


@dataclass(frozen=True)
class Hold:
    id: str
    show_id: str
    user_id: str
    seat_ids: tuple[str, ...]
    expires_at: datetime
    amount: int


class BookingStatus(Enum):
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"


@dataclass
class Booking:
    id: str
    show_id: str
    user_id: str
    seat_ids: tuple[str, ...]
    amount: int
    status: BookingStatus = BookingStatus.CONFIRMED


class SeatsUnavailable(Exception):
    pass


class HoldExpired(Exception):
    pass


class Show:
    """One screening. Owns its seat map and a lock: shows never contend with each other."""

    def __init__(self, show_id: str, seats: list[ShowSeat]) -> None:
        self.id = show_id
        self.seats = {s.seat_id: s for s in seats}
        self.lock = threading.Lock()


class PaymentGateway:
    def charge(self, user_id: str, amount: int, idempotency_key: str) -> bool:
        return True


class BookingService:
    HOLD_TTL = timedelta(minutes=10)

    def __init__(self, payments: PaymentGateway, now: Callable[[], datetime] = datetime.now) -> None:
        self._shows: dict[str, Show] = {}
        self._holds: dict[str, Hold] = {}
        self._bookings: dict[str, Booking] = {}
        self._payments, self._now = payments, now

    def add_show(self, show: Show) -> None:
        self._shows[show.id] = show

    def _release_if_expired(self, seat: ShowSeat, now: datetime) -> None:
        if seat.state is SeatState.HELD and seat.hold_expires and seat.hold_expires <= now:
            seat.state, seat.hold_id, seat.hold_expires = SeatState.AVAILABLE, None, None

    def hold(self, show_id: str, user_id: str, seat_ids: list[str]) -> Hold:
        show = self._shows[show_id]
        now = self._now()
        with show.lock:                                 # all-or-nothing across the requested seats
            seats = [show.seats[s] for s in seat_ids]
            for s in seats:
                self._release_if_expired(s, now)        # lazy expiry: no sweeper thread required
            taken = [s.seat_id for s in seats if s.state is not SeatState.AVAILABLE]
            if taken:
                raise SeatsUnavailable(taken)
            hold = Hold(uuid.uuid4().hex, show_id, user_id, tuple(seat_ids),
                        now + self.HOLD_TTL, sum(s.price for s in seats))
            for s in seats:
                s.state, s.hold_id, s.hold_expires = SeatState.HELD, hold.id, hold.expires_at
            self._holds[hold.id] = hold
            return hold

    def confirm(self, hold_id: str) -> Booking:
        hold = self._holds[hold_id]
        show = self._shows[hold.show_id]
        # Pay OUTSIDE the lock: never hold a lock across a slow network call.
        if self._now() >= hold.expires_at:
            raise HoldExpired(hold_id)
        if not self._payments.charge(hold.user_id, hold.amount, idempotency_key=hold_id):
            self.release(hold_id)
            raise RuntimeError("payment declined")
        with show.lock:
            seats = [show.seats[s] for s in hold.seat_ids]
            if any(s.hold_id != hold_id for s in seats):    # expired and re-taken meanwhile
                raise HoldExpired(f"{hold_id}: refund required")
            for s in seats:
                s.state, s.hold_id, s.hold_expires = SeatState.BOOKED, None, None
            booking = Booking(uuid.uuid4().hex, show.id, hold.user_id, hold.seat_ids, hold.amount)
            self._bookings[booking.id] = booking
            del self._holds[hold_id]
            return booking

    def release(self, hold_id: str) -> None:
        hold = self._holds.pop(hold_id, None)
        if not hold:
            return
        show = self._shows[hold.show_id]
        with show.lock:
            for sid in hold.seat_ids:
                s = show.seats[sid]
                if s.hold_id == hold_id:
                    s.state, s.hold_id, s.hold_expires = SeatState.AVAILABLE, None, None


# --- race test: 20 users fight for the same 2 seats; exactly one wins ---
svc = BookingService(PaymentGateway())
svc.add_show(Show("s1", [ShowSeat(f"A{i}", 250) for i in range(1, 11)]))
wins: list[str] = []
lk = threading.Lock()


def attempt(user: str) -> None:
    try:
        h = svc.hold("s1", user, ["A1", "A2"])
        svc.confirm(h.id)
        with lk:
            wins.append(user)
    except SeatsUnavailable:
        pass


ts = [threading.Thread(target=attempt, args=(f"u{i}",)) for i in range(20)]
for t in ts: t.start()
for t in ts: t.join()
assert len(wins) == 1

# --- expiry test with a controllable clock ---
now = [datetime(2026, 1, 1, 18, 0)]
svc2 = BookingService(PaymentGateway(), now=lambda: now[0])
svc2.add_show(Show("s2", [ShowSeat("B1", 300)]))
svc2.hold("s2", "alice", ["B1"])
now[0] += timedelta(minutes=11)
h_bob = svc2.hold("s2", "bob", ["B1"])          # alice's hold lapsed lazily
assert svc2.confirm(h_bob.id).user_id == "bob"

Design decisions worth saying out loud:

• Two-phase: hold, then confirm. The hold is fast and in-memory (or one DB statement). Payment happens between the phases, outside any lock.
• Lock granularity = one show. Seats in different shows never contend, and taking one lock for a multi-seat request makes all-or-nothing trivial and deadlock impossible. Per-seat locks would need ordered acquisition (sort seat IDs) for multi-seat holds.
• Lazy expiry. Expired holds are released when someone next touches the seat, so no sweeper thread is needed for correctness. A background sweeper can still run to keep the seat map display accurate.
• Re-validate on confirm. If payment took longer than the TTL and someone else grabbed the seat, confirm detects that the hold_id no longer matches and triggers a refund. The idempotency key (the hold ID) means a retried confirm can't double-charge.

With a relational database shared by many app servers, the same logic becomes either a pessimistic transaction or a single conditional update:

# The same idea against a relational DB (pseudo-code, psycopg-style placeholders).
# Pessimistic: lock the rows, then check and update inside one transaction.
HOLD_PESSIMISTIC = """
BEGIN;
SELECT seat_id, state FROM show_seats
 WHERE show_id = %(show)s AND seat_id = ANY(%(seats)s)
 FOR UPDATE;                         -- row locks until COMMIT
-- application checks every row is AVAILABLE (or its hold has expired), else ROLLBACK
UPDATE show_seats SET state = 'HELD', hold_id = %(hold)s, hold_expires = now() + interval '10 minutes'
 WHERE show_id = %(show)s AND seat_id = ANY(%(seats)s);
COMMIT;
"""

# Optimistic / conditional: one atomic statement, check the affected row count.
HOLD_CONDITIONAL = """
UPDATE show_seats SET state = 'HELD', hold_id = %(hold)s, hold_expires = now() + interval '10 minutes'
 WHERE show_id = %(show)s AND seat_id = ANY(%(seats)s)
   AND (state = 'AVAILABLE' OR (state = 'HELD' AND hold_expires < now()));
-- if rowcount != len(seats): ROLLBACK (someone else got at least one seat)
"""

Patterns: Facade (BookingService), State (seat lifecycle, as an enum here), Strategy (pricing by category/time, payment method), Observer (send the confirmation email/SMS on booking.confirmed), Repository (shows, bookings).

Extensions: dynamic pricing (Strategy), coupons (Decorator over pricing or Chain of discount rules), waitlist when sold out (queue + Observer on release), seat-adjacency suggestions ("best 4 together"), cancellation with refund policy (Strategy by time-before-show), and a distributed version (DB conditional updates or a Redis SET key value NX PX ttl-style hold, idempotent payment webhooks).

7.6 Splitwise / expense sharing

Prompt: "Design an app where friends record shared expenses and see who owes whom."

• Clarify: split types (equal, exact, percentage, shares)? Groups? Multiple currencies? Must balances simplify to the fewest transfers? Settling up?
• Entities: User, Group, Expense (payer, amount, shares), SplitStrategy (Equal/Exact/Percent), ExpenseManager or Ledger (net balances), Settlement.
• Key modelling choice: store the immutable list of expenses (the source of truth) and maintain a net balance per user (positive = is owed). Pairwise "A owes B" tables are a view you derive, not the core state.

from __future__ import annotations

import heapq
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass


@dataclass(frozen=True)
class User:
    id: str
    name: str


class SplitStrategy(ABC):
    @abstractmethod
    def shares(self, amount: int, participants: list[str]) -> dict[str, int]: ...


class EqualSplit(SplitStrategy):
    def shares(self, amount: int, participants: list[str]) -> dict[str, int]:
        base, rem = divmod(amount, len(participants))          # integer paise/cents, no floats
        return {u: base + (1 if i < rem else 0) for i, u in enumerate(participants)}


class ExactSplit(SplitStrategy):
    def __init__(self, amounts: dict[str, int]) -> None:
        self.amounts = amounts

    def shares(self, amount: int, participants: list[str]) -> dict[str, int]:
        if sum(self.amounts.values()) != amount:
            raise ValueError("exact amounts must add up to the total")
        return dict(self.amounts)


class PercentSplit(SplitStrategy):
    def __init__(self, pct: dict[str, float]) -> None:
        if abs(sum(pct.values()) - 100) > 1e-9:
            raise ValueError("percentages must add up to 100")
        self.pct = pct

    def shares(self, amount: int, participants: list[str]) -> dict[str, int]:
        raw = {u: amount * p / 100 for u, p in self.pct.items()}
        out = {u: int(v) for u, v in raw.items()}
        leftover = amount - sum(out.values())
        for u in sorted(raw, key=lambda u: raw[u] - out[u], reverse=True)[:leftover]:
            out[u] += 1                                          # largest-remainder rounding
        return out


@dataclass(frozen=True)
class Expense:
    id: str
    paid_by: str
    amount: int
    shares: dict[str, int]


class ExpenseManager:
    def __init__(self) -> None:
        self.expenses: list[Expense] = []
        self.net: defaultdict[str, int] = defaultdict(int)       # +ve: is owed, -ve: owes

    def add(self, eid: str, paid_by: str, amount: int, participants: list[str],
            strategy: SplitStrategy) -> Expense:
        shares = strategy.shares(amount, participants)
        e = Expense(eid, paid_by, amount, shares)
        self.expenses.append(e)
        self.net[paid_by] += amount
        for u, s in shares.items():
            self.net[u] -= s
        return e

    def simplify(self) -> list[tuple[str, str, int]]:
        """Greedy: match the largest creditor with the largest debtor. At most n-1 transfers.
        (Finding the true minimum number of transfers is NP-hard; this heuristic is the usual answer.)"""
        creditors = [(-v, u) for u, v in self.net.items() if v > 0]
        debtors = [(v, u) for u, v in self.net.items() if v < 0]
        heapq.heapify(creditors); heapq.heapify(debtors)
        out: list[tuple[str, str, int]] = []
        while creditors and debtors:
            c_amt, c = heapq.heappop(creditors)
            d_amt, d = heapq.heappop(debtors)
            pay = min(-c_amt, -d_amt)
            out.append((d, c, pay))
            if -c_amt > pay: heapq.heappush(creditors, (c_amt + pay, c))
            if -d_amt > pay: heapq.heappush(debtors, (d_amt + pay, d))
        return out


m = ExpenseManager()
m.add("e1", "A", 3000, ["A", "B", "C"], EqualSplit())         # B, C owe A 1000 each
m.add("e2", "B", 1000, ["B", "C"], ExactSplit({"B": 0, "C": 1000}))
assert sum(m.net.values()) == 0
# net: A +2000, B 0, C -2000  ->  one transfer instead of two
assert m.simplify() == [("C", "A", 2000)]
assert PercentSplit({"A": 33.34, "B": 33.33, "C": 33.33}).shares(100, []) == {"A": 34, "B": 33, "C": 33}

Debt simplification: compute each person's net balance, then repeatedly match the largest creditor with the largest debtor. This needs at most n−1 transfers. Finding the true minimum number of transfers is NP-hard in general (it contains a subset-sum-style problem), so say "greedy, at most n−1, and optimal minimum is NP-hard" rather than claiming optimality.

Money: integer minor units; equal splits distribute the remainder one unit at a time; percentage splits use largest-remainder rounding so shares always add up to the total exactly.

Patterns: Strategy (split types), Factory (parse "EXACT 100 200" into a strategy in a machine-coding CLI), Observer (notify participants), Command (each CLI line).

Concurrency: adding an expense updates several balances, so lock per group (or per ledger) to keep the sum-to-zero invariant. Balances are derivable from the expense list, so you can rebuild them after a crash.

Extensions: multiple currencies (store per-currency balances; convert only at settlement), edit/delete expense (reverse the old shares, apply the new ones; the immutable list makes this auditable), recurring expenses, groups within groups.

7.7 Vending machine (State pattern)

Prompt: "Design a vending machine." This is the textbook State-pattern problem: the same buttons do different things depending on whether money has been inserted or the machine is sold out.

• Clarify: coins/notes accepted? Give change, and what if it can't? Cancel/refund? Admin restock? Card payments?
• States: Idle, HasMoney, SoldOut (optionally Dispensing, Maintenance).
• Entities: VendingMachine (context), VMState subclasses, Slot/Inventory, Coin enum, a change-making component.

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import IntEnum


class Coin(IntEnum):
    FIVE = 5
    TEN = 10
    TWENTY_FIVE = 25
    HUNDRED = 100


@dataclass
class Slot:
    name: str
    price: int
    qty: int


class VMState(ABC):
    def insert(self, vm: VendingMachine, coin: Coin) -> None:
        raise RuntimeError(f"cannot insert coin in {type(self).__name__}")

    def select(self, vm: VendingMachine, code: str) -> None:
        raise RuntimeError(f"cannot select in {type(self).__name__}")

    def cancel(self, vm: VendingMachine) -> list[Coin]:
        return []


class Idle(VMState):
    def insert(self, vm: VendingMachine, coin: Coin) -> None:
        vm.balance += coin
        vm.state = HasMoney()


class HasMoney(VMState):
    def insert(self, vm: VendingMachine, coin: Coin) -> None:
        vm.balance += coin

    def select(self, vm: VendingMachine, code: str) -> None:
        slot = vm.slots.get(code)
        if slot is None or slot.qty == 0:
            raise ValueError("sold out or invalid code")
        if vm.balance < slot.price:
            raise ValueError(f"insert {slot.price - vm.balance} more")
        change = vm.make_change(vm.balance - slot.price)      # check change BEFORE dispensing
        slot.qty -= 1
        vm.balance = 0
        vm.dispensed.append(slot.name)
        vm.change_tray.extend(change)
        vm.state = Idle() if any(s.qty for s in vm.slots.values()) else SoldOut()

    def cancel(self, vm: VendingMachine) -> list[Coin]:
        refund = vm.make_change(vm.balance)
        vm.balance = 0
        vm.state = Idle()
        return refund


class SoldOut(VMState):
    pass                                   # every action rejected via base-class defaults


class VendingMachine:
    def __init__(self, slots: dict[str, Slot], float_coins: dict[Coin, int]) -> None:
        self.slots = slots
        self.coins = dict(float_coins)       # coins available for change
        self.balance = 0
        self.state: VMState = Idle()
        self.dispensed: list[str] = []
        self.change_tray: list[Coin] = []

    def make_change(self, amount: int) -> list[Coin]:
        out: list[Coin] = []
        for c in sorted(Coin, reverse=True):  # greedy works for this canonical coin set
            while amount >= c and self.coins.get(c, 0) > 0:
                amount -= c
                self.coins[c] -= 1
                out.append(c)
        if amount:
            for c in out:                    # roll back
                self.coins[c] += 1
            raise ValueError("exact change only")
        return out

    def insert(self, coin: Coin) -> None:
        self.state.insert(self, coin)
        self.coins[coin] = self.coins.get(coin, 0) + 1

    def select(self, code: str) -> None: self.state.select(self, code)
    def cancel(self) -> list[Coin]: return self.state.cancel(self)


vm = VendingMachine({"A1": Slot("chips", 65, 1)}, {Coin.TEN: 5, Coin.FIVE: 5})
vm.insert(Coin.HUNDRED)
vm.select("A1")
assert vm.dispensed == ["chips"] and sum(vm.change_tray) == 35 and isinstance(vm.state, SoldOut)

Two details interviewers look for: check you can make change before dispensing (otherwise the customer loses money or you give away product), and roll back the coin counts if change can't be made. Greedy change-making is correct for canonical coin systems like the one above, but not for arbitrary denominations, where you'd need dynamic programming. Say so.

Patterns: State (core), Strategy (change-making, payment method), Singleton (often requested; resist or accept), Observer (low-stock alerts to the operator).

Concurrency: a physical machine has one user at a time, so concurrency is usually out of scope. If a remote app can also buy, guard each state transition with one lock.

7.8 In-memory pub-sub / message queue

Prompt: "Design an in-memory message queue like a mini Kafka" or "a pub-sub system with topics and subscribers."

• Clarify: push (broker calls subscribers) or pull (consumers poll)? Delivery guarantee: at-most-once or at-least-once? Ordering (per topic, per key)? Consumer groups (each message to one member of a group, every group gets every message)? Retention and replay?
• Entities: Broker, Topic (append-only log), Message (offset, key, payload), Consumer, consumer-group offsets.
• Key design: an append-only log with per-group offsets gives replay, multiple independent consumer groups and ordering for free. Committing the offset after processing gives at-least-once delivery; committing before gives at-most-once.

from __future__ import annotations

import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable
import itertools
import time


@dataclass(frozen=True)
class Message:
    offset: int
    key: str | None
    payload: dict
    ts: float = field(default_factory=time.time)


class Topic:
    """Append-only log; consumers track their own offsets (Kafka-style, single partition)."""

    def __init__(self, name: str) -> None:
        self.name = name
        self.log: list[Message] = []
        self.cond = threading.Condition()

    def append(self, key: str | None, payload: dict) -> int:
        with self.cond:
            msg = Message(len(self.log), key, payload)
            self.log.append(msg)
            self.cond.notify_all()
            return msg.offset


class Broker:
    def __init__(self) -> None:
        self.topics: dict[str, Topic] = {}
        self.group_offsets: defaultdict[tuple[str, str], int] = defaultdict(int)  # (group, topic)
        self._lock = threading.Lock()

    def topic(self, name: str) -> Topic:
        with self._lock:
            return self.topics.setdefault(name, Topic(name))

    def publish(self, topic: str, payload: dict, key: str | None = None) -> int:
        return self.topic(topic).append(key, payload)

    def poll(self, group: str, topic: str, max_msgs: int = 10, timeout: float = 0.5) -> list[Message]:
        t = self.topic(topic)
        with t.cond:
            start = self.group_offsets[(group, topic)]
            if start >= len(t.log):
                t.cond.wait(timeout)                     # long-poll
            return t.log[start:start + max_msgs]

    def commit(self, group: str, topic: str, next_offset: int) -> None:
        with self._lock:                                 # at-least-once: commit after processing
            cur = self.group_offsets[(group, topic)]
            self.group_offsets[(group, topic)] = max(cur, next_offset)


class Consumer(threading.Thread):
    def __init__(self, broker: Broker, group: str, topic: str, handler: Callable[[Message], None]) -> None:
        super().__init__(daemon=True)
        self.b, self.g, self.t, self.h = broker, group, topic, handler
        self.stop_evt = threading.Event()

    def run(self) -> None:
        while not self.stop_evt.is_set():
            batch = self.b.poll(self.g, self.t)
            for m in batch:
                self.h(m)                                # if this crashes, the batch is redelivered
            if batch:
                self.b.commit(self.g, self.t, batch[-1].offset + 1)


broker = Broker()
got_a: list[int] = []
got_b: list[int] = []
ca = Consumer(broker, "billing", "orders", lambda m: got_a.append(m.payload["id"]))
cb = Consumer(broker, "email", "orders", lambda m: got_b.append(m.payload["id"]))
ca.start(); cb.start()
for i in range(5):
    broker.publish("orders", {"id": i}, key=f"user-{i % 2}")
deadline = time.time() + 3
while (len(got_a) < 5 or len(got_b) < 5) and time.time() < deadline:
    time.sleep(0.01)
ca.stop_evt.set(); cb.stop_evt.set()
assert got_a == got_b == [0, 1, 2, 3, 4]       # each group sees every message, in order

Patterns: Observer/pub-sub (core), producer–consumer with condition variables (long-polling), Strategy (partitioner: hash of key → partition), Command (messages as commands for workers).

Concurrency: each topic's condition variable guards its log; the broker lock guards topic creation and offsets. Consumers block in poll without busy-waiting. Since commits only move forward (max), a slow duplicate commit can't rewind a group.

Extensions: partitions (N logs per topic, key-hash routing, one consumer per partition per group, which is how per-key ordering survives parallelism), retention by size/time (drop the log prefix, keep a base offset), dead-letter topic after K failed deliveries, back-pressure (bounded topics), push delivery via a dispatcher thread per subscriber with its own queue.

7.9 Logger / notification service

Prompt: "Design a logging library" (or the close cousin: "a notification service that sends email, SMS and push"). Both are about fan-out to pluggable sinks with filtering, formatting and non-blocking delivery.

• Clarify: levels and per-logger thresholds? Sinks (console, file, remote)? Formats (text, JSON)? Must logging ever block the caller? What to do when the buffer is full? For notifications: user preferences per channel, retries, templates, rate limits, quiet hours?
• Entities: Logger, LogRecord (value), Level (IntEnum, ordered), Formatter, Sink/Appender, AsyncDispatcher. For notifications: Notification, Channel, ChannelSender per channel, PreferenceStore, TemplateRenderer, RetryPolicy.

from __future__ import annotations

import queue
import sys
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import IntEnum
from typing import TextIO


class Level(IntEnum):
    DEBUG = 10
    INFO = 20
    WARN = 30
    ERROR = 40


@dataclass(frozen=True)
class LogRecord:
    level: Level
    msg: str
    logger: str
    ts: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    extra: dict = field(default_factory=dict)


class Formatter(ABC):
    @abstractmethod
    def format(self, r: LogRecord) -> str: ...


class TextFormatter(Formatter):
    def format(self, r: LogRecord) -> str:
        return f"{r.ts:%H:%M:%S} {r.level.name:<5} [{r.logger}] {r.msg}"


class Sink(ABC):                               # a.k.a. Appender / Handler
    def __init__(self, min_level: Level, fmt: Formatter) -> None:
        self.min_level, self.fmt = min_level, fmt

    def handle(self, r: LogRecord) -> None:
        if r.level >= self.min_level:
            self.emit(self.fmt.format(r))

    @abstractmethod
    def emit(self, line: str) -> None: ...


class StreamSink(Sink):
    def __init__(self, min_level: Level, fmt: Formatter, stream: TextIO = sys.stdout) -> None:
        super().__init__(min_level, fmt)
        self.stream = stream

    def emit(self, line: str) -> None:
        print(line, file=self.stream)


class MemorySink(Sink):
    def __init__(self, min_level: Level, fmt: Formatter) -> None:
        super().__init__(min_level, fmt)
        self.lines: list[str] = []

    def emit(self, line: str) -> None:
        self.lines.append(line)


class AsyncDispatcher:
    """Callers enqueue and return immediately; one background thread fans out to sinks."""

    def __init__(self, sinks: list[Sink], maxsize: int = 10_000) -> None:
        self.sinks = sinks
        self.q: queue.Queue[LogRecord | None] = queue.Queue(maxsize)
        self.dropped = 0
        self.t = threading.Thread(target=self._run, daemon=True)
        self.t.start()

    def submit(self, r: LogRecord) -> None:
        try:
            self.q.put_nowait(r)               # never block the request path on logging
        except queue.Full:
            self.dropped += 1

    def _run(self) -> None:
        while (r := self.q.get()) is not None:
            for s in self.sinks:
                try:
                    s.handle(r)
                except Exception:
                    pass                        # a broken sink must not kill the logger

    def close(self) -> None:
        self.q.put(None)
        self.t.join()


class Logger:
    def __init__(self, name: str, dispatcher: AsyncDispatcher, level: Level = Level.INFO) -> None:
        self.name, self.d, self.level = name, dispatcher, level

    def log(self, level: Level, msg: str, **extra: object) -> None:
        if level >= self.level:                 # cheap early filter before building a record
            self.d.submit(LogRecord(level, msg, self.name, extra=extra))

    def info(self, msg: str, **kw: object) -> None: self.log(Level.INFO, msg, **kw)
    def error(self, msg: str, **kw: object) -> None: self.log(Level.ERROR, msg, **kw)
    def debug(self, msg: str, **kw: object) -> None: self.log(Level.DEBUG, msg, **kw)


mem = MemorySink(Level.WARN, TextFormatter())
disp = AsyncDispatcher([mem])
log = Logger("checkout", disp, Level.DEBUG)
log.debug("cart loaded"); log.info("paying"); log.error("card declined", order=42)
disp.close()
assert len(mem.lines) == 1 and "card declined" in mem.lines[0]

Patterns: Strategy (formatter), Chain of Responsibility or a list of sinks with level filters (Observer-like fan-out), Singleton/registry for named loggers (Python's own logging.getLogger(name) returns the same logger per name), Decorator (add context fields), producer–consumer (async dispatch). For notifications: Factory (channel → sender), Template Method (render → send → record), Strategy (retry/backoff), Observer (domain events trigger notifications).

Concurrency: the request path only enqueues (put_nowait); one background thread owns the sinks, so sinks need no locking. Decide the full-buffer policy explicitly: drop and count (shown), block, or drop DEBUG first. Flush on shutdown (close() enqueues a sentinel and joins).

Extensions: structured JSON logs, sampling of DEBUG, per-module levels, file rotation, a remote sink with batching and retry. For notifications: idempotency key per notification (so retries don't double-send), per-user rate limits, priority queues (OTP before marketing), and a dead-letter store.

7.10 Games: tic-tac-toe, snake-and-ladder (and a chess template)

Game problems all share one skeleton: Game (turn loop, status, history) → Board (state, legal-move check, win detection) → Player (abstract, with Human/Bot implementations supplying moves) → Move (value object) → rule components (dice, win checker) behind interfaces. Get the skeleton right and the specific game is a few methods.

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum


class Mark(Enum):
    X = "X"
    O = "O"


class GameStatus(Enum):
    IN_PROGRESS = "in_progress"
    WON = "won"
    DRAW = "draw"


@dataclass(frozen=True)
class Move:
    row: int
    col: int


class Player(ABC):
    def __init__(self, name: str, mark: Mark) -> None:
        self.name, self.mark = name, mark

    @abstractmethod
    def next_move(self, board: Board) -> Move: ...


class ScriptedPlayer(Player):              # stands in for HumanPlayer / BotPlayer
    def __init__(self, name: str, mark: Mark, moves: list[Move]) -> None:
        super().__init__(name, mark)
        self._moves = iter(moves)

    def next_move(self, board: Board) -> Move:
        return next(self._moves)


class Board:
    def __init__(self, n: int = 3) -> None:
        self.n = n
        self.cells: list[list[Mark | None]] = [[None] * n for _ in range(n)]
        # O(1) win check: running sums per row/col/diagonal (+1 for X, -1 for O)
        self.rows, self.cols = [0] * n, [0] * n
        self.diag = self.anti = 0
        self.filled = 0

    def place(self, m: Move, mark: Mark) -> bool:
        if not (0 <= m.row < self.n and 0 <= m.col < self.n) or self.cells[m.row][m.col]:
            raise ValueError(f"illegal move {m}")
        self.cells[m.row][m.col] = mark
        self.filled += 1
        d = 1 if mark is Mark.X else -1
        self.rows[m.row] += d
        self.cols[m.col] += d
        if m.row == m.col: self.diag += d
        if m.row + m.col == self.n - 1: self.anti += d
        return self.n in (abs(self.rows[m.row]), abs(self.cols[m.col]), abs(self.diag), abs(self.anti))

    @property
    def full(self) -> bool:
        return self.filled == self.n * self.n


class Game:
    def __init__(self, board: Board, players: list[Player]) -> None:
        self.board, self.players = board, players
        self.turn = 0
        self.status = GameStatus.IN_PROGRESS
        self.winner: Player | None = None
        self.history: list[tuple[str, Move]] = []

    def play_turn(self) -> None:
        if self.status is not GameStatus.IN_PROGRESS:
            raise RuntimeError("game over")
        p = self.players[self.turn % len(self.players)]
        mv = p.next_move(self.board)
        won = self.board.place(mv, p.mark)
        self.history.append((p.name, mv))
        if won:
            self.status, self.winner = GameStatus.WON, p
        elif self.board.full:
            self.status = GameStatus.DRAW
        self.turn += 1

    def play(self) -> GameStatus:
        while self.status is GameStatus.IN_PROGRESS:
            self.play_turn()
        return self.status


x = ScriptedPlayer("ann", Mark.X, [Move(0, 0), Move(1, 1), Move(2, 2)])
o = ScriptedPlayer("bob", Mark.O, [Move(0, 1), Move(0, 2)])
g = Game(Board(3), [x, o])
assert g.play() is GameStatus.WON and g.winner is x

The O(1) win check (running row/column/diagonal sums) is a detail interviewers like: it works for any N×N board without rescanning. Snake-and-ladder reuses the same skeleton with a Dice interface (injectable for tests) and the observation that snakes and ladders are the same thing: a jump table.

from __future__ import annotations

import random
from abc import ABC, abstractmethod
from collections import deque
from dataclasses import dataclass


class Dice(ABC):
    @abstractmethod
    def roll(self) -> int: ...


class RandomDice(Dice):
    def __init__(self, faces: int = 6, seed: int | None = None) -> None:
        self.faces, self.rng = faces, random.Random(seed)

    def roll(self) -> int:
        return self.rng.randint(1, self.faces)


@dataclass
class Player:
    name: str
    pos: int = 0


class SnakeLadderBoard:
    def __init__(self, size: int, jumps: dict[int, int]) -> None:
        # snakes and ladders are the same thing: a jump from one cell to another
        if any(a in (0, size) for a in jumps):
            raise ValueError("no jump on start/end cells")
        self.size, self.jumps = size, jumps

    def resolve(self, pos: int) -> int:
        return self.jumps.get(pos, pos)


class SnakeLadderGame:
    def __init__(self, board: SnakeLadderBoard, players: list[Player], dice: Dice) -> None:
        self.board, self.players, self.dice = board, deque(players), dice
        self.winner: Player | None = None

    def play_turn(self) -> None:
        p = self.players[0]
        target = p.pos + self.dice.roll()
        if target <= self.board.size:            # rule choice: must land exactly on the last cell
            p.pos = self.board.resolve(target)
        if p.pos == self.board.size:
            self.winner = p
        self.players.rotate(-1)

    def play(self, max_turns: int = 10_000) -> Player:
        for _ in range(max_turns):
            self.play_turn()
            if self.winner:
                return self.winner
        raise RuntimeError("no winner within turn limit")


game = SnakeLadderGame(SnakeLadderBoard(100, {3: 22, 5: 8, 27: 1, 99: 10}),
                       [Player("a"), Player("b")], RandomDice(seed=7))
assert game.play().pos == 100

Chess template: Piece (abstract) with legal_moves(board, from_sq) per subclass (or a movement Strategy per piece type), Board of 64 squares, Move with special flags (castle, en passant, promotion), Game with a MoveValidator that rejects moves leaving your own king in check, and Command for moves so undo works. Don't try to implement every rule. Agree on a subset and show where the rest would go.

Patterns: Template Method (turn loop), Strategy (bot move choice, dice type, win rule), Factory (pieces from a starting layout), Command (moves, for undo/replay), State (game status).

Extensions: N×N board with K-in-a-row, more players, undo (Command), save/replay (history of moves), networked play (each turn becomes a message; validate on the server).

8. AI-flavoured LLD problems

AI-engineering loops increasingly replace "design a parking lot" with "design the LLM client your team would use" or "build a tool-calling agent loop". The craft is the same: entities, interfaces at the points of variation, correct state handling, concurrency, tests. The domain adds a few concerns of its own: unreliable and rate-limited providers, token budgets, non-deterministic outputs, streaming, cost, and multi-tenancy. All code below is provider-agnostic and runs against fakes. It does not reproduce any vendor SDK's API; a real implementation would put each vendor behind an Adapter. For the product-level view of these components, see B1, B3 · Memory, B4 · Agents and B9 · Harness engineering.

8a. LLM client library: providers, retries, fallback, streaming, cost

Requirements: one interface over several providers; retries with exponential backoff and jitter on transient errors; respect a server's retry-after hint; fall back to another provider (possibly with a different model) when one is down; timeouts; streaming; per-tenant token and cost accounting. Non-goals: prompt management, caching (8h), scheduling (8d).

Entities: value objects ChatMessage, CompletionRequest, CompletionResponse, Usage; an error taxonomy LLMError → RateLimited / ProviderUnavailable (retryable) / BadRequest (not); the LLMProvider interface; wrappers RetryingProvider, MeteredProvider; FallbackChain; UsageLedger.

from __future__ import annotations

import random
import time
from abc import ABC, abstractmethod
from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Callable, Literal


# ---------- value objects ----------
@dataclass(frozen=True)
class ChatMessage:
    role: Literal["system", "user", "assistant", "tool"]
    content: str


@dataclass(frozen=True)
class CompletionRequest:
    model: str
    messages: tuple[ChatMessage, ...]
    max_tokens: int = 512
    temperature: float = 0.0
    timeout_s: float = 30.0


@dataclass(frozen=True)
class Usage:
    input_tokens: int
    output_tokens: int


@dataclass(frozen=True)
class CompletionResponse:
    text: str
    model: str
    provider: str
    usage: Usage
    finish_reason: str = "stop"


# ---------- error taxonomy: decides retry vs fallback vs fail ----------
class LLMError(Exception):
    retryable = False


class RateLimited(LLMError):
    retryable = True

    def __init__(self, retry_after: float | None = None) -> None:
        super().__init__("rate limited")
        self.retry_after = retry_after


class ProviderUnavailable(LLMError):     # 5xx, connection reset, timeout
    retryable = True


class BadRequest(LLMError):              # 4xx: retrying the same payload won't help
    retryable = False


# ---------- Strategy: one interface, many providers ----------
class LLMProvider(ABC):
    name: str

    @abstractmethod
    def complete(self, req: CompletionRequest) -> CompletionResponse: ...

    @abstractmethod
    def stream(self, req: CompletionRequest) -> Iterator[str]: ...


class FakeProvider(LLMProvider):
    """Test double. Real adapters translate CompletionRequest into a vendor SDK call
    and map vendor errors into the taxonomy above."""

    def __init__(self, name: str, fail_times: int = 0, error: type[LLMError] = ProviderUnavailable) -> None:
        self.name, self.fail_times, self.error, self.calls = name, fail_times, error, 0

    def complete(self, req: CompletionRequest) -> CompletionResponse:
        self.calls += 1
        if self.calls <= self.fail_times:
            raise self.error()
        text = f"[{self.name}] answer"
        return CompletionResponse(text, req.model, self.name, Usage(sum(len(m.content) // 4 for m in req.messages), 5))

    def stream(self, req: CompletionRequest) -> Iterator[str]:
        yield from self.complete(req).text.split(" ")


# ---------- Decorator: retries with exponential backoff + full jitter ----------
class RetryingProvider(LLMProvider):
    def __init__(self, inner: LLMProvider, max_attempts: int = 4, base: float = 0.5, cap: float = 20.0,
                 sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None) -> None:
        self.inner, self.max_attempts, self.base, self.cap = inner, max_attempts, base, cap
        self.name, self._sleep, self._rng = inner.name, sleep, rng or random.Random()

    def _delay(self, attempt: int, err: LLMError) -> float:
        if isinstance(err, RateLimited) and err.retry_after is not None:
            return err.retry_after                                  # honour the server's hint
        return self._rng.uniform(0, min(self.cap, self.base * 2 ** attempt))   # "full jitter"

    def complete(self, req: CompletionRequest) -> CompletionResponse:
        for attempt in range(self.max_attempts):
            try:
                return self.inner.complete(req)
            except LLMError as e:
                if not e.retryable or attempt == self.max_attempts - 1:
                    raise
                self._sleep(self._delay(attempt, e))
        raise AssertionError("unreachable")

    def stream(self, req: CompletionRequest) -> Iterator[str]:
        # Only retry before the first chunk; once tokens reach the caller, a retry would duplicate them.
        for attempt in range(self.max_attempts):
            it = self.inner.stream(req)
            try:
                first = next(it)
            except StopIteration:
                return
            except LLMError as e:
                if not e.retryable or attempt == self.max_attempts - 1:
                    raise
                self._sleep(self._delay(attempt, e))
                continue
            yield first
            yield from it
            return


# ---------- Decorator: token + cost accounting ----------
@dataclass
class UsageLedger:
    prices_per_mtok: dict[str, tuple[float, float]]               # model -> (input $, output $) per 1M tokens
    totals: dict[str, float] = field(default_factory=dict)

    def record(self, tenant: str, resp: CompletionResponse) -> float:
        pin, pout = self.prices_per_mtok.get(resp.model, (0.0, 0.0))
        cost = (resp.usage.input_tokens * pin + resp.usage.output_tokens * pout) / 1_000_000
        self.totals[tenant] = self.totals.get(tenant, 0.0) + cost
        return cost


class MeteredProvider(LLMProvider):
    def __init__(self, inner: LLMProvider, ledger: UsageLedger, tenant: str) -> None:
        self.inner, self.ledger, self.tenant, self.name = inner, ledger, tenant, inner.name

    def complete(self, req: CompletionRequest) -> CompletionResponse:
        resp = self.inner.complete(req)
        self.ledger.record(self.tenant, resp)
        return resp

    def stream(self, req: CompletionRequest) -> Iterator[str]:
        yield from self.inner.stream(req)     # real impl: read usage from the final stream event


# ---------- Chain of Responsibility: fallback across providers ----------
class AllProvidersFailed(LLMError):
    def __init__(self, errors: list[tuple[str, Exception]]) -> None:
        super().__init__("; ".join(f"{n}: {e!r}" for n, e in errors))
        self.errors = errors


class FallbackChain(LLMProvider):
    def __init__(self, providers: list[LLMProvider], model_map: dict[str, str] | None = None) -> None:
        self.providers, self.name = providers, "fallback"
        self.model_map = model_map or {}      # provider name -> model id to use on that provider

    def complete(self, req: CompletionRequest) -> CompletionResponse:
        errors: list[tuple[str, Exception]] = []
        for p in self.providers:
            r = CompletionRequest(self.model_map.get(p.name, req.model), req.messages,
                                  req.max_tokens, req.temperature, req.timeout_s)
            try:
                return p.complete(r)
            except BadRequest:
                raise                         # our bug, not theirs: don't burn the fallback chain
            except LLMError as e:
                errors.append((p.name, e))
        raise AllProvidersFailed(errors)

    def stream(self, req: CompletionRequest) -> Iterator[str]:
        yield from self.providers[0].stream(req)


# ---------- composition root ----------
no_sleep: Callable[[float], None] = lambda s: None
primary = FakeProvider("primary", fail_times=10)                 # hard down
secondary = FakeProvider("secondary", fail_times=1)              # one transient blip
ledger = UsageLedger({"small-model": (0.5, 1.5)})
client = MeteredProvider(
    FallbackChain([RetryingProvider(primary, sleep=no_sleep), RetryingProvider(secondary, sleep=no_sleep)],
                  model_map={"secondary": "small-model"}),
    ledger, tenant="acme")
resp = client.complete(CompletionRequest("big-model", (ChatMessage("user", "hello there, model"),)))
assert resp.provider == "secondary" and primary.calls == 4 and secondary.calls == 2
assert ledger.totals["acme"] > 0
assert list(RetryingProvider(FakeProvider("s", fail_times=2), sleep=no_sleep).stream(
    CompletionRequest("m", (ChatMessage("user", "hi"),)))) == ["[s]", "answer"]

Design decisions worth saying out loud:

• The error taxonomy drives everything. Retry decides on retryable; fallback skips providers on availability errors but stops on BadRequest, because a malformed request will fail everywhere and burning the chain hides your bug.
• Full jitter. Sleeping a random time in [0, min(cap, base·2ⁿ)] spreads out retries from many clients that failed at once. AWS's architecture blog compares jitter variants and is the usual reference. AWS Architecture Blog
• Streaming retries only before the first chunk. After tokens have reached the user, retrying would duplicate output. Mid-stream failure is surfaced to the caller, who can decide to restart visibly.
• Timeouts belong in the request object and are enforced by each adapter (connect + read timeouts on the HTTP client). Add an overall deadline across retries so retry × timeout can't exceed the caller's budget.
• Accounting uses the provider-reported usage, not your own estimate, and prices live in config (they change). For streams, usage usually arrives in a final event; read it there.
• Injected sleep and RNG make retry logic testable in microseconds.

Extensions: circuit breaker per provider (after K failures, skip it for T seconds, then half-open: let one probe through), hedged requests (send to a second provider if the first hasn't responded by p95 latency; cancel the loser), async version (async def complete, asyncio.timeout), structured output with validation and repair-retry, request/response hooks for tracing.

8b. Tool registry and tool-calling agent loop

Requirements: register tools with schemas; expose schemas to the model; execute the model's tool calls with argument validation; feed results and errors back; stop on a final answer or a max-steps guard; optional approval for side-effecting tools. Entities: ToolSpec (name, description, parameter schema), Tool (Command), ToolRegistry, ToolCall/ModelTurn (provider-neutral value objects), Agent, AgentResult.

from __future__ import annotations

import json
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable


class ToolError(Exception):
    pass


@dataclass(frozen=True)
class ToolSpec:
    name: str
    description: str
    params: dict[str, type]                 # minimal schema: name -> python type
    required: tuple[str, ...] = ()

    def validate(self, args: dict[str, Any]) -> None:
        missing = [k for k in self.required if k not in args]
        unknown = [k for k in args if k not in self.params]
        wrong = [k for k, v in args.items() if k in self.params and not isinstance(v, self.params[k])]
        if missing or unknown or wrong:
            raise ToolError(f"invalid args: missing={missing} unknown={unknown} wrong_type={wrong}")

    def to_json_schema(self) -> dict[str, Any]:
        tmap = {str: "string", int: "integer", float: "number", bool: "boolean"}
        return {"name": self.name, "description": self.description,
                "input_schema": {"type": "object",
                                 "properties": {k: {"type": tmap[t]} for k, t in self.params.items()},
                                 "required": list(self.required)}}


class Tool(ABC):                            # Command pattern: a request wrapped as an object
    spec: ToolSpec
    side_effecting: bool = False

    @abstractmethod
    def run(self, **kwargs: Any) -> str: ...


class Calculator(Tool):
    spec = ToolSpec("add", "Add two integers", {"a": int, "b": int}, ("a", "b"))

    def run(self, a: int, b: int) -> str:
        return str(a + b)


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        if tool.spec.name in self._tools:
            raise ValueError(f"duplicate tool {tool.spec.name}")
        self._tools[tool.spec.name] = tool

    def schemas(self) -> list[dict[str, Any]]:
        return [t.spec.to_json_schema() for t in self._tools.values()]

    def execute(self, name: str, args: dict[str, Any]) -> str:
        tool = self._tools.get(name)
        if tool is None:
            raise ToolError(f"unknown tool {name!r}; available: {sorted(self._tools)}")
        tool.spec.validate(args)
        return tool.run(**args)


# ---- the model's turn, provider-agnostic ----
@dataclass(frozen=True)
class ToolCall:
    id: str
    name: str
    args: dict[str, Any]


@dataclass(frozen=True)
class ModelTurn:
    text: str = ""
    tool_calls: tuple[ToolCall, ...] = ()


Model = Callable[[list[dict[str, Any]], list[dict[str, Any]]], ModelTurn]   # (messages, tool schemas) -> turn


@dataclass
class AgentResult:
    answer: str
    steps: int
    transcript: list[dict[str, Any]] = field(default_factory=list)
    stopped_reason: str = "final_answer"


class Agent:
    def __init__(self, model: Model, registry: ToolRegistry, max_steps: int = 8,
                 approve: Callable[[ToolCall], bool] = lambda c: True) -> None:
        self.model, self.reg, self.max_steps, self.approve = model, registry, max_steps, approve

    def run(self, user_msg: str) -> AgentResult:
        msgs: list[dict[str, Any]] = [{"role": "user", "content": user_msg}]
        for step in range(1, self.max_steps + 1):
            turn = self.model(msgs, self.reg.schemas())
            msgs.append({"role": "assistant", "content": turn.text,
                         "tool_calls": [c.__dict__ for c in turn.tool_calls]})
            if not turn.tool_calls:
                return AgentResult(turn.text, step, msgs)
            for call in turn.tool_calls:
                try:
                    if not self.approve(call):
                        raise ToolError("denied by policy")
                    out = self.reg.execute(call.name, call.args)
                    ok = True
                except ToolError as e:
                    out, ok = str(e), False    # feed the error back so the model can self-correct
                except Exception as e:         # tool bug: report, don't crash the loop
                    out, ok = f"tool crashed: {type(e).__name__}", False
                msgs.append({"role": "tool", "tool_call_id": call.id, "content": out, "ok": ok})
        return AgentResult("I couldn't finish within the step budget.", self.max_steps, msgs, "max_steps")


# ---- a scripted fake model for tests ----
def scripted(turns: list[ModelTurn]) -> Model:
    it = iter(turns)
    return lambda msgs, tools: next(it)


reg = ToolRegistry()
reg.register(Calculator())
model = scripted([
    ModelTurn(tool_calls=(ToolCall("c1", "add", {"a": 2, "b": "3"}),)),   # wrong type -> error fed back
    ModelTurn(tool_calls=(ToolCall("c2", "add", {"a": 2, "b": 3}),)),
    ModelTurn(text="2 + 3 = 5"),
])
res = Agent(model, reg).run("what is 2+3?")
assert res.answer == "2 + 3 = 5" and res.steps == 3
assert res.transcript[2]["ok"] is False and res.transcript[4]["content"] == "5"
looping = Agent(lambda m, t: ModelTurn(tool_calls=(ToolCall("x", "add", {"a": 1, "b": 1}),)), reg, max_steps=3)
assert looping.run("loop forever").stopped_reason == "max_steps"
json.dumps(reg.schemas())

Design decisions: errors from validation, policy denial and tool crashes are returned to the model as tool results so it can correct itself (the first scripted turn passes a string for an int and recovers). The max-steps guard is non-negotiable: models can loop. Tool calls are Command objects, which gives you logging, replay, approval gates and (for reversible tools) undo. The model is injected as a plain callable, so tests use a scripted fake.

Extensions: real JSON Schema validation (a library like jsonschema or Pydantic models instead of the minimal type map), parallel tool calls (run independent calls in a thread pool or asyncio.gather, preserving tool_call_id pairing), per-tool timeouts, a token budget as well as a step budget, truncating large tool outputs before they go back into context, human-in-the-loop approval for side-effecting tools (side_effecting=True → approve callback), idempotency keys for side-effecting tools so a retried step doesn't double-send an email.

8c. Conversation memory with pluggable strategies

Requirements: store conversation turns; build a context that fits a token budget; support strategies (full buffer, last-k window, running summary); allocate the window between system prompt, memory, user message and reserved output. Entities: Msg, TokenCounter (Protocol), MemoryStrategy (Strategy), BufferMemory, WindowMemory, SummaryMemory, PromptAssembler.

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Callable, Protocol


@dataclass(frozen=True)
class Msg:
    role: str
    content: str


class TokenCounter(Protocol):
    def count(self, text: str) -> int: ...


class ApproxCounter:
    def count(self, text: str) -> int:
        return max(1, len(text) // 4)          # rough heuristic; use the provider's tokenizer in prod


class MemoryStrategy(ABC):
    @abstractmethod
    def add(self, m: Msg) -> None: ...

    @abstractmethod
    def context(self, budget_tokens: int) -> list[Msg]: ...


class BufferMemory(MemoryStrategy):
    """Keep everything; trim oldest at read time to fit the budget."""

    def __init__(self, counter: TokenCounter) -> None:
        self.msgs: list[Msg] = []
        self.counter = counter

    def add(self, m: Msg) -> None:
        self.msgs.append(m)

    def context(self, budget_tokens: int) -> list[Msg]:
        out: list[Msg] = []
        used = 0
        for m in reversed(self.msgs):              # newest first
            t = self.counter.count(m.content)
            if used + t > budget_tokens:
                break
            out.append(m)
            used += t
        return list(reversed(out))


class WindowMemory(BufferMemory):
    """Last k messages only (then budget-trimmed)."""

    def __init__(self, counter: TokenCounter, k: int) -> None:
        super().__init__(counter)
        self.k = k

    def add(self, m: Msg) -> None:
        super().add(m)
        self.msgs = self.msgs[-self.k:]


class SummaryMemory(MemoryStrategy):
    """Recent turns verbatim; older turns folded into a running summary by an injected summarizer."""

    def __init__(self, counter: TokenCounter, summarize: Callable[[str, list[Msg]], str],
                 keep_recent: int = 4) -> None:
        self.counter, self.summarize, self.keep = counter, summarize, keep_recent
        self.summary = ""
        self.recent: list[Msg] = []

    def add(self, m: Msg) -> None:
        self.recent.append(m)
        if len(self.recent) > self.keep * 2:       # fold in batches, not every turn (cost)
            old, self.recent = self.recent[:-self.keep], self.recent[-self.keep:]
            self.summary = self.summarize(self.summary, old)

    def context(self, budget_tokens: int) -> list[Msg]:
        head = [Msg("system", f"Conversation so far: {self.summary}")] if self.summary else []
        used = sum(self.counter.count(m.content) for m in head)
        tail: list[Msg] = []
        for m in reversed(self.recent):
            t = self.counter.count(m.content)
            if used + t > budget_tokens:
                break
            tail.append(m)
            used += t
        return head + list(reversed(tail))


class PromptAssembler:
    """Allocates the context window: system prompt + memory + reserved output tokens."""

    def __init__(self, window: int, reserve_output: int, counter: TokenCounter) -> None:
        self.window, self.reserve, self.counter = window, reserve_output, counter

    def build(self, system: str, memory: MemoryStrategy, user: str) -> list[Msg]:
        fixed = self.counter.count(system) + self.counter.count(user)
        budget = self.window - self.reserve - fixed
        if budget < 0:
            raise ValueError("system prompt + user message exceed the window")
        return [Msg("system", system), *memory.context(budget), Msg("user", user)]


fake_summarizer = lambda prev, msgs: (prev + " | " if prev else "") + f"{len(msgs)} older msgs"
mem = SummaryMemory(ApproxCounter(), fake_summarizer, keep_recent=2)
for i in range(6):
    mem.add(Msg("user", f"message number {i}"))
ctx = PromptAssembler(window=60, reserve_output=20, counter=ApproxCounter()).build("be brief", mem, "and now?")
assert ctx[0].role == "system" and ctx[1].content.startswith("Conversation so far") and ctx[-1].content == "and now?"
w = WindowMemory(ApproxCounter(), k=2)
for i in range(5):
    w.add(Msg("user", str(i)))
assert [m.content for m in w.context(100)] == ["3", "4"]

Design decisions: the token counter is injected (the 4-characters-per-token heuristic is only a placeholder; use the provider's tokenizer or token-counting endpoint for real budgets); the summariser is injected (it's an LLM call, so tests use a fake); the summary memory folds old turns in batches rather than every turn to control cost; the assembler reserves output tokens before filling memory, so long histories can't squeeze out the answer.

Extensions: retrieval-based memory (embed turns, retrieve relevant past turns via the vector store in 8f), entity/fact memory (extract facts into a key-value store), per-user persistence via a Repository, thread safety when one conversation can receive concurrent messages (lock per conversation ID). B3 covers memory architectures in depth.

8d. Token-aware rate limiter / request scheduler with per-tenant quotas

Requirements: providers limit requests per minute and tokens per minute; your platform has tenants on different tiers; higher-priority tenants go first; one noisy tenant mustn't block others; you don't know output tokens until the response arrives. Entities: TenantQuota, token buckets per tenant (RPM and TPM) plus a global provider TPM bucket, a priority queue of requests, TokenAwareScheduler.

from __future__ import annotations

import heapq
import itertools
import threading
import time
from dataclasses import dataclass, field
from typing import Callable


@dataclass
class TenantQuota:
    rpm: int                    # requests per minute
    tpm: int                    # tokens per minute (input + max output)
    priority: int = 1           # lower = more important


class _Bucket:
    def __init__(self, capacity: float, per_second: float, now: float) -> None:
        self.cap, self.rate, self.tokens, self.t = capacity, per_second, capacity, now

    def refill(self, now: float) -> None:
        self.tokens = min(self.cap, self.tokens + (now - self.t) * self.rate)
        self.t = now

    def wait_time(self, need: float) -> float:
        return 0.0 if self.tokens >= need else (need - self.tokens) / self.rate


@dataclass(order=True)
class _Queued:
    priority: int
    seq: int
    tenant: str = field(compare=False)
    est_tokens: int = field(compare=False)
    job: Callable[[], object] = field(compare=False)


class TokenAwareScheduler:
    """Per-tenant RPM + TPM buckets, a global provider TPM bucket, and a priority queue.
    Charges an *estimate* up front (prompt tokens + max_tokens) and refunds the unused part."""

    def __init__(self, quotas: dict[str, TenantQuota], global_tpm: int,
                 clock: Callable[[], float] = time.monotonic) -> None:
        self.clock = clock
        now = clock()
        self.quotas = quotas
        self.rpm = {t: _Bucket(q.rpm, q.rpm / 60, now) for t, q in quotas.items()}
        self.tpm = {t: _Bucket(q.tpm, q.tpm / 60, now) for t, q in quotas.items()}
        self.global_tpm = _Bucket(global_tpm, global_tpm / 60, now)
        self.q: list[_Queued] = []
        self.seq = itertools.count()
        self.lock = threading.Lock()

    def submit(self, tenant: str, est_tokens: int, job: Callable[[], object]) -> None:
        q = self.quotas[tenant]
        if est_tokens > q.tpm:
            raise ValueError("request can never fit this tenant's TPM quota")
        with self.lock:
            heapq.heappush(self.q, _Queued(q.priority, next(self.seq), tenant, est_tokens, job))

    def try_dispatch(self) -> tuple[str, int] | None:
        """Pop the highest-priority request whose buckets all have room. Returns (tenant, tokens) or None."""
        with self.lock:
            now = self.clock()
            for b in (*self.rpm.values(), *self.tpm.values(), self.global_tpm):
                b.refill(now)
            skipped: list[_Queued] = []
            chosen = None
            while self.q:
                item = heapq.heappop(self.q)
                t = item.tenant
                if (self.rpm[t].tokens >= 1 and self.tpm[t].tokens >= item.est_tokens
                        and self.global_tpm.tokens >= item.est_tokens):
                    self.rpm[t].tokens -= 1
                    self.tpm[t].tokens -= item.est_tokens
                    self.global_tpm.tokens -= item.est_tokens
                    chosen = item
                    break
                skipped.append(item)           # tenant over quota: don't let it block others
            for s in skipped:
                heapq.heappush(self.q, s)
        if chosen is None:
            return None
        chosen.job()
        return chosen.tenant, chosen.est_tokens

    def settle(self, tenant: str, estimated: int, actual: int) -> None:
        """After the response arrives, refund over-estimation (never below zero usage)."""
        refund = max(0, estimated - actual)
        with self.lock:
            for b in (self.tpm[tenant], self.global_tpm):
                b.tokens = min(b.cap, b.tokens + refund)


now = [0.0]
s = TokenAwareScheduler({"free": TenantQuota(rpm=60, tpm=1_000, priority=2),
                         "pro": TenantQuota(rpm=600, tpm=10_000, priority=1)},
                        global_tpm=8_000, clock=lambda: now[0])
ran: list[str] = []
s.submit("free", 800, lambda: ran.append("free-1"))
s.submit("free", 800, lambda: ran.append("free-2"))      # exceeds free TPM until refill
s.submit("pro", 2_000, lambda: ran.append("pro-1"))
while s.try_dispatch():
    pass
assert ran == ["pro-1", "free-1"]                        # priority first; free-2 waits for quota
s.settle("free", estimated=800, actual=300)              # refund 500 unused tokens
now[0] += 6.0                                            # +100 tokens refill for free tier
s.try_dispatch()
assert ran[-1] == "free-2"

Design decisions: charge an estimate up front (prompt tokens + max_tokens) so concurrent requests can't jointly overshoot, then settle with actual usage and refund the difference. A request is only dispatched if all its buckets have room, and only then are they all debited (no partial consumption). Items from over-quota tenants are skipped and re-queued, so they don't block the head of the queue. Reject at submit time any request that can never fit.

Extensions: weighted fair queuing across tenants instead of strict priority (strict priority can starve the free tier), deadlines (drop requests whose caller has given up), adaptive limits from provider rate-limit response headers, a waiting dispatcher thread that sleeps until the earliest bucket has room (wait_time), distributed state in Redis.

8e. Prompt template engine with versioning

Requirements: named templates with variables; strict rendering (missing or extra variables are errors); immutable versions; labels like prod and canary that can move (deploy, rollback) without code changes; a content hash for traceability in logs. Entities: PromptTemplate (immutable value object), PromptRegistry.

from __future__ import annotations

import hashlib
import string
from dataclasses import dataclass, field
from datetime import datetime, timezone


class TemplateError(Exception):
    pass


@dataclass(frozen=True)
class PromptTemplate:
    name: str
    version: int
    body: str                                      # uses $var placeholders (string.Template)
    model_hint: str | None = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    @property
    def variables(self) -> set[str]:
        return {m.group("named") or m.group("braced")
                for m in string.Template.pattern.finditer(self.body)
                if m.group("named") or m.group("braced")}

    @property
    def content_hash(self) -> str:
        return hashlib.sha256(self.body.encode()).hexdigest()[:12]

    def render(self, **values: object) -> str:
        missing = self.variables - values.keys()
        extra = values.keys() - self.variables
        if missing or extra:
            raise TemplateError(f"{self.name}@v{self.version}: missing={sorted(missing)} extra={sorted(extra)}")
        return string.Template(self.body).substitute(**{k: str(v) for k, v in values.items()})


class PromptRegistry:
    """Immutable versions + movable labels (e.g. 'prod', 'canary'). Rollback = move a label."""

    def __init__(self) -> None:
        self._versions: dict[str, list[PromptTemplate]] = {}
        self._labels: dict[tuple[str, str], int] = {}

    def publish(self, name: str, body: str, model_hint: str | None = None) -> PromptTemplate:
        versions = self._versions.setdefault(name, [])
        if versions and versions[-1].body == body:
            return versions[-1]                    # idempotent re-publish
        t = PromptTemplate(name, len(versions) + 1, body, model_hint)
        versions.append(t)
        return t

    def set_label(self, name: str, label: str, version: int) -> None:
        if not 1 <= version <= len(self._versions.get(name, [])):
            raise KeyError(f"{name}@v{version} does not exist")
        self._labels[(name, label)] = version

    def get(self, name: str, *, version: int | None = None, label: str = "prod") -> PromptTemplate:
        v = version or self._labels.get((name, label))
        if v is None:
            raise KeyError(f"no '{label}' label for {name}")
        return self._versions[name][v - 1]


reg = PromptRegistry()
v1 = reg.publish("summarize", "Summarize for a $audience reader:\n$text")
v2 = reg.publish("summarize", "You are concise. Summarize for a ${audience} reader in $n bullets:\n$text")
reg.set_label("summarize", "prod", 1)
reg.set_label("summarize", "canary", 2)
assert reg.get("summarize").version == 1 and reg.get("summarize", label="canary").variables == {"audience", "n", "text"}
assert "3 bullets" in reg.get("summarize", label="canary").render(audience="exec", n=3, text="...")
try:
    v1.render(audience="exec")
except TemplateError as e:
    assert "missing=['text']" in str(e)

Design decisions: versions are append-only and never edited, so a log line saying "summarize@v2 (hash abc…)" always means the same text. Labels are pointers, so rollback is moving a pointer. Rendering is strict, because a silently empty variable produces a subtly wrong prompt that's hard to debug. The standard library's string.Template is enough here and avoids executing arbitrary template logic; reach for a full template engine only if you need loops and conditionals.

Extensions: A/B routing by tenant hash between two labels, evaluation results attached to each version (block promotion to prod if evals regress), per-model variants of one template, persistence via a Repository, escaping of user-supplied values (delimit untrusted input clearly to reduce prompt-injection risk).

8f. Vector-store interface with pluggable backends and metadata filtering

Requirements: upsert documents with embeddings and metadata; top-k similarity search with filters; delete; swap the backend (in-memory, a Postgres extension, a managed vector DB) without touching application code; tenant isolation. Entities: Document, Hit, a small filter AST (Eq, In, And), the VectorStore interface, InMemoryVectorStore, Retriever (facade).

from __future__ import annotations

import math
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass(frozen=True)
class Document:
    id: str
    text: str
    embedding: tuple[float, ...]
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class Hit:
    doc: Document
    score: float


# ---- a tiny, backend-neutral filter language (translate to each backend's syntax) ----
@dataclass(frozen=True)
class Eq:
    field: str
    value: Any


@dataclass(frozen=True)
class In:
    field: str
    values: tuple[Any, ...]


@dataclass(frozen=True)
class And:
    clauses: tuple["Filter", ...]


Filter = Eq | In | And


def matches(f: Filter | None, md: dict[str, Any]) -> bool:
    if f is None:
        return True
    if isinstance(f, Eq):
        return md.get(f.field) == f.value
    if isinstance(f, In):
        return md.get(f.field) in f.values
    return all(matches(c, md) for c in f.clauses)


class VectorStore(ABC):
    @abstractmethod
    def upsert(self, docs: list[Document]) -> None: ...

    @abstractmethod
    def query(self, vector: tuple[float, ...], k: int = 5, where: Filter | None = None) -> list[Hit]: ...

    @abstractmethod
    def delete(self, ids: list[str]) -> None: ...


def cosine(a: tuple[float, ...], b: tuple[float, ...]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
    return dot / (na * nb) if na and nb else 0.0


class InMemoryVectorStore(VectorStore):
    """Brute-force exact search. A pgvector/managed-DB adapter implements the same interface."""

    def __init__(self, dim: int) -> None:
        self.dim = dim
        self._docs: dict[str, Document] = {}

    def upsert(self, docs: list[Document]) -> None:
        for d in docs:
            if len(d.embedding) != self.dim:
                raise ValueError(f"{d.id}: expected dim {self.dim}, got {len(d.embedding)}")
            self._docs[d.id] = d

    def query(self, vector: tuple[float, ...], k: int = 5, where: Filter | None = None) -> list[Hit]:
        hits = [Hit(d, cosine(vector, d.embedding)) for d in self._docs.values() if matches(where, d.metadata)]
        return sorted(hits, key=lambda h: h.score, reverse=True)[:k]   # filter BEFORE top-k

    def delete(self, ids: list[str]) -> None:
        for i in ids:
            self._docs.pop(i, None)


class Retriever:
    """Facade the app uses: embeds the query, enforces tenant isolation, calls the store."""

    def __init__(self, store: VectorStore, embed: Callable[[str], tuple[float, ...]]) -> None:
        self.store, self.embed = store, embed

    def search(self, tenant: str, query: str, k: int = 3, where: Filter | None = None) -> list[Hit]:
        scope: Filter = Eq("tenant", tenant)
        f = And((scope, where)) if where else scope        # tenant filter is never optional
        return self.store.query(self.embed(query), k, f)


def toy_embed(text: str) -> tuple[float, ...]:
    return (text.count("cat") + 0.1, text.count("dog") + 0.1, len(text) / 100)


store = InMemoryVectorStore(dim=3)
store.upsert([Document(str(i), t, toy_embed(t), {"tenant": tn, "lang": lg})
              for i, (t, tn, lg) in enumerate([("cat cat", "a", "en"), ("dog", "a", "en"),
                                               ("cat", "b", "en"), ("le cat", "a", "fr")])])
r = Retriever(store, toy_embed)
hits = r.search("a", "cat", k=5, where=In("lang", ("en",)))
assert [h.doc.id for h in hits] == ["0", "1"]       # tenant b's doc and the fr doc never appear

Design decisions: the filter is a backend-neutral AST (a Specification/Composite); each backend adapter translates it into its own filter syntax. That avoids leaking one vendor's query language into application code. Filtering happens before taking top-k. With approximate indexes, post-filtering can return fewer than k results, so real backends need pre-filtering or over-fetching, which is worth mentioning. The Retriever always adds the tenant filter, so a caller can't forget it. Dimension checks catch the classic bug of mixing embedding models.

Extensions: hybrid search (keyword BM25 + vector, fused with reciprocal rank fusion), batch upserts, namespaces per tenant, re-ranking stage, embedding-model version stored on each document so you can re-embed incrementally. B2 · Retrieval goes deeper.

8g. Workflow / DAG executor for agent steps

Requirements: steps with dependencies; run independent steps in parallel; pass each step the outputs of its dependencies; retry failing steps; skip steps whose upstream failed; detect cycles and unknown dependencies before running. Entities: Step, StepStatus, RunResult, DagExecutor.

Python's standard library includes graphlib.TopologicalSorter, which supports exactly this incremental pattern: prepare(), then loop on get_ready() and done() while is_active(), and it raises CycleError for cycles. Python docs: graphlib In an interview, be ready to write Kahn's algorithm yourself: compute in-degrees, start with the zero in-degree nodes, and decrement neighbours as each node finishes; if nodes remain at the end, there's a cycle.

from __future__ import annotations

import time
from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
from dataclasses import dataclass, field
from enum import Enum
from graphlib import CycleError, TopologicalSorter
from typing import Any, Callable


class StepStatus(Enum):
    PENDING = "pending"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    SKIPPED = "skipped"         # an upstream dependency failed


@dataclass
class Step:
    name: str
    fn: Callable[[dict[str, Any]], Any]          # receives outputs of its dependencies
    deps: tuple[str, ...] = ()
    max_retries: int = 2
    backoff_s: float = 0.0


@dataclass
class RunResult:
    status: dict[str, StepStatus] = field(default_factory=dict)
    outputs: dict[str, Any] = field(default_factory=dict)
    errors: dict[str, str] = field(default_factory=dict)


class DagExecutor:
    def __init__(self, steps: list[Step], max_workers: int = 4) -> None:
        self.steps = {s.name: s for s in steps}
        unknown = {d for s in steps for d in s.deps if d not in self.steps}
        if unknown:
            raise ValueError(f"unknown dependencies: {unknown}")
        self.max_workers = max_workers
        self._graph = {s.name: set(s.deps) for s in steps}
        try:
            tuple(TopologicalSorter(self._graph).static_order())    # validate up front
        except CycleError as e:
            raise ValueError(f"cycle detected: {e.args[1]}") from None

    def _run_step(self, step: Step, inputs: dict[str, Any]) -> Any:
        for attempt in range(step.max_retries + 1):
            try:
                return step.fn(inputs)
            except Exception:
                if attempt == step.max_retries:
                    raise
                time.sleep(step.backoff_s * 2 ** attempt)

    def run(self) -> RunResult:
        res = RunResult({n: StepStatus.PENDING for n in self.steps})
        ts = TopologicalSorter(self._graph)
        ts.prepare()
        running: dict[Future[Any], str] = {}
        with ThreadPoolExecutor(self.max_workers) as pool:
            while ts.is_active():
                for name in ts.get_ready():                       # all deps are done
                    step = self.steps[name]
                    if any(res.status[d] is not StepStatus.SUCCEEDED for d in step.deps):
                        res.status[name] = StepStatus.SKIPPED
                        ts.done(name)
                        continue
                    inputs = {d: res.outputs[d] for d in step.deps}
                    running[pool.submit(self._run_step, step, inputs)] = name
                if not running:
                    continue                                       # only skips this round
                finished, _ = wait(running, return_when=FIRST_COMPLETED)
                for fut in finished:
                    name = running.pop(fut)
                    try:
                        res.outputs[name] = fut.result()
                        res.status[name] = StepStatus.SUCCEEDED
                    except Exception as e:
                        res.status[name] = StepStatus.FAILED
                        res.errors[name] = repr(e)
                    ts.done(name)
        return res


flaky_calls = {"n": 0}


def flaky(_: dict[str, Any]) -> str:
    flaky_calls["n"] += 1
    if flaky_calls["n"] < 2:
        raise TimeoutError("transient")
    return "web results"


dag = DagExecutor([
    Step("plan", lambda i: "plan"),
    Step("search_web", flaky, deps=("plan",)),
    Step("search_docs", lambda i: "doc results", deps=("plan",)),
    Step("broken", lambda i: 1 / 0, deps=("plan",), max_retries=0),
    Step("after_broken", lambda i: "never", deps=("broken",)),
    Step("synthesize", lambda i: f"{i['search_web']} + {i['search_docs']}", deps=("search_web", "search_docs")),
])
out = dag.run()
assert out.outputs["synthesize"] == "web results + doc results"
assert out.status["broken"] is StepStatus.FAILED and out.status["after_broken"] is StepStatus.SKIPPED
try:
    DagExecutor([Step("a", lambda i: 1, deps=("b",)), Step("b", lambda i: 1, deps=("a",))])
except ValueError as e:
    assert "cycle" in str(e)

Design decisions: validate the graph at construction (fail fast), then schedule dynamically: a step starts as soon as its dependencies are done, not in fixed "levels", which maximises parallelism. Failure propagates as SKIPPED downstream rather than aborting unrelated branches. Retries live in the step runner with exponential backoff.

Extensions: per-step timeouts, cancellation of the whole run, conditional edges (run a branch only if a predicate on upstream output holds), persistence of step results so a crashed run resumes from the last completed step (this is what durable workflow engines provide), idempotent steps so resumption is safe, and an async version for I/O-bound LLM steps.

8h. Semantic cache

Requirements: avoid paying for an LLM call when an equivalent prompt was answered recently; exact match first, then semantic similarity above a threshold; scope entries so different tenants, models, prompt versions or parameters never share answers; TTL; bounded size; opt out for personal or time-sensitive prompts. Entities: CacheEntry, SemanticCache, CachedLLM (Decorator/Proxy over the model).

from __future__ import annotations

import math
import threading
import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Callable


@dataclass
class CacheEntry:
    prompt: str
    embedding: tuple[float, ...]
    response: str
    created: float
    scope: str                  # tenant + model + prompt-template version + params


def _cos(a: tuple[float, ...], b: tuple[float, ...]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    n = math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b))
    return dot / n if n else 0.0


class SemanticCache:
    """Exact-match fast path, then nearest-neighbour over embeddings within the same scope.
    Linear scan for clarity; a real one delegates the similarity search to a VectorStore."""

    def __init__(self, embed: Callable[[str], tuple[float, ...]], threshold: float = 0.92,
                 ttl_s: float = 3600, max_entries: int = 10_000, clock: Callable[[], float] = time.time) -> None:
        self.embed, self.threshold, self.ttl, self.max = embed, threshold, ttl_s, max_entries
        self.clock = clock
        self._exact: OrderedDict[tuple[str, str], CacheEntry] = OrderedDict()
        self._lock = threading.Lock()
        self.hits = self.semantic_hits = self.misses = 0

    @staticmethod
    def _norm(p: str) -> str:
        return " ".join(p.lower().split())

    def get(self, prompt: str, scope: str) -> str | None:
        now = self.clock()
        key = (scope, self._norm(prompt))
        with self._lock:
            e = self._exact.get(key)
            if e and now - e.created < self.ttl:
                self._exact.move_to_end(key)
                self.hits += 1
                return e.response
        q = self.embed(prompt)                                      # embed outside the lock (slow)
        with self._lock:
            best, best_score = None, -1.0
            for e in self._exact.values():
                if e.scope != scope or now - e.created >= self.ttl:
                    continue
                s = _cos(q, e.embedding)
                if s > best_score:
                    best, best_score = e, s
            if best and best_score >= self.threshold:
                self.semantic_hits += 1
                return best.response
            self.misses += 1
            return None

    def put(self, prompt: str, scope: str, response: str) -> None:
        e = CacheEntry(prompt, self.embed(prompt), response, self.clock(), scope)
        with self._lock:
            self._exact[(scope, self._norm(prompt))] = e
            while len(self._exact) > self.max:
                self._exact.popitem(last=False)                     # LRU eviction


class CachedLLM:
    def __init__(self, llm: Callable[[str], str], cache: SemanticCache,
                 cacheable: Callable[[str], bool] = lambda p: True) -> None:
        self.llm, self.cache, self.cacheable = llm, cache, cacheable

    def ask(self, prompt: str, scope: str) -> str:
        if not self.cacheable(prompt):                              # e.g. personal or time-sensitive
            return self.llm(prompt)
        if (hit := self.cache.get(prompt, scope)) is not None:
            return hit
        out = self.llm(prompt)
        self.cache.put(prompt, scope, out)
        return out


def toy_embed(t: str) -> tuple[float, ...]:
    words = set(t.lower().replace("?", "").split())
    vocab = ["refund", "policy", "return", "shipping", "time", "what", "is", "your"]
    return tuple(1.0 if w in words else 0.0 for w in vocab)


calls: list[str] = []
llm = CachedLLM(lambda p: calls.append(p) or f"answer to {p}", SemanticCache(toy_embed, threshold=0.85))
llm.ask("What is your refund policy?", "acme:v3")
llm.ask("what is your  REFUND policy?", "acme:v3")      # exact hit after normalisation
llm.ask("your refund policy is what", "acme:v3")        # semantic hit
llm.ask("What is your refund policy?", "globex:v3")     # different scope: miss
assert len(calls) == 2

Design decisions: the scope key is the safety feature: it includes tenant, model, prompt-template version and generation parameters. The threshold is a precision/recall knob, and a false hit returns a wrong answer confidently, so tune it on labelled pairs and start strict. The embedding call happens outside the lock (it's slow). Normalised exact matching catches the cheap cases before any embedding. The linear scan is for clarity; a real implementation delegates nearest-neighbour search to the vector store from 8f.

Extensions: invalidation when source documents change (tag entries with the document IDs used to answer), per-route thresholds, caching only for temperature=0 requests, hit-rate and false-hit monitoring (sample hits for human or LLM-judge review), and provider-side prompt caching, which is a different thing (it caches the prompt prefix computation, not the answer).

9. Testing your design

You rarely have time for a full test suite in the round, but two or three well-chosen tests are a strong signal, and a design that's easy to test is a design with good seams. Aim for: one happy-path test, one boundary test (exactly at capacity, exactly at TTL), one error test, and, if concurrency is in scope, one invariant test under contention.

Unit tests in the round

Use whatever is fastest: plain assert statements in a main block (as every snippet on this page does) are fine in a machine-coding round; unittest or pytest look more professional if you have a minute. The key enabler is the injected clock.

import unittest
from datetime import datetime, timedelta


class FakeClock:
    def __init__(self, t: datetime) -> None:
        self.t = t

    def now(self) -> datetime:
        return self.t

    def advance(self, **kw: float) -> None:
        self.t += timedelta(**kw)


class Session:
    def __init__(self, clock: FakeClock, ttl: timedelta) -> None:
        self.clock, self.ttl = clock, ttl
        self.started = clock.now()

    def expired(self) -> bool:
        return self.clock.now() - self.started >= self.ttl


class SessionTest(unittest.TestCase):
    def setUp(self) -> None:
        self.clock = FakeClock(datetime(2026, 1, 1))
        self.s = Session(self.clock, timedelta(minutes=10))

    def test_not_expired_before_ttl(self) -> None:
        self.clock.advance(minutes=9, seconds=59)
        self.assertFalse(self.s.expired())

    def test_expired_exactly_at_ttl(self) -> None:          # boundary cases are where bugs live
        self.clock.advance(minutes=10)
        self.assertTrue(self.s.expired())


if __name__ == "__main__":
    unittest.main(argv=["x"], exit=False)

Fakes vs mocks (vs stubs)

Fowler's essay is the standard explanation of the difference between state verification (fakes/stubs) and behaviour verification (mocks), and of the trade-offs. Fowler: Mocks Aren't Stubs In LLD rounds, prefer fakes for your own interfaces (they double as documentation of the contract) and mocks for verifying calls to external collaborators. Use Mock(spec=RealClass) so a typo or a renamed method fails the test instead of silently passing. Python docs: unittest.mock

from unittest.mock import Mock


class OrderService:
    def __init__(self, payments, notifier) -> None:
        self.payments, self.notifier = payments, notifier

    def checkout(self, user: str, cents: int) -> bool:
        if not self.payments.charge(user, cents):
            return False
        self.notifier.notify(user, "paid")
        return True


# Mock: verify an interaction happened (or didn't)
payments = Mock()
payments.charge.return_value = False
notifier = Mock()
assert OrderService(payments, notifier).checkout("u1", 500) is False
payments.charge.assert_called_once_with("u1", 500)
notifier.notify.assert_not_called()

# Mock(spec=...) catches typos and calls to methods that don't exist on the real class
class RealNotifier:
    def notify(self, to: str, msg: str) -> None: ...

strict = Mock(spec=RealNotifier)
try:
    strict.notfy("u", "x")          # typo -> AttributeError instead of a silently passing test
except AttributeError:
    pass

Testing concurrency

Concurrency bugs are probabilistic, so tests can show the presence of races but never prove their absence. Make them as sharp as you can:

• Assert invariants, not interleavings: "exactly one booking for the seat", "never oversold", "sum of balances is constant".
• Maximise contention: many threads, a Barrier to release them together, a tight loop, and a small resource (one seat, capacity 100).
• Detect deadlocks: join with a timeout and fail if threads are still alive.
• Surface worker exceptions: an exception in a thread doesn't fail the test by default; collect and re-raise.
• Make time deterministic: a frozen fake clock removes timing as a source of flakiness (the rate-limiter test does this).
• Design so the critical logic is a pure function you can test single-threaded, leaving only the locking to the stress test.

import threading


def run_concurrently(fn, n_threads: int, iterations: int) -> None:
    """Release all threads at once with a Barrier to maximise interleaving."""
    barrier = threading.Barrier(n_threads)
    errors: list[BaseException] = []

    def body() -> None:
        try:
            barrier.wait()
            for _ in range(iterations):
                fn()
        except BaseException as e:            # surface failures from worker threads
            errors.append(e)

    ts = [threading.Thread(target=body) for _ in range(n_threads)]
    for t in ts: t.start()
    for t in ts: t.join(timeout=10)
    assert not any(t.is_alive() for t in ts), "possible deadlock: threads still running"
    if errors:
        raise errors[0]


class Inventory:
    def __init__(self, stock: int) -> None:
        self.stock, self.sold = stock, 0
        self._lock = threading.Lock()

    def buy(self) -> bool:
        with self._lock:
            if self.stock == 0:
                return False
            self.stock -= 1
            self.sold += 1
            return True


inv = Inventory(stock=500)
run_concurrently(inv.buy, n_threads=16, iterations=100)
assert inv.sold == 500 and inv.stock == 0      # invariant: never oversell

10. Machine-coding round tactics

Project skeleton

Keep it small and conventional. One file is fine for 60 minutes; for 90–120 minutes a package like this makes extension and review easy:

What to build first

1. Models (enums, dataclasses): 5 minutes. They force decisions about state.
2. The service with the one or two core use cases, with no strategies yet (hard-code the first pricing rule inline if needed).
3. A demo driver in main.py that exercises the core use case and prints the expected output. Now you have something runnable. Commit mentally: this is your safety net.
4. Extract the interfaces where variation was mentioned, and move the hard-coded rule into the first strategy.
5. Remaining use cases, in order of importance from the problem statement.
6. Edge cases and errors: custom exceptions, input validation.
7. Concurrency if in scope, then tests.
8. Bonus requirements only after everything above works.

Time management

• Set checkpoints: by 40% of the time, the core flow runs end-to-end. By 75%, all mandatory requirements work. The rest is polish, extension and tests.
• If you're stuck on a detail (a tricky algorithm, a parsing edge case), stub it with a simple version and a TODO, and say so. Come back if time allows.
• Run the code often. Debugging 300 lines written blind is how machine-coding rounds are lost.
• Don't gold-plate: no logging framework, no config files, no CLI library, unless asked.
• Use the language features you know cold. A machine-coding round is the wrong time to try a new library.

Using an AI assistant or IDE in the round

Some companies now allow, or even expect, AI coding assistants in machine-coding rounds; many still forbid them. Ask at the start. If allowed, use it for boilerplate (dataclasses, test scaffolding), keep the design decisions visibly yours, and read every line it writes, because you'll be asked to explain and extend it. B7 covers working with coding agents in general.

Presenting at the end

Reserve 5–7 minutes. A good walkthrough takes about three minutes and follows this structure:

1. Run the demo (or tests) first, so they see it works.
2. Scope recap: "I built X, Y, Z; I left out W as agreed."
3. Structure tour: models → interfaces → service, one sentence each.
4. Two design decisions with their trade-offs: "Pricing is a Strategy, so the weekend rule is one new class. One lock per show, because…"
5. Known limitations and next steps: "Holds expire lazily; with more time I'd add a sweeper and persistence via the repository interface." Raising these yourself scores better than having them pointed out.
6. Invite the extension question. Then answer it by pointing at the seam you prepared.

11. Interview question bank