"""Order rules: ownership, totals, the status machine, idempotent creation, ETags and cursors.""" import base64 import hashlib import json from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from .db import utcnow from .errors import ApiError from .models import IdempotencyRecord, Order, OrderItem, Product from .outbox import order_event from .schemas import OrderIn, OrderOut from .security import Principal TRANSITIONS: dict[str, set[str]] = { "pending": {"paid", "cancelled"}, "paid": {"preparing", "cancelled"}, "preparing": {"ready"}, "ready": {"completed"}, "completed": set(), "cancelled": set(), } TERMINAL = {status for status, nexts in TRANSITIONS.items() if not nexts} # ---------- reading, always scoped to what the caller may see ---------- async def load_order(session: AsyncSession, order_id: int, principal: Principal) -> Order: stmt = select(Order).options(selectinload(Order.items)).where(Order.id == order_id) if not principal.is_admin: stmt = stmt.where(Order.customer_id == principal.id) # other people's orders simply don't exist order = (await session.execute(stmt)).scalar_one_or_none() if order is None: raise ApiError(404, f"order {order_id} not found", type_="not-found") return order def encode_cursor(order_id: int) -> str: return base64.urlsafe_b64encode(json.dumps({"before": order_id}).encode()).decode().rstrip("=") def decode_cursor(cursor: str) -> int: try: padded = cursor + "=" * (-len(cursor) % 4) before = json.loads(base64.urlsafe_b64decode(padded))["before"] if not isinstance(before, int): raise ValueError return before except (ValueError, KeyError, TypeError): raise ApiError(400, "invalid cursor; use the nextCursor value from the previous page", type_="invalid-cursor") from None async def list_orders(session: AsyncSession, principal: Principal, *, status: str | None, limit: int, cursor: str | None) -> tuple[list[Order], str | None]: stmt = select(Order).options(selectinload(Order.items)).order_by(Order.id.desc()).limit(limit + 1) if not principal.is_admin: stmt = stmt.where(Order.customer_id == principal.id) if status: stmt = stmt.where(Order.status == status) if cursor: stmt = stmt.where(Order.id < decode_cursor(cursor)) # keyset: fast on page 1 and page 10,000 rows = list((await session.scalars(stmt)).all()) next_cursor = encode_cursor(rows[limit - 1].id) if len(rows) > limit else None return rows[:limit], next_cursor # ---------- ETags ---------- def etag_of(order: Order) -> str: return f'"{order.id}.{order.version}"' def etag_matches(header: str, etag: str) -> bool: candidates = [value.strip().removeprefix("W/") for value in header.split(",")] return "*" in candidates or etag in candidates def check_if_match(if_match: str | None, order: Order, *, required: bool) -> None: if if_match is None: if required: raise ApiError(428, "send If-Match with the order's current ETag", type_="precondition-required") return if not etag_matches(if_match, etag_of(order)): raise ApiError(412, "the order was changed by someone else; GET it again and retry", type_="precondition-failed", currentEtag=etag_of(order)) # ---------- creating, idempotently ---------- async def _build_order(session: AsyncSession, principal: Principal, body: OrderIn) -> Order: skus = [item.sku for item in body.items] rows = await session.scalars(select(Product).where(Product.sku.in_(skus), Product.active)) products = {product.sku: product for product in rows} missing = sorted(set(skus) - products.keys()) if missing: raise ApiError(422, f"unknown or unavailable products: {', '.join(missing)}", type_="unknown-product") items = [OrderItem(sku=i.sku, name=products[i.sku].name, quantity=i.quantity, unit_price_minor=products[i.sku].price_minor) # prices from the database, never the client for i in body.items] return Order(customer_id=principal.id, note=body.note, status="pending", items=items, total_minor=sum(i.unit_price_minor * i.quantity for i in items)) async def _replay(session: AsyncSession, principal: Principal, key: str, request_hash: str) -> tuple[int, dict] | None: record = await session.get(IdempotencyRecord, (principal.id, key)) if record is None: return None if record.request_hash != request_hash: raise ApiError(422, "this Idempotency-Key was already used for a different request", type_="idempotency-key-reused") return record.status_code, json.loads(record.response_body) async def create_order(session: AsyncSession, principal: Principal, body: OrderIn, idempotency_key: str) -> tuple[int, dict, bool]: """Returns (status, body, replayed). The order, its outbox event and the key commit together.""" request_hash = hashlib.sha256(body.model_dump_json(by_alias=True).encode()).hexdigest() if replay := await _replay(session, principal, idempotency_key, request_hash): return *replay, True order = await _build_order(session, principal, body) session.add(order) await session.flush() # assigns the id and defaults session.add(order_event("order.created", order)) response = OrderOut.model_validate(order).model_dump(mode="json", by_alias=True) session.add(IdempotencyRecord(user_id=principal.id, key=idempotency_key, request_hash=request_hash, status_code=201, response_body=json.dumps(response))) try: await session.commit() except IntegrityError: # a concurrent retry with the same key won await session.rollback() if replay := await _replay(session, principal, idempotency_key, request_hash): return *replay, True raise return 201, response, False # ---------- changing ---------- async def change_status(session: AsyncSession, order: Order, new_status: str) -> Order: if new_status not in TRANSITIONS[order.status]: raise ApiError(409, f"cannot move an order from {order.status} to {new_status}", type_="invalid-transition", allowed=sorted(TRANSITIONS[order.status])) order.status = new_status order.updated_at = utcnow() session.add(order_event("order.status_changed", order)) return order def status_event(order: Order) -> dict: return {"orderId": order.id, "status": order.status, "version": order.version, "updatedAt": order.updated_at.isoformat()}