Design an inventory management system

01The design, in the six-step framework

1. Requirements and use cases

An inventory system answers one question correctly under pressure: how many of this item can I promise right now? Everything else (warehouses, reorders, reports) hangs off that promise.

• An operator receives stock into a warehouse and adjusts counts after audits.
• An order service reserves units for an order, confirms the reservation on payment or releases it on cancel or timeout.
• A planner reads availability per SKU across warehouses: on hand, reserved, available.
• The system raises a reorder event when available stock falls under a threshold, once per crossing.
• The system prefers a warehouse for a reservation by a rule (nearest, most stock).

Out of scope: supplier purchase orders, pricing and physical picking. Non-functional constraints: no oversell under concurrent reservations; the threshold rule and warehouse choice are pluggable; every stock movement is recorded for audit.

2. Core entities and relationships

• SKU: the product identity; a value.
• StockLevel: on-hand and reserved counts for one SKU in one warehouse, with its own lock.
• Warehouse: its stock levels by SKU and a location.
• Reservation: SKU, quantity, warehouse, expiry and a state (held, confirmed, released).
• ReorderRule: a specification over a stock level that says whether to raise a reorder.
• StockEvent: an immutable record of each movement; listeners consume them.
• InventoryService: the orchestrator.

A warehouse composes its stock levels; a level is meaningless without its warehouse and SKU. The service composes reservations (it issued them) and aggregates warehouses (they are long-lived and shared with other services).

3. Interfaces and key methods

class StockLevel:
    def available(self) -> int: ...                 # on_hand - reserved
    def try_reserve(self, qty: int) -> bool:
        """Atomically reserve if available >= qty; never partial."""
    def confirm(self, qty: int) -> None:
        """Reserved units leave the building: reserved -= qty, on_hand -= qty."""
    def release(self, qty: int) -> None: ...

class WarehouseChooser(Protocol):
    def order(self, sku: SKU, qty: int, warehouses: list[Warehouse]) -> list[Warehouse]:
        """Warehouses to try, in preference order."""

class ReorderRule(Protocol):
    def should_reorder(self, level: StockLevel) -> bool: ...

class InventoryService:
    def receive(self, warehouse_id: str, sku: SKU, qty: int) -> None: ...
    def reserve(self, sku: SKU, qty: int, ttl: float = 900) -> Reservation:
        """Try warehouses in order; raise OutOfStock if none can hold the whole quantity."""
    def confirm(self, reservation_id: str) -> None: ...
    def release(self, reservation_id: str) -> None: ...
    def expire(self, now: float) -> int:
        """Release every held reservation past its expiry; returns how many."""
    def availability(self, sku: SKU) -> dict[str, tuple[int, int, int]]: ...

Invariants: 0 <= reserved <= on_hand for every level; the sum of held reservation quantities for a (warehouse, SKU) equals that level's reserved; a reservation changes state at most once; a reorder event is emitted once when availability crosses below the threshold, not on every reservation below it.

4. Design patterns and the choices behind them

A lock per stock level

The race is two orders reserving the last unit. One global lock is correct and serialises every SKU in every warehouse; a lock per StockLevel lets unrelated SKUs proceed in parallel and keeps the critical section to a compare and an add. try_reserve does the check and the increment inside the lock; that is the whole correctness argument.

Reserve, confirm, release as a tiny state machine

A reservation is HELD, then CONFIRMED or RELEASED, and expiry is a release by the clock. Making it explicit prevents the double-release that frees stock twice; the use cases map onto exactly these three methods.

Events for reorders and audit

Every movement emits a StockEvent. The audit log is a listener; the reorder detector is a listener that applies a ReorderRule and remembers whether it already fired for that level. Keeping the rule out of StockLevel means thresholds per SKU or per season are configuration.

Warehouse choice as a strategy

Nearest to the customer, most stock, cheapest: the preference order is a strategy returning a list to try. The service walks that list and takes the first warehouse that can hold the full quantity; splitting across warehouses is an extension, not a default.

5. Core flow and code

1. reserve(sku, 3) asks the chooser for an ordered list of warehouses.
2. For each, it calls level.try_reserve(3); the first success creates a HELD reservation with an expiry and emits a reserve event.
3. The reorder listener receives the event, checks the rule on that level and emits a reorder once if availability has just dropped below the threshold.
4. confirm moves units out of on-hand; release or expire gives them back.

from __future__ import annotations
import itertools, threading, time
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Optional, Protocol


class OutOfStock(Exception): ...
class IllegalTransition(Exception): ...


@dataclass(frozen=True)
class SKU:
    code: str


class StockLevel:
    def __init__(self) -> None:
        self.on_hand = 0
        self.reserved = 0
        self.lock = threading.Lock()

    def available(self) -> int:
        return self.on_hand - self.reserved

    def add(self, qty: int) -> None:
        with self.lock:
            self.on_hand += qty

    def try_reserve(self, qty: int) -> bool:
        with self.lock:
            if self.available() < qty:
                return False
            self.reserved += qty
            return True

    def confirm(self, qty: int) -> None:
        with self.lock:
            self.reserved -= qty
            self.on_hand -= qty

    def release(self, qty: int) -> None:
        with self.lock:
            self.reserved -= qty


@dataclass
class Warehouse:
    id: str
    location: tuple[float, float]
    levels: dict[SKU, StockLevel] = field(default_factory=dict)
    def level(self, sku: SKU) -> StockLevel:
        return self.levels.setdefault(sku, StockLevel())


