Marketplace¶
A shop: a searchable catalog, a cart that follows you across pages and tabs, a checkout that can’t oversell, and order history.
marketplace/
├── app.py # composition root
├── settings.py # DATABASE_URL, shop name
├── models/ # database.py, product.py, order.py, account.py, money.py
├── services/ # shop.py (ShopService), seed.py
└── views/
├── components/ # cart.py (CartAware), header, shell, product_card
├── pages/ # catalog.py, checkout.py, account.py
└── styles/
1. Money and history¶
models/money.py¶
def price(cents: int) -> str:
"""``12900`` -> ``"$129.00"``. Money is stored and added up as integer cents, never floats."""
return f"${cents / 100:,.2f}"
Prices are integer cents. Order lines copy the product’s name and price, so an order never
changes when the catalog does. lazy="raise" turns an accidental lazy load (the N+1 query
problem) into an immediate error.
2. A cart in the session¶
class CartAware(Component):
"""Mix into any component that reads or changes the cart."""
cart = State(factory=dict, scope="session") # {"<product id>": quantity}
@property
def item_count(self) -> int:
return sum(self.cart.values())
def quantity(self, product_id: int) -> int:
return self.cart.get(str(product_id), 0)
def set_quantity(self, product: ProductOut, quantity: int) -> None:
cart = dict(self.cart) # reassign a new dict: in-place changes would not be noticed
quantity = max(0, min(quantity, product.stock))
if quantity:
cart[str(product.id)] = quantity
else:
cart.pop(str(product.id), None)
self.cart = cart
def add(self, product: ProductOut, quantity: int = 1) -> None:
self.set_quantity(product, self.quantity(product.id) + quantity)
def empty_cart(self) -> None:
self.cart = {}
def cart_items(self) -> dict[int, int]:
return {int(product_id): quantity for product_id, quantity in self.cart.items()}
State(scope="session") lives in the browser session on the server, shared by every
component that inherits CartAware: the header, the product cards, the cart page. When a card
adds a product, the header’s count updates in the same response. Other tabs catch up on their
next interaction.
class ShopHeader(CartAware):
"""Reads the cart, so it re-renders by itself whenever any component changes it."""
def sign_out(self):
self.context.session.logout() # also forgets the cart: it lives in the session
self.navigate("/")
def render(self):
user = self.context.user
count = self.item_count
return Header(
Row(
Link(settings.SHOP_NAME, href="/", style=BRAND),
Link("Shop", href="/"),
Link("Orders", href="/orders") if user else None,
Spacer(),
Text(f"Hi, {user}", muted=True, size="sm") if user else None,
Button("Sign out", variant="ghost", size="sm", on_click=self.sign_out) if user else None,
Link("Sign in", href="/login") if not user else None,
ButtonLink(
"Cart ",
Badge(str(count), tone="primary" if count else "neutral"),
href="/cart",
variant="outline",
size="sm",
aria_label=f"Cart, {count} item{'s' if count != 1 else ''}",
),
gap="md",
wrap=True,
),
style=HEADER,
)
3. Search as you type¶
@router.page("/")
class Catalog(Page):
title = "Rational Market"
query = State("")
category = State("")
sort = State("name")
products = State(factory=list)
categories = State(factory=list)
async def mount(self):
self.categories = await self.use(ShopService).categories()
await self.search()
async def search(self):
self.products = await self.use(ShopService).products(self.query, self.category, self.sort)
async def set_query(self, value: str): # debounced: one request after typing pauses
self.query = value
await self.search()
async def set_category(self, value: str):
self.category = value if value in self.categories else ""
await self.search()
async def set_sort(self, value: str):
self.sort = value if value in dict(SORTS) else "name"
await self.search()
def render(self):
filters = Row(
Input(
"Search",
name="q",
type="search",
placeholder="Search products",
value=self.query,
on_input=self.set_query,
hide_label=True,
style=SEARCH,
),
Select(
"Category",
[("", "All categories"), *self.categories],
value=self.category,
on_change=self.set_category,
hide_label=True,
),
Select("Sort by", SORTS, value=self.sort, on_change=self.set_sort, hide_label=True),
wrap=True,
align="start",
)
count = len(self.products)
results = (
Grid(*[ProductCard(product=p, key=p.id) for p in self.products], min_column_width=230)
if self.products
else Card(Text("No products match your search."), style=FLAT_CARD)
)
return Shell(
Column(Heading("Shop", level=1), Text("Small things, made well.", muted=True), gap="xs"),
filters,
Text(f"{count} product{'s' if count != 1 else ''}", muted=True, size="sm", aria_live="polite"),
results,
)
4. A checkout that can’t oversell¶
@remote(auth="required")
async def checkout(self, items: dict[int, int], address: str) -> OrderOut:
"""Place an order for ``{product_id: quantity}``. Stock is checked and reserved atomically."""
items = {pid: qty for pid, qty in items.items() if qty > 0}
if not items:
raise RemoteError("Your cart is empty")
if len(address.strip()) < 10:
raise FieldError("address", "Please enter a full shipping address")
async with self.db.transaction() as db: # all or nothing: any error below rolls everything back
products = {p.id: p for p in await db.scalars(select(Product).where(Product.id.in_(items)))}
order = Order(customer=self.user, address=address.strip(), total_cents=0, lines=[])
for product_id, quantity in items.items():
product = products.get(product_id)
if product is None:
raise RemoteError("A product in your cart is no longer available")
if product.stock < quantity:
raise RemoteError(f"Only {product.stock} left of “{product.name}”")
product.stock -= quantity
order.lines.append(
OrderLine(
product_id=product.id,
name=product.name,
quantity=quantity,
price_cents=product.price_cents,
)
)
order.total_cents += quantity * product.price_cents
db.add(order)
await db.flush()
return OrderOut.model_validate(order)
Stock is checked and decremented inside one transaction. If any line fails, nothing is
written. The cart page turns that RemoteError into a friendly message:
async def place_order(self, form: CheckoutForm):
try:
order = await self.use(ShopService).checkout(self.cart_items(), form.address)
except RemoteError as error: # e.g. someone else just bought the last one
self.problem = error.message
self.products = await self.use(ShopService).products_by_id(list(self.cart_items()))
return
self.empty_cart()
self.navigate(f"/orders?placed={order.id}")