Inventory Management
DDD-inspired inventory system with Specification for composable product queries and reorder rules, Repository for clean data access, and Strategy for configurable replenishment algorithms.
Key Abstractions
Interface : is_satisfied_by(item) with and_spec/or_spec/not_spec combinators for composable queries
Concrete specifications for filtering products by availability, category, price range, and reorder eligibility
Repository : add, find_by_id, find_matching(spec), find_all. Collection-like abstraction over data access
Entity : id, name, category, price, quantity, reorder_point. Core domain object
Strategy : JustInTime orders exactly what's needed, SafetyStock adds a buffer, EconomicOrderQuantity optimizes batch size
State enum : RECEIVED, PICKING, PACKING, SHIPPED, DELIVERED. Tracks warehouse order lifecycle
Observer : alerts on low stock events and triggers automatic reorder workflows
Class Diagram
How It Works
Two patterns from Domain-Driven Design fit together naturally here: Specification and Repository. Together, they solve the eternal problem of "how do I query my domain objects without scattering filter logic everywhere?"
The Specification Pattern encodes a business rule as an object. InStockSpec knows what "in stock" means. CategorySpec("electronics") knows what "belongs to electronics" means. The key is the combinator methods: and_spec, or_spec, and not_spec let you compose simple specs into complex queries without writing new classes. InStockSpec().and_spec(CategorySpec("electronics")) reads almost like English.
The Repository Pattern presents data access as a collection. You add products, find_by_id, or find_matching(spec). The repository accepts a specification object and returns all matching items. This pair shows up constantly in DDD for good reason: the repository does not need a new method for every query shape. One find_matching method handles infinite query combinations.
The same Specification pattern serves two roles in this system. First, it drives product search and filtering (what the user sees). Second, it identifies products that need reordering (BelowReorderPointSpec). Same interface, same composition rules, completely different business context. That dual use is exactly why the pattern exists: it externalizes rules so you can reuse them across features.
Strategy handles the replenishment decision. Different product categories have different inventory economics. Perishables rot, so JustInTime orders exactly the deficit. Electronics have long lead times, so SafetyStock maintains a buffer. Furniture ships in bulk, so EconomicOrderQuantity uses fixed batch sizes. The strategy is assigned per category, not per product, keeping configuration manageable.
Observer closes the loop. When stock drops below the reorder point, observers fire alerts. The inventory service does not know what happens with those alerts: it just notifies. One observer logs to console. Another could send a Slack message. Another could auto-trigger a purchase order.
Requirements
Functional
- Product catalog with categories, prices, and stock levels
- Composable product queries: filter by category, price range, stock status, and combinations thereof
- Automatic reorder detection when stock falls below reorder points
- Per-category replenishment strategies with different order quantity calculations
- Warehouse order lifecycle tracking (received through delivered)
- Low-stock alerts and reorder notifications
Non-Functional
- Query logic must be reusable across features (search, reporting, reorder rules)
- Data access layer must be swappable (in-memory for tests, database for production)
- Replenishment algorithms must be configurable per product category
- Alert system must be extensible without modifying inventory logic
Design Decisions
What if we just added more repository methods instead?
The alternative is to add a method for every query shape: findByCategory, findByPriceRange, findByCategoryAndPriceRange, findInStockByCategory. This leads to a combinatorial explosion. With six filter dimensions, you need dozens of methods. Specification gives you six classes and infinite combinations via and_spec/or_spec/not_spec. Adding a new filter dimension means one new class, zero changes to the repository.
Is the in-memory repository just for testing?
The InMemoryProductRepository is not just a shortcut for testing. It proves that business logic is decoupled from storage. The InventoryService calls repo.find_matching(spec) without knowing whether products live in a HashMap, PostgreSQL, or an external API. Swapping the repository implementation is a constructor change, not a rewrite.
Can't we use one reorder formula for everything?
Hardcoding reorder logic means a single algorithm for all products. But perishable goods with a 3-day shelf life cannot use the same reorder math as electronics with a 6-week lead time. Strategy lets you assign the right algorithm per category and change it at runtime. Seasonal products might even switch strategies as demand shifts.
Should the inventory service send Slack messages directly?
The inventory service should not know about Slack, email, or dashboard updates. It knows stock changed. Observers decide what to do about it. This is especially important because alert channels change frequently. Adding SMS notifications should not require touching inventory code.
Interview Follow-ups
- "How would you implement CQRS for read-heavy inventory?" Separate the read model (optimized for search queries, denormalized, possibly cached) from the write model (handles stock updates with full validation). The Specification pattern maps cleanly to the read side: specs become query predicates against a read-optimized store.
- "How would you add event sourcing for stock movements?" Replace direct quantity mutation with a
StockEventlog (received, sold, damaged, transferred). Current quantity is computed by replaying events. This gives you a full audit trail and the ability to reconstruct inventory state at any point in time. - "How would you handle distributed inventory across warehouses?" Each warehouse gets its own repository. A
FederatedRepositoryaggregates queries across warehouses. Specifications remain unchanged: they operate on products regardless of which warehouse holds them. Replenishment strategies become warehouse-aware, considering transfer costs between locations. - "How would you handle reserved stock for pending orders?" Add
reserved_quantityto Product. Available stock isquantity - reserved_quantity. TheInStockSpecchecks available stock, not raw quantity. Reservations expire after a timeout to prevent phantom locks on inventory.
Code Implementation
1 from abc import ABC, abstractmethod
2 from enum import Enum
3 from dataclasses import dataclass, field
4 from typing import Optional
5 import uuid
6
7
8 # ── Specification Pattern ─────────────────────────────────────
9
10 class Specification(ABC):
11 """Base specification interface with composable combinators."""
12
13 @abstractmethod
14 def is_satisfied_by(self, item) -> bool:
15 pass
16
17 def and_spec(self, other: "Specification") -> "Specification":
18 return AndSpec(self, other)
19
20 def or_spec(self, other: "Specification") -> "Specification":
21 return OrSpec(self, other)
22
23 def not_spec(self) -> "Specification":
24 return NotSpec(self)
25
26
27 class AndSpec(Specification):
28 def __init__(self, left: Specification, right: Specification):
29 self._left = left
30 self._right = right
31
32 def is_satisfied_by(self, item) -> bool:
33 return self._left.is_satisfied_by(item) and self._right.is_satisfied_by(item)
34
35
36 class OrSpec(Specification):
37 def __init__(self, left: Specification, right: Specification):
38 self._left = left
39 self._right = right
40
41 def is_satisfied_by(self, item) -> bool:
42 return self._left.is_satisfied_by(item) or self._right.is_satisfied_by(item)
43
44
45 class NotSpec(Specification):
46 def __init__(self, spec: Specification):
47 self._spec = spec
48
49 def is_satisfied_by(self, item) -> bool:
50 return not self._spec.is_satisfied_by(item)
51
52
53 # ── Concrete Specifications ───────────────────────────────────
54
55 class InStockSpec(Specification):
56 def is_satisfied_by(self, item) -> bool:
57 return item.quantity > 0
58
59
60 class CategorySpec(Specification):
61 def __init__(self, category: str):
62 self._category = category
63
64 def is_satisfied_by(self, item) -> bool:
65 return item.category == self._category
66
67
68 class PriceRangeSpec(Specification):
69 def __init__(self, min_price: float, max_price: float):
70 self._min_price = min_price
71 self._max_price = max_price
72
73 def is_satisfied_by(self, item) -> bool:
74 return self._min_price <= item.price <= self._max_price
75
76
77 class BelowReorderPointSpec(Specification):
78 def is_satisfied_by(self, item) -> bool:
79 return item.quantity <= item.reorder_point
80
81
82 # ── Product Entity ────────────────────────────────────────────
83
84 @dataclass
85 class Product:
86 id: str
87 name: str
88 category: str
89 price: float
90 quantity: int
91 reorder_point: int
92
93 @staticmethod
94 def create(name: str, category: str, price: float, quantity: int, reorder_point: int) -> "Product":
95 return Product(
96 id=str(uuid.uuid4())[:8],
97 name=name,
98 category=category,
99 price=price,
100 quantity=quantity,
101 reorder_point=reorder_point,
102 )
103
104
105 # ── Repository Pattern ────────────────────────────────────────
106
107 class ProductRepository(ABC):
108 """Collection-like abstraction over product data access."""
109
110 @abstractmethod
111 def add(self, product: Product) -> None:
112 pass
113
114 @abstractmethod
115 def find_by_id(self, product_id: str) -> Optional[Product]:
116 pass
117
118 @abstractmethod
119 def find_matching(self, spec: Specification) -> list[Product]:
120 pass
121
122 @abstractmethod
123 def find_all(self) -> list[Product]:
124 pass
125
126
127 class InMemoryProductRepository(ProductRepository):
128 def __init__(self):
129 self._products: dict[str, Product] = {}
130
131 def add(self, product: Product) -> None:
132 self._products[product.id] = product
133
134 def find_by_id(self, product_id: str) -> Optional[Product]:
135 return self._products.get(product_id)
136
137 def find_matching(self, spec: Specification) -> list[Product]:
138 return [p for p in self._products.values() if spec.is_satisfied_by(p)]
139
140 def find_all(self) -> list[Product]:
141 return list(self._products.values())
142
143
144 # ── Strategy Pattern (Replenishment) ──────────────────────────
145
146 class ReplenishmentStrategy(ABC):
147 @abstractmethod
148 def calculate_order_quantity(self, product: Product) -> int:
149 pass
150
151
152 class JustInTimeStrategy(ReplenishmentStrategy):
153 """Order exactly the deficit to reach the reorder point."""
154
155 def calculate_order_quantity(self, product: Product) -> int:
156 if product.quantity >= product.reorder_point:
157 return 0
158 return product.reorder_point - product.quantity
159
160
161 class SafetyStockStrategy(ReplenishmentStrategy):
162 """Order enough to reach reorder_point * multiplier for a safety buffer."""
163
164 def __init__(self, safety_multiplier: float = 2.0):
165 self._multiplier = safety_multiplier
166
167 def calculate_order_quantity(self, product: Product) -> int:
168 target = int(product.reorder_point * self._multiplier)
169 if product.quantity >= target:
170 return 0
171 return target - product.quantity
172
173
174 class EconomicOrderQuantityStrategy(ReplenishmentStrategy):
175 """Fixed batch size optimized for ordering cost vs holding cost."""
176
177 def __init__(self, batch_size: int = 50):
178 self._batch_size = batch_size
179
180 def calculate_order_quantity(self, product: Product) -> int:
181 if product.quantity > product.reorder_point:
182 return 0
183 return self._batch_size
184
185
186 # ── State Pattern (Order Lifecycle) ───────────────────────────
187
188 class OrderState(Enum):
189 RECEIVED = "received"
190 PICKING = "picking"
191 PACKING = "packing"
192 SHIPPED = "shipped"
193 DELIVERED = "delivered"
194
195 def next_state(self) -> "OrderState":
196 transitions = {
197 OrderState.RECEIVED: OrderState.PICKING,
198 OrderState.PICKING: OrderState.PACKING,
199 OrderState.PACKING: OrderState.SHIPPED,
200 OrderState.SHIPPED: OrderState.DELIVERED,
201 }
202 nxt = transitions.get(self)
203 if nxt is None:
204 raise ValueError(f"No transition from {self.value}")
205 return nxt
206
207
208 @dataclass
209 class WarehouseOrder:
210 id: str
211 product_id: str
212 quantity: int
213 state: OrderState = OrderState.RECEIVED
214
215 def advance(self) -> None:
216 old = self.state
217 self.state = self.state.next_state()
218 print(f" Order {self.id}: {old.value} -> {self.state.value}")
219
220
221 # ── Observer Pattern ──────────────────────────────────────────
222
223 class StockObserver(ABC):
224 @abstractmethod
225 def on_low_stock(self, product: Product) -> None:
226 pass
227
228 @abstractmethod
229 def on_reorder_triggered(self, product: Product, order_qty: int) -> None:
230 pass
231
232
233 class AlertingObserver(StockObserver):
234 def on_low_stock(self, product: Product) -> None:
235 print(f" [ALERT] Low stock: {product.name} has {product.quantity} units (reorder point: {product.reorder_point})")
236
237 def on_reorder_triggered(self, product: Product, order_qty: int) -> None:
238 print(f" [REORDER] Placing order for {order_qty} units of {product.name}")
239
240
241 # ── Inventory Service (ties everything together) ──────────────
242
243 class InventoryService:
244 def __init__(self, repository: ProductRepository):
245 self._repo = repository
246 self._observers: list[StockObserver] = []
247 self._strategies: dict[str, ReplenishmentStrategy] = {}
248
249 def add_observer(self, observer: StockObserver) -> None:
250 self._observers.append(observer)
251
252 def set_strategy(self, category: str, strategy: ReplenishmentStrategy) -> None:
253 self._strategies[category] = strategy
254
255 def add_product(self, product: Product) -> None:
256 self._repo.add(product)
257
258 def search(self, spec: Specification) -> list[Product]:
259 return self._repo.find_matching(spec)
260
261 def update_stock(self, product_id: str, delta: int) -> None:
262 product = self._repo.find_by_id(product_id)
263 if product is None:
264 raise ValueError(f"Product {product_id} not found")
265 product.quantity += delta
266 if product.quantity <= product.reorder_point:
267 for obs in self._observers:
268 obs.on_low_stock(product)
269
270 def check_reorder_needs(self) -> list[WarehouseOrder]:
271 reorder_spec = BelowReorderPointSpec()
272 needing_reorder = self._repo.find_matching(reorder_spec)
273 orders: list[WarehouseOrder] = []
274 for product in needing_reorder:
275 strategy = self._strategies.get(product.category)
276 if strategy is None:
277 strategy = JustInTimeStrategy()
278 qty = strategy.calculate_order_quantity(product)
279 if qty > 0:
280 for obs in self._observers:
281 obs.on_reorder_triggered(product, qty)
282 order = WarehouseOrder(
283 id=str(uuid.uuid4())[:8],
284 product_id=product.id,
285 quantity=qty,
286 )
287 orders.append(order)
288 return orders
289
290
291 # ── Demo ──────────────────────────────────────────────────────
292
293 if __name__ == "__main__":
294 # Setup repository and service
295 repo = InMemoryProductRepository()
296 service = InventoryService(repo)
297 service.add_observer(AlertingObserver())
298
299 # Configure per-category replenishment strategies
300 service.set_strategy("electronics", SafetyStockStrategy(safety_multiplier=2.0))
301 service.set_strategy("perishables", JustInTimeStrategy())
302 service.set_strategy("furniture", EconomicOrderQuantityStrategy(batch_size=20))
303
304 # Add products
305 laptop = Product.create("Laptop", "electronics", 999.99, 15, 10)
306 phone = Product.create("Smartphone", "electronics", 499.99, 3, 10)
307 headphones = Product.create("Wireless Headphones", "electronics", 79.99, 50, 15)
308 milk = Product.create("Organic Milk", "perishables", 4.99, 8, 20)
309 bread = Product.create("Sourdough Bread", "perishables", 6.49, 2, 10)
310 desk = Product.create("Standing Desk", "furniture", 349.99, 5, 8)
311 chair = Product.create("Ergonomic Chair", "furniture", 249.99, 0, 5)
312
313 for p in [laptop, phone, headphones, milk, bread, desk, chair]:
314 service.add_product(p)
315
316 # ── Query 1: Electronics under $500 ──
317 print("=== Electronics under $500 ===")
318 spec = CategorySpec("electronics").and_spec(PriceRangeSpec(0, 500))
319 results = service.search(spec)
320 for p in results:
321 print(f" {p.name}: ${p.price:.2f} (qty: {p.quantity})")
322
323 # ── Query 2: In-stock items ──
324 print("\n=== In-stock items ===")
325 in_stock = service.search(InStockSpec())
326 for p in in_stock:
327 print(f" {p.name}: qty {p.quantity}")
328
329 # ── Query 3: Items needing reorder ──
330 print("\n=== Items below reorder point ===")
331 low_stock = service.search(BelowReorderPointSpec())
332 for p in low_stock:
333 print(f" {p.name}: qty {p.quantity}, reorder at {p.reorder_point}")
334
335 # ── Query 4: Composed NOT spec - out-of-stock items ──
336 print("\n=== Out-of-stock items ===")
337 out_of_stock = service.search(InStockSpec().not_spec())
338 for p in out_of_stock:
339 print(f" {p.name}: qty {p.quantity}")
340
341 # ── Reorder check with different strategies ──
342 print("\n=== Checking reorder needs (per-category strategies) ===")
343 orders = service.check_reorder_needs()
344
345 # ── Advance order states ──
346 print("\n=== Advancing warehouse orders ===")
347 for order in orders:
348 product = repo.find_by_id(order.product_id)
349 name = product.name if product else order.product_id
350 print(f" Order for {name} ({order.quantity} units):")
351 order.advance() # RECEIVED -> PICKING
352 order.advance() # PICKING -> PACKING
353 order.advance() # PACKING -> SHIPPED
354
355 # ── Simulate stock depletion triggering observer ──
356 print("\n=== Simulating stock depletion ===")
357 service.update_stock(laptop.id, -10)Common Mistakes
- ✗Embedding query logic in service classes: leads to duplicated filtering across different features
- ✗Not separating specification from repository: putting WHERE clause logic inside the repository method names (findByPriceAndCategory)
- ✗Mutable specifications: shared specs modified in one feature corrupt another
- ✗Single replenishment strategy for all products: perishables need JustInTime, electronics need SafetyStock
Key Points
- ✓Specification + Repository: `repo.find_matching(InStockSpec().and_spec(CategorySpec('electronics')))` accepts spec objects for querying. This pair shows up constantly in DDD for good reason.
- ✓Two uses of Specification in one system: (1) product search/filtering (2) reorder eligibility rules. Same pattern, different domains.
- ✓Repository: abstracts data access behind a collection-like interface. Swap InMemoryRepository for DatabaseRepository without changing business logic.
- ✓Strategy: replenishment algorithms calculate different reorder quantities. JustInTime orders exactly what's needed, SafetyStock adds a buffer.