class ResState(Enum):
    HELD = "held"; CONFIRMED = "confirmed"; RELEASED = "released"


@dataclass
class Reservation:
    id: str
    sku: SKU
    qty: int
    warehouse_id: str
    expires_at: float
    state: ResState = ResState.HELD


@dataclass(frozen=True)
class StockEvent:
    kind: str
    sku: SKU
    qty: int
    warehouse_id: str


class WarehouseChooser(Protocol):
    def order(self, sku: SKU, qty: int, warehouses: list[Warehouse]) -> list[Warehouse]: ...


class MostStockFirst:
    def order(self, sku: SKU, qty: int, warehouses: list[Warehouse]) -> list[Warehouse]:
        return sorted(warehouses, key=lambda w: -w.level(sku).available())


class ReorderRule(Protocol):
    def should_reorder(self, level: StockLevel) -> bool: ...


@dataclass
class BelowThreshold:
    threshold: int
    def should_reorder(self, level: StockLevel) -> bool:
        return level.available() < self.threshold


class ReorderDetector:
    """Listener that fires once per crossing below the threshold."""
    def __init__(self, rule: ReorderRule, on_reorder: Callable[[str, SKU], None]) -> None:
        self.rule, self.on_reorder = rule, on_reorder
        self.armed: dict[tuple[str, SKU], bool] = {}

    def __call__(self, svc: InventoryService, event: StockEvent) -> None:
        key = (event.warehouse_id, event.sku)
        level = svc.warehouses[event.warehouse_id].level(event.sku)
        low = self.rule.should_reorder(level)
        if low and self.armed.get(key, True):
            self.armed[key] = False
            self.on_reorder(event.warehouse_id, event.sku)
        elif not low:
            self.armed[key] = True


class InventoryService:
    def __init__(self, chooser: WarehouseChooser) -> None:
        self.warehouses: dict[str, Warehouse] = {}
        self.reservations: dict[str, Reservation] = {}
        self.chooser = chooser
        self.listeners: list[Callable[[InventoryService, StockEvent], None]] = []
        self._ids = itertools.count(1)

    def _emit(self, kind: str, sku: SKU, qty: int, warehouse_id: str) -> None:
        ev = StockEvent(kind, sku, qty, warehouse_id)
        for fn in self.listeners:
            fn(self, ev)

    def receive(self, warehouse_id: str, sku: SKU, qty: int) -> None:
        self.warehouses[warehouse_id].level(sku).add(qty)
        self._emit("receive", sku, qty, warehouse_id)

    def reserve(self, sku: SKU, qty: int, ttl: float = 900) -> Reservation:
        for wh in self.chooser.order(sku, qty, list(self.warehouses.values())):
            if wh.level(sku).try_reserve(qty):
                res = Reservation(f"r{next(self._ids)}", sku, qty, wh.id, time.time() + ttl)
                self.reservations[res.id] = res
                self._emit("reserve", sku, qty, wh.id)
                return res
        raise OutOfStock(sku.code)

    def _finish(self, res: Reservation, to: ResState) -> None:
        if res.state is not ResState.HELD:
            raise IllegalTransition(res.state.value)
        res.state = to
        level = self.warehouses[res.warehouse_id].level(res.sku)
        if to is ResState.CONFIRMED:
            level.confirm(res.qty)
        else:
            level.release(res.qty)
        self._emit(to.value, res.sku, res.qty, res.warehouse_id)

    def confirm(self, reservation_id: str) -> None:
        self._finish(self.reservations[reservation_id], ResState.CONFIRMED)

    def release(self, reservation_id: str) -> None:
        self._finish(self.reservations[reservation_id], ResState.RELEASED)

    def expire(self, now: float) -> int:
        n = 0
        for res in list(self.reservations.values()):
            if res.state is ResState.HELD and res.expires_at <= now:
                self._finish(res, ResState.RELEASED); n += 1
        return n

    def availability(self, sku: SKU) -> dict[str, tuple[int, int, int]]:
        out = {}
        for wh in self.warehouses.values():
            l = wh.level(sku)
            out[wh.id] = (l.on_hand, l.reserved, l.available())
        return out

6. Extensions, concurrency and what interviewers probe

• Split reservations across warehouses when none can hold the whole quantity: reserve partials in order and roll back all of them if the total falls short. The rollback is why partial holds must be recorded before the next warehouse is tried.
• Expiry: expire(now) is a sweep for clarity; a heap of reservations by expires_at makes it O(log n) per expiry and is the expected follow-up.
• Audit adjustments that lower on-hand below reserved must be refused or must release reservations; name the choice, since both are defensible.
• Where the lock sits: inside StockLevel, around check-and-increment. _finish guards against double release through the state check; under threads that check also needs the level lock or a lock on the reservation.
• Test first: two threads reserving one unit each from a level with one unit end with exactly one success; confirm then release raises; a threshold of 5 with stock dropping 6, 4, 3, 2 emits one reorder; receiving stock back above 5 and dropping again emits a second.
• Reservation timing: reserve at checkout start, not when an item enters a cart, otherwise abandoned carts starve real buyers; the ttl argument is what bounds the damage when a buyer walks away mid-payment.
• Common mistake: computing availability as on_hand and forgetting reservations, which is where oversell comes from in otherwise correct code.

02Where this question was reported

• Hello Interview · Hello Interview breakdown (Hard, premium)
• Codemia · Codemia Medium; also 'Warehouse Inventory Tracker'
• LLD Canvas · Event-driven low-stock triggers
• LLD Mastery · Weak: commercial tag only