PYTH-102 · Lesson 12 of 12 · 70 min
Capstone: a modular inventory program
▶ Video coming soon A video version of this lesson is on the way.
The PYTH-101 capstone, ridgeline.py, did its job: Theo could list the coffees, sell and restock, see low stock and save everything to a JSON file. But it was one long file built from dictionaries and loose functions, with prices stored as floats, sales recorded without a date or time, and the data saved only when someone remembered to type save. In this capstone Jordan rebuilds it as a proper package using everything from this course: modules and a package, dataclasses with properties, an Inventory class with its own exceptions, special methods, Decimal money, datetimes for every sale, type hints throughout, a decorator that registers commands, generators and comprehensions where they help, and automatic saving after every change. Rosa's goal is a tool the whole shop can rely on, and that Jordan can keep extending without fear of breaking it.
This lesson walks through the design, then each module in turn, then runs the assembled program on a scripted session, tests it, and finishes with a review of the decisions behind it and ideas for extending it. Read the modules in order; each one depends only on the ones before it.
The design
The first step is to decide what the pieces are and which may depend on which. Each module has one job, and imports flow in one direction, so there are no circular imports and each module can be understood and tested on its own.
- money.py: exact prices with Decimal, and formatting. It imports nothing from the package.
- models.py: the Coffee and Sale dataclasses. It imports money.
- inventory.py: the Inventory class and its exceptions, with the selling and restocking rules. It imports models and money.
- storage.py: converting the inventory to and from JSON, and saving it safely. It imports inventory, models and money.
- cli.py: the commands and the command loop. It imports everything above, and nothing imports it.
- __init__.py: the package's front door, exposing the most used names; main.py outside the package holds the main guard.
The design keeps two kinds of code apart. The rules of the business, such as "you cannot sell more bags than are in stock", live in inventory.py and models.py and know nothing about typing commands or files. The parts that deal with the outside world, the terminal and the disk, live in cli.py and storage.py. Keeping the rules separate from input and output is what makes them easy to test and lets the same Inventory class be used later by a website in PYTH-220, with no changes.
money.py: exact prices
Prices are Decimal values rounded to the cent, as lesson 10 recommended, created from text so that no float error ever gets in. to_money accepts strings, ints, Decimals or even floats, converting through str so that 18.5 becomes exactly 18.50. fmt adds a thousands separator, which Decimal's formatting supports just as floats do.
# Exact money: prices are Decimal amounts rounded to the cent.
from decimal import Decimal, ROUND_HALF_UP
CENT = Decimal("0.01")
ZERO = Decimal("0.00")
def to_money(value: object) -> Decimal:
return Decimal(str(value)).quantize(CENT, rounding=ROUND_HALF_UP)
def fmt(amount: Decimal) -> str:
return f"{amount:,.2f}"from decimal import Decimal, ROUND_HALF_UP
CENT = Decimal("0.01")
def to_money(value):
return Decimal(str(value)).quantize(CENT, rounding=ROUND_HALF_UP)
print(to_money("18.5"), to_money(18.5), to_money(3), to_money("2.675"))
print(f"{to_money('12345.6'):,.2f}", to_money("0.10") * 3)18.50 18.50 3.00 2.68 12,345.60 0.30
models.py: the data
Coffee is a dataclass with a __post_init__ method. A dataclass calls __post_init__ right after its generated __init__, which makes it the place to clean and check the fields: the SKU is normalized to upper case, the price is converted to money, and negative prices or stock raise ValueError immediately, before a bad object can exist. is_low and value are properties, computed from the current stock so they are never out of date. Sale is a frozen dataclass: once created, its fields cannot be changed, which suits a record of something that happened.
# The data: coffees on the shelf and the sales made from them.
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from .money import to_money
REORDER_POINT = 10
@dataclass
class Coffee:
sku: str
name: str
price: Decimal
stock: int = 0
def __post_init__(self) -> None:
self.sku = self.sku.strip().upper()
self.price = to_money(self.price)
if self.price < 0:
raise ValueError(f"{self.sku}: price cannot be negative")
if self.stock < 0:
raise ValueError(f"{self.sku}: stock cannot be negative")
@property
def is_low(self) -> bool:
return self.stock < REORDER_POINT
@property
def value(self) -> Decimal:
return self.price * self.stock
@dataclass(frozen=True)
class Sale:
sku: str
qty: int
price: Decimal
when: datetime
@property
def amount(self) -> Decimal:
return self.price * self.qtyEach Sale records the price at the moment of sale, not a reference to the coffee. That is deliberate: if the price of Sunrise Blend goes up next month, last month's sales must still show what customers actually paid. Recording the time with each sale is what makes daily and weekly reports possible.
inventory.py: the rules
Inventory composes Coffee objects, as in lesson 5, stored in a dictionary keyed by SKU. Its exceptions form a small hierarchy, so the command line can catch every inventory problem with one except clause. Special methods let the rest of the program use it naturally: len(inv), sku in inv, and for coffee in inv, which yields coffees in SKU order. The clock parameter is the most important design decision in the file: instead of calling datetime.now directly, the inventory calls whatever function it was given, which defaults to datetime.now.
# The rules: an Inventory holds the coffees and records every sale.
from collections.abc import Callable, Iterator
from datetime import datetime
from decimal import Decimal
from .models import Coffee, Sale
from .money import ZERO
class InventoryError(Exception):
pass
class UnknownCoffee(InventoryError):
pass
class OutOfStock(InventoryError):
pass
class Inventory:
def __init__(self, clock: Callable[[], datetime] = datetime.now) -> None:
self._coffees: dict[str, Coffee] = {}
self.sales: list[Sale] = []
self._clock = clock
def add(self, coffee: Coffee) -> None:
if coffee.sku in self._coffees:
raise InventoryError(f"duplicate SKU {coffee.sku}")
self._coffees[coffee.sku] = coffee
def get(self, sku: str) -> Coffee:
try:
return self._coffees[sku.strip().upper()]
except KeyError:
raise UnknownCoffee(f"no coffee with SKU {sku}") from None
def __len__(self) -> int:
return len(self._coffees)
def __iter__(self) -> Iterator[Coffee]:
return iter(sorted(self._coffees.values(), key=lambda c: c.sku))
def __contains__(self, sku: object) -> bool:
return isinstance(sku, str) and sku.strip().upper() in self._coffees
def sell(self, sku: str, qty: int) -> Sale:
if qty <= 0:
raise InventoryError("quantity must be positive")
coffee = self.get(sku)
if qty > coffee.stock:
raise OutOfStock(f"only {coffee.stock} bags of {coffee.name} left")
coffee.stock -= qty
sale = Sale(coffee.sku, qty, coffee.price, self._clock())
self.sales.append(sale)
return sale
def restock(self, sku: str, qty: int) -> Coffee:
if qty <= 0:
raise InventoryError("quantity must be positive")
coffee = self.get(sku)
coffee.stock += qty
return coffee
def low_stock(self) -> list[Coffee]:
return [c for c in self if c.is_low]
def total_value(self) -> Decimal:
return sum((c.value for c in self), ZERO)
def sales_by_day(self) -> dict[str, Decimal]:
totals: dict[str, Decimal] = {}
for sale in self.sales:
day = sale.when.date().isoformat()
totals[day] = totals.get(day, ZERO) + sale.amount
return totalsPassing in the clock is a technique called dependency injection. In the shop, the default gives the real time. In tests and in the listings on this page, a fake clock that returns fixed times is passed instead, so the output is the same on every run and can be checked. The same idea, handing an object the things it depends on instead of letting it reach out for them, makes code testable throughout professional software. The raise ... from None in get replaces the internal KeyError with a clear UnknownCoffee message, so users never see a confusing chained traceback.
storage.py: saving safely
storage converts between an Inventory and plain dictionaries that JSON can hold. Prices are stored as strings, such as "18.50", because JSON has no decimal type and storing them as floats would reintroduce exactly the errors Decimal removes. Datetimes are stored in ISO format. When there is no file yet, load builds the starter inventory, which matches the stock Ridgeline began PYTH-101 with.
# Saving and loading: the inventory as JSON, written safely.
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from .inventory import Inventory
from .models import Coffee, Sale
from .money import to_money
STARTER = {"coffees": [
{"sku": "RC-101", "name": "Sunrise Blend", "price": "18.50", "stock": 42},
{"sku": "RC-102", "name": "Ethiopia Guji", "price": "22.00", "stock": 15},
{"sku": "RC-103", "name": "Colombia Huila", "price": "19.75", "stock": 27},
{"sku": "RC-104", "name": "Sumatra Dark", "price": "20.25", "stock": 9},
{"sku": "RC-105", "name": "Decaf Swiss Water", "price": "19.00", "stock": 12},
{"sku": "RC-106", "name": "Espresso Ridge", "price": "21.50", "stock": 30},
], "sales": []}
def to_dict(inv: Inventory) -> dict[str, Any]:
return {
"coffees": [{"sku": c.sku, "name": c.name, "price": str(c.price), "stock": c.stock} for c in inv],
"sales": [{"sku": s.sku, "qty": s.qty, "price": str(s.price), "when": s.when.isoformat()}
for s in inv.sales],
}
def from_dict(data: dict[str, Any], **options: Any) -> Inventory:
inv = Inventory(**options)
for row in data.get("coffees", []):
inv.add(Coffee(row["sku"], row["name"], row["price"], int(row["stock"])))
for row in data.get("sales", []):
when = datetime.fromisoformat(row["when"])
inv.sales.append(Sale(row["sku"], int(row["qty"]), to_money(row["price"]), when))
return inv
def save(inv: Inventory, path: Path) -> None:
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(to_dict(inv), indent=2))
tmp.replace(path)
def load(path: Path, **options: Any) -> Inventory:
if not path.exists():
return from_dict(STARTER, **options)
return from_dict(json.loads(path.read_text()), **options)save never writes over the real file directly. It writes a temporary file first and then replaces the real one in a single step. If the power fails or the program crashes halfway through writing, the old file is still intact, because the replacement only happens once the new file is complete. This write-then-replace pattern is how careful programs avoid corrupted save files, and since the program now saves after every sell or restock, it matters.
cli.py: the command line
The command loop uses the dispatch table from PYTH-101, but now the table is filled by a decorator, as in lesson 2: @command("sell") above a function registers it under that name. Adding a new command means writing one decorated function; nothing else changes. Every command takes the inventory and the list of words after the command name, and returns the text to print, so commands never print directly and can be tested by checking what they return.
# The command line: one function per command, registered by a decorator.
from collections.abc import Callable, Iterable
from datetime import datetime
from pathlib import Path
from . import storage
from .inventory import Inventory, InventoryError
from .money import fmt
Handler = Callable[[Inventory, list[str]], str]
COMMANDS: dict[str, Handler] = {}
CHANGES = {"sell", "restock"}
def command(name: str) -> Callable[[Handler], Handler]:
def register(func: Handler) -> Handler:
COMMANDS[name] = func
return func
return register
@command("list")
def list_coffees(inv: Inventory, args: list[str]) -> str:
return "\n".join(f"{c.sku} {c.name:<18}{c.stock:>4} {fmt(c.price):>6}" + (" LOW" if c.is_low else "")
for c in inv)
@command("sell")
def sell(inv: Inventory, args: list[str]) -> str:
sku, qty = args
sale = inv.sell(sku, int(qty))
return f"sold {sale.qty} x {sale.sku} for {fmt(sale.amount)}"
@command("restock")
def restock(inv: Inventory, args: list[str]) -> str:
sku, qty = args
coffee = inv.restock(sku, int(qty))
return f"{coffee.sku} now has {coffee.stock} bags"
@command("low")
def low(inv: Inventory, args: list[str]) -> str:
names = [f"{c.sku} ({c.stock})" for c in inv.low_stock()]
return "low stock: " + (", ".join(names) if names else "none")
@command("report")
def report(inv: Inventory, args: list[str]) -> str:
lines = [f"stock value {fmt(inv.total_value())}"]
for day, total in sorted(inv.sales_by_day().items()):
lines.append(f"sales {day}: {fmt(total)}")
return "\n".join(lines)
@command("help")
def show_help(inv: Inventory, args: list[str]) -> str:
return "commands: " + ", ".join(sorted(COMMANDS)) + ", quit"
def run_line(inv: Inventory, line: str, path: Path | None = None) -> str:
words = line.split()
if not words:
return ""
name, args = words[0].lower(), words[1:]
handler = COMMANDS.get(name)
if handler is None:
return f"unknown command {name!r}; try help"
try:
result = handler(inv, args)
except InventoryError as err:
return f"error: {err}"
except ValueError:
return f"could not understand {line!r}; try help"
if name in CHANGES and path is not None:
storage.save(inv, path)
return result
def main(script: Iterable[str] | None = None, path: Path = Path("ridgeline.json"),
clock: Callable[[], datetime] | None = None) -> None:
options = {"clock": clock} if clock else {}
inv = storage.load(path, **options)
lines = script if script is not None else iter(lambda: input("ridgeline> "), "quit")
for line in lines:
out = run_line(inv, line, path)
if out:
print(out)run_line does the work for one line of input: split it, find the command, run it, turn expected problems into friendly messages, and save if the command changed the data. InventoryError covers the business rules; ValueError covers malformed input, such as a missing quantity or a quantity that is not a number. Anything else, such as a TypeError from a bug, is deliberately not caught, so bugs fail loudly during development instead of being hidden behind a vague message. main accepts a script, a list of command lines, so the whole program can be run without anyone typing; when no script is given, iter(lambda: input(...), "quit") keeps asking for lines until the user types quit.
# Ridgeline Coffee Roasters tools: the most used names, in one place.
from .inventory import Inventory, InventoryError, OutOfStock, UnknownCoffee
from .models import REORDER_POINT, Coffee, Sale
from .money import fmt, to_money
__all__ = ["Coffee", "Sale", "REORDER_POINT", "Inventory", "InventoryError",
"OutOfStock", "UnknownCoffee", "fmt", "to_money"]# Entry point: python main.py starts the interactive tool.
from ridgeline.cli import main
if __name__ == "__main__":
main()Running the program
The listing below writes all seven files and then runs a scripted session with a fake clock. It is long because it contains the whole package as text; the part to read is at the very end, after the files are written. The second call to main starts the program again from the saved file, showing that the sales and stock survived the restart. Run it, then try changing the script: sell more than is in stock, restock a coffee that does not exist, or type a command that is not on the list.
import importlib, os, sys
sys.dont_write_bytecode = True
sys.path.insert(0, os.getcwd())
FILES = {}
FILES['ridgeline/__init__.py'] = r"""# Ridgeline Coffee Roasters tools: the most used names, in one place.
from .inventory import Inventory, InventoryError, OutOfStock, UnknownCoffee
from .models import REORDER_POINT, Coffee, Sale
from .money import fmt, to_money
__all__ = ["Coffee", "Sale", "REORDER_POINT", "Inventory", "InventoryError",
"OutOfStock", "UnknownCoffee", "fmt", "to_money"]
"""
FILES['ridgeline/money.py'] = r"""# Exact money: prices are Decimal amounts rounded to the cent.
from decimal import Decimal, ROUND_HALF_UP
CENT = Decimal("0.01")
ZERO = Decimal("0.00")
def to_money(value: object) -> Decimal:
return Decimal(str(value)).quantize(CENT, rounding=ROUND_HALF_UP)
def fmt(amount: Decimal) -> str:
return f"{amount:,.2f}"
"""
FILES['ridgeline/models.py'] = r"""# The data: coffees on the shelf and the sales made from them.
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from .money import to_money
REORDER_POINT = 10
@dataclass
class Coffee:
sku: str
name: str
price: Decimal
stock: int = 0
def __post_init__(self) -> None:
self.sku = self.sku.strip().upper()
self.price = to_money(self.price)
if self.price < 0:
raise ValueError(f"{self.sku}: price cannot be negative")
if self.stock < 0:
raise ValueError(f"{self.sku}: stock cannot be negative")
@property
def is_low(self) -> bool:
return self.stock < REORDER_POINT
@property
def value(self) -> Decimal:
return self.price * self.stock
@dataclass(frozen=True)
class Sale:
sku: str
qty: int
price: Decimal
when: datetime
@property
def amount(self) -> Decimal:
return self.price * self.qty
"""
FILES['ridgeline/inventory.py'] = r"""# The rules: an Inventory holds the coffees and records every sale.
from collections.abc import Callable, Iterator
from datetime import datetime
from decimal import Decimal
from .models import Coffee, Sale
from .money import ZERO
class InventoryError(Exception):
pass
class UnknownCoffee(InventoryError):
pass
class OutOfStock(InventoryError):
pass
class Inventory:
def __init__(self, clock: Callable[[], datetime] = datetime.now) -> None:
self._coffees: dict[str, Coffee] = {}
self.sales: list[Sale] = []
self._clock = clock
def add(self, coffee: Coffee) -> None:
if coffee.sku in self._coffees:
raise InventoryError(f"duplicate SKU {coffee.sku}")
self._coffees[coffee.sku] = coffee
def get(self, sku: str) -> Coffee:
try:
return self._coffees[sku.strip().upper()]
except KeyError:
raise UnknownCoffee(f"no coffee with SKU {sku}") from None
def __len__(self) -> int:
return len(self._coffees)
def __iter__(self) -> Iterator[Coffee]:
return iter(sorted(self._coffees.values(), key=lambda c: c.sku))
def __contains__(self, sku: object) -> bool:
return isinstance(sku, str) and sku.strip().upper() in self._coffees
def sell(self, sku: str, qty: int) -> Sale:
if qty <= 0:
raise InventoryError("quantity must be positive")
coffee = self.get(sku)
if qty > coffee.stock:
raise OutOfStock(f"only {coffee.stock} bags of {coffee.name} left")
coffee.stock -= qty
sale = Sale(coffee.sku, qty, coffee.price, self._clock())
self.sales.append(sale)
return sale
def restock(self, sku: str, qty: int) -> Coffee:
if qty <= 0:
raise InventoryError("quantity must be positive")
coffee = self.get(sku)
coffee.stock += qty
return coffee
def low_stock(self) -> list[Coffee]:
return [c for c in self if c.is_low]
def total_value(self) -> Decimal:
return sum((c.value for c in self), ZERO)
def sales_by_day(self) -> dict[str, Decimal]:
totals: dict[str, Decimal] = {}
for sale in self.sales:
day = sale.when.date().isoformat()
totals[day] = totals.get(day, ZERO) + sale.amount
return totals
"""
FILES['ridgeline/storage.py'] = r"""# Saving and loading: the inventory as JSON, written safely.
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from .inventory import Inventory
from .models import Coffee, Sale
from .money import to_money
STARTER = {"coffees": [
{"sku": "RC-101", "name": "Sunrise Blend", "price": "18.50", "stock": 42},
{"sku": "RC-102", "name": "Ethiopia Guji", "price": "22.00", "stock": 15},
{"sku": "RC-103", "name": "Colombia Huila", "price": "19.75", "stock": 27},
{"sku": "RC-104", "name": "Sumatra Dark", "price": "20.25", "stock": 9},
{"sku": "RC-105", "name": "Decaf Swiss Water", "price": "19.00", "stock": 12},
{"sku": "RC-106", "name": "Espresso Ridge", "price": "21.50", "stock": 30},
], "sales": []}
def to_dict(inv: Inventory) -> dict[str, Any]:
return {
"coffees": [{"sku": c.sku, "name": c.name, "price": str(c.price), "stock": c.stock} for c in inv],
"sales": [{"sku": s.sku, "qty": s.qty, "price": str(s.price), "when": s.when.isoformat()}
for s in inv.sales],
}
def from_dict(data: dict[str, Any], **options: Any) -> Inventory:
inv = Inventory(**options)
for row in data.get("coffees", []):
inv.add(Coffee(row["sku"], row["name"], row["price"], int(row["stock"])))
for row in data.get("sales", []):
when = datetime.fromisoformat(row["when"])
inv.sales.append(Sale(row["sku"], int(row["qty"]), to_money(row["price"]), when))
return inv
def save(inv: Inventory, path: Path) -> None:
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(to_dict(inv), indent=2))
tmp.replace(path)
def load(path: Path, **options: Any) -> Inventory:
if not path.exists():
return from_dict(STARTER, **options)
return from_dict(json.loads(path.read_text()), **options)
"""
FILES['ridgeline/cli.py'] = r"""# The command line: one function per command, registered by a decorator.
from collections.abc import Callable, Iterable
from datetime import datetime
from pathlib import Path
from . import storage
from .inventory import Inventory, InventoryError
from .money import fmt
Handler = Callable[[Inventory, list[str]], str]
COMMANDS: dict[str, Handler] = {}
CHANGES = {"sell", "restock"}
def command(name: str) -> Callable[[Handler], Handler]:
def register(func: Handler) -> Handler:
COMMANDS[name] = func
return func
return register
@command("list")
def list_coffees(inv: Inventory, args: list[str]) -> str:
return "\n".join(f"{c.sku} {c.name:<18}{c.stock:>4} {fmt(c.price):>6}" + (" LOW" if c.is_low else "")
for c in inv)
@command("sell")
def sell(inv: Inventory, args: list[str]) -> str:
sku, qty = args
sale = inv.sell(sku, int(qty))
return f"sold {sale.qty} x {sale.sku} for {fmt(sale.amount)}"
@command("restock")
def restock(inv: Inventory, args: list[str]) -> str:
sku, qty = args
coffee = inv.restock(sku, int(qty))
return f"{coffee.sku} now has {coffee.stock} bags"
@command("low")
def low(inv: Inventory, args: list[str]) -> str:
names = [f"{c.sku} ({c.stock})" for c in inv.low_stock()]
return "low stock: " + (", ".join(names) if names else "none")
@command("report")
def report(inv: Inventory, args: list[str]) -> str:
lines = [f"stock value {fmt(inv.total_value())}"]
for day, total in sorted(inv.sales_by_day().items()):
lines.append(f"sales {day}: {fmt(total)}")
return "\n".join(lines)
@command("help")
def show_help(inv: Inventory, args: list[str]) -> str:
return "commands: " + ", ".join(sorted(COMMANDS)) + ", quit"
def run_line(inv: Inventory, line: str, path: Path | None = None) -> str:
words = line.split()
if not words:
return ""
name, args = words[0].lower(), words[1:]
handler = COMMANDS.get(name)
if handler is None:
return f"unknown command {name!r}; try help"
try:
result = handler(inv, args)
except InventoryError as err:
return f"error: {err}"
except ValueError:
return f"could not understand {line!r}; try help"
if name in CHANGES and path is not None:
storage.save(inv, path)
return result
def main(script: Iterable[str] | None = None, path: Path = Path("ridgeline.json"),
clock: Callable[[], datetime] | None = None) -> None:
options = {"clock": clock} if clock else {}
inv = storage.load(path, **options)
lines = script if script is not None else iter(lambda: input("ridgeline> "), "quit")
for line in lines:
out = run_line(inv, line, path)
if out:
print(out)
"""
FILES['main.py'] = r"""# Entry point: python main.py starts the interactive tool.
from ridgeline.cli import main
if __name__ == "__main__":
main()
"""
for path, text in FILES.items():
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w") as f:
f.write(text)
for name in [n for n in sys.modules if n == "ridgeline" or n.startswith("ridgeline.")]:
del sys.modules[name]
importlib.invalidate_caches()
for leftover in ("ridgeline.json", "ridgeline.tmp"):
if os.path.exists(leftover):
os.remove(leftover)
import itertools
from datetime import datetime, timedelta
ticks = (datetime(2026, 11, 10, 8, 0) + timedelta(hours=9 * i) for i in itertools.count())
clock = lambda: next(ticks)
from ridgeline.cli import main
main(["list", "sell RC-101 2", "sell rc-104 3", "sell RC-104 40", "restock RC-104 30",
"sell RC-999 1", "sell RC-102", "low", "report", "help"], clock=clock)
print("---- restarted: the saved file is loaded ----")
main(["sell RC-106 1", "report"], clock=clock)RC-101 Sunrise Blend 42 18.50 RC-102 Ethiopia Guji 15 22.00 RC-103 Colombia Huila 27 19.75 RC-104 Sumatra Dark 9 20.25 LOW RC-105 Decaf Swiss Water 12 19.00 RC-106 Espresso Ridge 30 21.50 sold 2 x RC-101 for 37.00 sold 3 x RC-104 for 60.75 error: only 6 bags of Sumatra Dark left RC-104 now has 36 bags error: no coffee with SKU RC-999 could not understand 'sell RC-102'; try help low stock: none stock value 3,205.25 sales 2026-11-10: 97.75 commands: help, list, low, report, restock, sell, quit ---- restarted: the saved file is loaded ---- sold 1 x RC-106 for 21.50 stock value 3,183.75 sales 2026-11-10: 97.75 sales 2026-11-11: 21.50
Every message comes from the code you have read. sell rc-104 3 worked despite the lower-case SKU, because Inventory.get normalizes it. Selling 40 bags of Sumatra Dark was refused with an OutOfStock message, and the stock was left unchanged. The unknown SKU and the missing quantity produced friendly errors rather than tracebacks. After the restart, the report shows the earlier sales again, read back from ridgeline.json, together with the new one.
Try it yourself
Add a value command
The setup has written the whole ridgeline package. Add a new command, value, without changing any of the package files: import command from ridgeline.cli, and write a decorated function value(inv, args) that returns the text "stock value " followed by the total stock value formatted with fmt.
Then the checker runs run_line(inv, "value") on a fresh inventory. For the starter inventory the result is stock value 2,695.50.
from ridgeline.cli import command, run_line
from ridgeline.money import fmt
from ridgeline import storage
from pathlib import Path
inv = storage.load(Path('ridgeline.json'))
print(run_line(inv, 'value'))Testing the package
Because the rules live in classes that do not touch the terminal or the disk, testing them is simple: create an inventory with a fixed clock, do something, and assert on the result. The test functions below follow the conventions of pytest, the testing tool PYTH-230 teaches: each test is a function whose name starts with test_, sets up what it needs, and uses plain assert statements. Here a three-line loop at the bottom runs them, so that the listing works in the browser; with pytest installed, typing pytest in a terminal would find and run them automatically.
import importlib, os, sys
sys.dont_write_bytecode = True
sys.path.insert(0, os.getcwd())
FILES = {}
FILES['ridgeline/__init__.py'] = r"""# Ridgeline Coffee Roasters tools: the most used names, in one place.
from .inventory import Inventory, InventoryError, OutOfStock, UnknownCoffee
from .models import REORDER_POINT, Coffee, Sale
from .money import fmt, to_money
__all__ = ["Coffee", "Sale", "REORDER_POINT", "Inventory", "InventoryError",
"OutOfStock", "UnknownCoffee", "fmt", "to_money"]
"""
FILES['ridgeline/money.py'] = r"""# Exact money: prices are Decimal amounts rounded to the cent.
from decimal import Decimal, ROUND_HALF_UP
CENT = Decimal("0.01")
ZERO = Decimal("0.00")
def to_money(value: object) -> Decimal:
return Decimal(str(value)).quantize(CENT, rounding=ROUND_HALF_UP)
def fmt(amount: Decimal) -> str:
return f"{amount:,.2f}"
"""
FILES['ridgeline/models.py'] = r"""# The data: coffees on the shelf and the sales made from them.
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from .money import to_money
REORDER_POINT = 10
@dataclass
class Coffee:
sku: str
name: str
price: Decimal
stock: int = 0
def __post_init__(self) -> None:
self.sku = self.sku.strip().upper()
self.price = to_money(self.price)
if self.price < 0:
raise ValueError(f"{self.sku}: price cannot be negative")
if self.stock < 0:
raise ValueError(f"{self.sku}: stock cannot be negative")
@property
def is_low(self) -> bool:
return self.stock < REORDER_POINT
@property
def value(self) -> Decimal:
return self.price * self.stock
@dataclass(frozen=True)
class Sale:
sku: str
qty: int
price: Decimal
when: datetime
@property
def amount(self) -> Decimal:
return self.price * self.qty
"""
FILES['ridgeline/inventory.py'] = r"""# The rules: an Inventory holds the coffees and records every sale.
from collections.abc import Callable, Iterator
from datetime import datetime
from decimal import Decimal
from .models import Coffee, Sale
from .money import ZERO
class InventoryError(Exception):
pass
class UnknownCoffee(InventoryError):
pass
class OutOfStock(InventoryError):
pass
class Inventory:
def __init__(self, clock: Callable[[], datetime] = datetime.now) -> None:
self._coffees: dict[str, Coffee] = {}
self.sales: list[Sale] = []
self._clock = clock
def add(self, coffee: Coffee) -> None:
if coffee.sku in self._coffees:
raise InventoryError(f"duplicate SKU {coffee.sku}")
self._coffees[coffee.sku] = coffee
def get(self, sku: str) -> Coffee:
try:
return self._coffees[sku.strip().upper()]
except KeyError:
raise UnknownCoffee(f"no coffee with SKU {sku}") from None
def __len__(self) -> int:
return len(self._coffees)
def __iter__(self) -> Iterator[Coffee]:
return iter(sorted(self._coffees.values(), key=lambda c: c.sku))
def __contains__(self, sku: object) -> bool:
return isinstance(sku, str) and sku.strip().upper() in self._coffees
def sell(self, sku: str, qty: int) -> Sale:
if qty <= 0:
raise InventoryError("quantity must be positive")
coffee = self.get(sku)
if qty > coffee.stock:
raise OutOfStock(f"only {coffee.stock} bags of {coffee.name} left")
coffee.stock -= qty
sale = Sale(coffee.sku, qty, coffee.price, self._clock())
self.sales.append(sale)
return sale
def restock(self, sku: str, qty: int) -> Coffee:
if qty <= 0:
raise InventoryError("quantity must be positive")
coffee = self.get(sku)
coffee.stock += qty
return coffee
def low_stock(self) -> list[Coffee]:
return [c for c in self if c.is_low]
def total_value(self) -> Decimal:
return sum((c.value for c in self), ZERO)
def sales_by_day(self) -> dict[str, Decimal]:
totals: dict[str, Decimal] = {}
for sale in self.sales:
day = sale.when.date().isoformat()
totals[day] = totals.get(day, ZERO) + sale.amount
return totals
"""
FILES['ridgeline/storage.py'] = r"""# Saving and loading: the inventory as JSON, written safely.
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from .inventory import Inventory
from .models import Coffee, Sale
from .money import to_money
STARTER = {"coffees": [
{"sku": "RC-101", "name": "Sunrise Blend", "price": "18.50", "stock": 42},
{"sku": "RC-102", "name": "Ethiopia Guji", "price": "22.00", "stock": 15},
{"sku": "RC-103", "name": "Colombia Huila", "price": "19.75", "stock": 27},
{"sku": "RC-104", "name": "Sumatra Dark", "price": "20.25", "stock": 9},
{"sku": "RC-105", "name": "Decaf Swiss Water", "price": "19.00", "stock": 12},
{"sku": "RC-106", "name": "Espresso Ridge", "price": "21.50", "stock": 30},
], "sales": []}
def to_dict(inv: Inventory) -> dict[str, Any]:
return {
"coffees": [{"sku": c.sku, "name": c.name, "price": str(c.price), "stock": c.stock} for c in inv],
"sales": [{"sku": s.sku, "qty": s.qty, "price": str(s.price), "when": s.when.isoformat()}
for s in inv.sales],
}
def from_dict(data: dict[str, Any], **options: Any) -> Inventory:
inv = Inventory(**options)
for row in data.get("coffees", []):
inv.add(Coffee(row["sku"], row["name"], row["price"], int(row["stock"])))
for row in data.get("sales", []):
when = datetime.fromisoformat(row["when"])
inv.sales.append(Sale(row["sku"], int(row["qty"]), to_money(row["price"]), when))
return inv
def save(inv: Inventory, path: Path) -> None:
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(to_dict(inv), indent=2))
tmp.replace(path)
def load(path: Path, **options: Any) -> Inventory:
if not path.exists():
return from_dict(STARTER, **options)
return from_dict(json.loads(path.read_text()), **options)
"""
FILES['ridgeline/cli.py'] = r"""# The command line: one function per command, registered by a decorator.
from collections.abc import Callable, Iterable
from datetime import datetime
from pathlib import Path
from . import storage
from .inventory import Inventory, InventoryError
from .money import fmt
Handler = Callable[[Inventory, list[str]], str]
COMMANDS: dict[str, Handler] = {}
CHANGES = {"sell", "restock"}
def command(name: str) -> Callable[[Handler], Handler]:
def register(func: Handler) -> Handler:
COMMANDS[name] = func
return func
return register
@command("list")
def list_coffees(inv: Inventory, args: list[str]) -> str:
return "\n".join(f"{c.sku} {c.name:<18}{c.stock:>4} {fmt(c.price):>6}" + (" LOW" if c.is_low else "")
for c in inv)
@command("sell")
def sell(inv: Inventory, args: list[str]) -> str:
sku, qty = args
sale = inv.sell(sku, int(qty))
return f"sold {sale.qty} x {sale.sku} for {fmt(sale.amount)}"
@command("restock")
def restock(inv: Inventory, args: list[str]) -> str:
sku, qty = args
coffee = inv.restock(sku, int(qty))
return f"{coffee.sku} now has {coffee.stock} bags"
@command("low")
def low(inv: Inventory, args: list[str]) -> str:
names = [f"{c.sku} ({c.stock})" for c in inv.low_stock()]
return "low stock: " + (", ".join(names) if names else "none")
@command("report")
def report(inv: Inventory, args: list[str]) -> str:
lines = [f"stock value {fmt(inv.total_value())}"]
for day, total in sorted(inv.sales_by_day().items()):
lines.append(f"sales {day}: {fmt(total)}")
return "\n".join(lines)
@command("help")
def show_help(inv: Inventory, args: list[str]) -> str:
return "commands: " + ", ".join(sorted(COMMANDS)) + ", quit"
def run_line(inv: Inventory, line: str, path: Path | None = None) -> str:
words = line.split()
if not words:
return ""
name, args = words[0].lower(), words[1:]
handler = COMMANDS.get(name)
if handler is None:
return f"unknown command {name!r}; try help"
try:
result = handler(inv, args)
except InventoryError as err:
return f"error: {err}"
except ValueError:
return f"could not understand {line!r}; try help"
if name in CHANGES and path is not None:
storage.save(inv, path)
return result
def main(script: Iterable[str] | None = None, path: Path = Path("ridgeline.json"),
clock: Callable[[], datetime] | None = None) -> None:
options = {"clock": clock} if clock else {}
inv = storage.load(path, **options)
lines = script if script is not None else iter(lambda: input("ridgeline> "), "quit")
for line in lines:
out = run_line(inv, line, path)
if out:
print(out)
"""
FILES['main.py'] = r"""# Entry point: python main.py starts the interactive tool.
from ridgeline.cli import main
if __name__ == "__main__":
main()
"""
for path, text in FILES.items():
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w") as f:
f.write(text)
for name in [n for n in sys.modules if n == "ridgeline" or n.startswith("ridgeline.")]:
del sys.modules[name]
importlib.invalidate_caches()
for leftover in ("ridgeline.json", "ridgeline.tmp"):
if os.path.exists(leftover):
os.remove(leftover)
from datetime import datetime
from decimal import Decimal
from ridgeline import Coffee, Inventory, OutOfStock, UnknownCoffee, to_money
from ridgeline import storage
FIXED = datetime(2026, 11, 10, 9, 30)
def make_inventory():
inv = Inventory(clock=lambda: FIXED)
inv.add(Coffee("RC-104", "Sumatra Dark", "20.25", 9))
return inv
def test_money_rounds_half_up():
assert to_money("2.675") == Decimal("2.68")
def test_sell_reduces_stock_and_records_sale():
inv = make_inventory()
sale = inv.sell("rc-104", 3)
assert inv.get("RC-104").stock == 6
assert sale.amount == Decimal("60.75") and sale.when == FIXED
def test_cannot_oversell():
inv = make_inventory()
try:
inv.sell("RC-104", 10)
except OutOfStock:
pass
else:
raise AssertionError("expected OutOfStock")
assert inv.get("RC-104").stock == 9 and inv.sales == []
def test_unknown_sku():
try:
make_inventory().get("RC-999")
except UnknownCoffee as err:
assert "RC-999" in str(err)
def test_save_and_load_round_trip(tmp="test_round_trip.json"):
from pathlib import Path
inv = make_inventory()
inv.sell("RC-104", 2)
storage.save(inv, Path(tmp))
again = storage.load(Path(tmp))
assert storage.to_dict(again) == storage.to_dict(inv)
Path(tmp).unlink()
tests = [f for name, f in sorted(globals().items()) if name.startswith("test_")]
for test in tests:
test()
print("passed", test.__name__)
print(len(tests), "tests passed")passed test_cannot_oversell passed test_money_rounds_half_up passed test_save_and_load_round_trip passed test_sell_reduces_stock_and_records_sale passed test_unknown_sku 5 tests passed
Each test checks one behaviour and is named for it, so a failure says exactly what broke. test_cannot_oversell checks not only that the exception is raised but also that nothing changed: the stock is still 9 and no sale was recorded. Checking that a failed operation leaves no trace is easy to forget and catches real bugs. The round-trip test saves and loads and compares the results, the standard way to test any save format.
Try it yourself
Find coffees by name
Write a function find_coffees(inv, text) that returns a list of the SKUs of the coffees whose names contain text, ignoring case, in SKU order. Use a comprehension over the inventory, which iterates in SKU order.
Then register it as a command find that returns the matching SKUs joined by ", ", or "no match" if there are none. For the starter inventory, run_line(inv, "find dark") returns "RC-104".
from ridgeline.cli import command, run_line
from ridgeline import storage
def find_coffees(inv, text):
return []
inv = storage.from_dict(storage.STARTER)
print(find_coffees(inv, 'ia'))Reviewing the design
Before calling a project finished, professional teams review it. Rosa asked Jordan to explain the main decisions, and they make a good summary of what this course has been about.
- Why a package? Each module can be read on its own, and a change to the save format touches only storage.py.
- Why dataclasses with __post_init__? They remove boilerplate while still guaranteeing that a Coffee is always valid.
- Why Decimal and prices stored as strings? So that totals are exact and nothing is lost in the JSON file.
- Why a clock parameter? So that time-dependent code can be tested with fixed times.
- Why do commands return text instead of printing? So that they can be tested and reused, for example by a web page.
- Why save after every change, and through a temporary file? So that a crash loses at most one command and never corrupts the data.
- Why type hints? So that an editor or mypy can catch mistakes, and so that each function's signature documents what it expects.
The program is still small, and it has limits worth knowing. Two people running it at once on the same file could overwrite each other's changes, which is one reason PYTH-215 moves the data into a database. The command parsing is simple, and PYTH-240 replaces it with argparse. Its tests run by hand, and PYTH-230 runs them with pytest on every change.
Worked problem
Planning the next feature
Kai wants roast batches in the tool: a roast command that records a batch (batch id, SKU, green kilograms, roasted kilograms) and adds the bags it fills to that coffee's stock. Plan the change module by module.
Show solution
- models.py: add a frozen RoastBatch dataclass with a bags property, computed as roasted kilograms times 1000 divided by 340, rounded down.
- inventory.py: add self.batches and a roast(batch) method that checks the SKU, restocks by batch.bags and records the batch.
- storage.py: add batches to to_dict and from_dict, with weights stored as strings or numbers.
- cli.py: add @command("roast") that parses the four arguments and calls inv.roast, and add "roast" to CHANGES so it saves.
- Tests: roasting 10.2 kg adds 30 bags; an unknown SKU raises UnknownCoffee and records nothing; batches survive a save and load.
Answer: Each layer gets one small change, and the dependency direction stays the same.
Try it yourself
Best-selling coffee
Write best_seller(inv) that returns a tuple (sku, bags) for the coffee with the most bags sold across inv.sales, or None if there are no sales. Use collections.Counter.
The checker sells some coffee from a fresh starter inventory and calls your function.
from collections import Counter
from ridgeline import storage
def best_seller(inv):
return inv.sales[0]
inv = storage.from_dict(storage.STARTER)
inv.sell('RC-101', 2)
print(best_seller(inv))Common mistakes
MistakePutting business rules in the command functions, mixed with input parsing.
InsteadKeep rules in Inventory and models; commands only parse words and call them.
MistakeCalling datetime.now() deep inside the code, so tests cannot control time.
InsteadInject a clock function and pass a fixed one in tests.
MistakeStoring Decimal prices in JSON as floats.
InsteadStore them as strings and convert back with to_money.
MistakeWriting straight over the save file.
InsteadWrite a temporary file and replace the real one in one step.
MistakeCatching every exception in the command loop.
InsteadCatch the errors you expect, InventoryError and ValueError, and let bugs surface.
MistakeLetting modules import each other in both directions.
InsteadKeep imports flowing one way, from the command line down to the data.
Summary and key terms
- The program is split into a package of modules with one job each and imports in one direction.
- Dataclasses with __post_init__ and properties keep every Coffee valid and its derived values current.
- Inventory composes coffees, enforces the rules, raises its own exceptions and supports len, in and iteration.
- Money is Decimal throughout, stored as strings in JSON; every sale records its price and time.
- Injecting a clock makes time-dependent code testable; commands return text so they can be tested and reused.
- A decorator registers commands, and run_line turns expected errors into friendly messages and saves after changes.
- Saves go through a temporary file, and test functions check behaviour, failures and save round trips.
- package
- A folder of modules with an __init__.py, here ridgeline.
- __post_init__
- A dataclass method that runs after the generated __init__, used to validate fields.
- frozen dataclass
- A dataclass whose fields cannot be changed after creation.
- dependency injection
- Passing an object the things it depends on, such as a clock, instead of creating them inside.
- dispatch table
- A dictionary mapping command names to the functions that handle them.
- registration decorator
- A decorator that records a function in a table and returns it unchanged.
- atomic write
- Writing to a temporary file and replacing the original in one step.
- round-trip test
- A test that saves data, loads it back and checks that nothing changed.
- separation of concerns
- Keeping business rules apart from input, output and storage code.
- entry point
- The script, here main.py, that starts a program.
Practice problems
Work each problem on paper first, then reveal the answer to check.
Which module would change if Ridgeline switched from JSON to CSV files?
Show answer
Only storage.py.
Why does Sale store the price instead of a reference to the Coffee?
Show answer
So past sales keep the price actually paid even if the coffee's price changes.
How would you add a command that removes a coffee from sale?
Show answer
Add a remove method to Inventory, then a decorated remove command in cli.py, and add it to CHANGES.
What does the clock parameter make possible in tests?
Show answer
Fixed, predictable times for sales, so results can be checked exactly.
Why is TypeError not caught in run_line?
Show answer
It signals a bug; catching it would hide the problem.
What would a test for restocking check?
Show answer
That the stock rises by the quantity, and that zero, negative or unknown SKUs raise errors and change nothing.
Lesson quiz · 5 questions
Score 60% or more to complete this lesson.
Create a free account to save your progress, take quizzes and earn a certificate.