"""orders — a test brick for erp_platform.scenarios, small but with every shape a real brick has:
* row visibility (a clerk sees their own orders; a manager every one) — another clerk gets 404;
* optimistic versions (two people saving one order: one wins, the other 409);
* money, exact (NUMERIC(18,2), Decimal through both doors);
* a due date by a working-day timing rule in Europe/London, from work.now() — DST-correct;
* a tick (orders past due become overdue) and the demo clock (erp_platform.democlock);
* a two-step use case: store the order, THEN reserve stock through the stock seam (which can be down);
* a capability (mail.send) that may have no provider — a manual task — or a failing one;
* a public APPROVE link for the customer, single use.
DEFECTS lets the harness's own tests put a known defect back, to prove a scenario catches it.
"""
import json
from contextlib import contextmanager
from datetime import timedelta, timezone
from decimal import ROUND_HALF_UP, Decimal
from typing import Optional
from pydantic import BaseModel
from erp_platform import auth, democlock, public, repository as platform_repository, work
from erp_platform.bricks import Brick, Refused
from erp_platform.capabilities import CapabilitySeam, ManualTask
from erp_platform.seam import Seam
ORDERS = Brick("orders", version="0.1.0", requires=("mail.send",), machine_permissions=("orders:tick:run",),
instructions="Customer orders: place them, reserve their stock, confirm them to the customer.")
STOCK = Seam("stock", "STOCK_BASE_URL", roles=["service"], permissions=["stock:reservation:write"])
MAIL = CapabilitySeam("mail.send")
DUE = work.TimingRule("placed", 3, "09:00", "Europe/London", offset_unit="working_days")
CALENDAR = work.WorkCalendar()
DEFECTS: set = set() # tests only: "others_get_403", "retry_conflict", "owner_from_body", "replay_bumps"
democlock.mount(ORDERS)
@contextmanager
def _repository_default():
from erp_platform import db
from scn_orders.infrastructure.tables import OrderRepository
with db.engine.begin() as conn:
yield OrderRepository(conn)
repository = _repository_default
class LineIn(BaseModel):
sku: str
quantity: int
unit_price: Decimal
def _iso(value) -> Optional[str]:
if value is None:
return None
return work.aware(value).astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _money(value) -> Decimal:
return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
def _view(o: dict) -> dict:
lines = [{"line_no": ln["line_no"], "sku": ln["sku"], "quantity": int(ln["quantity"]),
"unit_price": _money(ln["unit_price"]), "amount": _money(Decimal(ln["quantity"]) * _money(ln["unit_price"]))}
for ln in o["lines"]]
return {"order_id": o["order_id"], "customer": o["customer"], "owner": o["owner"], "status": o["status"],
"currency": o["currency"], "placed_at": _iso(o["placed_at"]), "due_at": _iso(o["due_at"]),
"reserved": bool(o["reserved"]), "reserve_note": o["reserve_note"], "note": o["note"],
"version": int(o["version"] or 0), "lines": lines,
"total": _money(sum((ln["amount"] for ln in lines), Decimal("0")))}
def _visible(repo, caller: dict, order_id: str) -> dict:
o, scope = repo.get(order_id), auth.owner_scope(caller)
if o is None or (scope is not None and o["owner"] != scope):
if o is not None and "others_get_403" in DEFECTS:
raise Refused(403, "not your order")
raise Refused(404, f"no order {order_id}")
return o
# ── placing an order: two steps, each its own unit of work ─────────────────────────────────
def _store_new(caller: dict, order_id: str, customer: str, lines: list, currency: str, owner: Optional[str]) -> dict:
"""Step 1: the order itself. Placing the same id again returns what is stored — also when a concurrent
placement of the same id won the version claim (the loser reads what the winner wrote)."""
try:
return _create(caller, order_id, customer, lines, currency, owner)
except Exception as exc: # noqa: BLE001 — re-raised unless a lost claim
if platform_repository.as_conflict(exc) is None:
raise
return _create(caller, order_id, customer, lines, currency, owner)
def _create(caller: dict, order_id: str, customer: str, lines: list, currency: str, owner: Optional[str]) -> dict:
with repository() as repo:
existing = repo.get(order_id)
if existing is not None:
scope = auth.owner_scope(caller)
if scope is not None and existing["owner"] != scope:
raise Refused(409, f"order id {order_id} is taken")
return existing
if not lines:
raise Refused(422, "an order needs at least one line")
placed = work.now()
who = owner if ("owner_from_body" in DEFECTS and owner) else auth.resolve_owner(caller, owner)
order = {"order_id": order_id, "customer": customer, "owner": who, "status": "open", "currency": currency,
"placed_at": placed, "due_at": DUE.due_at({"placed": placed}, CALENDAR), "reserved": False,
"reserve_note": None, "note": None, "version": 0,
"lines": [{"order_id": order_id, "line_no": i, "sku": ln.sku, "quantity": ln.quantity,
"unit_price": _money(ln.unit_price)} for i, ln in enumerate(lines, 1)]}
order["version"] = repo.save(order, expected_version=0)
return order
def _reserve(caller: dict, order: dict) -> dict:
"""Step 2: reserve each line through the stock seam. A neighbour that is down is a fact on the order."""
notes = []
for ln in order["lines"]:
r = STOCK.post("/reservations", {"reservation_key": f"{order['order_id']}|{ln['line_no']}", "sku": ln["sku"],
"quantity": int(ln["quantity"])}, caller.get("email", ""))
if not r:
notes.append(r.reason[:120] or f"stock answered {r.status}")
reserved, note = not notes, ("; ".join(notes) or None)
try:
with repository() as repo:
current = repo.get(order["order_id"])
if (bool(current["reserved"]), current["reserve_note"]) == (reserved, note) or current["reserved"]:
return current # nothing changed: no save, so a replay answers the same
current.update(reserved=reserved, reserve_note=note)
current["version"] = repo.save(current, expected_version=int(current["version"] or 0))
return current
except Exception as exc: # noqa: BLE001 — a concurrent placement saved first: take its word
if platform_repository.as_conflict(exc) is None:
raise
with repository() as repo:
return repo.get(order["order_id"])
@ORDERS.use_case(kind="command", route="POST /orders", permission="orders:order:write", idempotent=True)
def place_order(caller: dict, order_id: str, customer: str, lines: list[LineIn], currency: str = "EUR",
owner: Optional[str] = None) -> dict:
"""Place an order (the caller owns it; a manager may name the owner) and reserve its stock. Due three
working days later at 09:00 London time. Placing the same order id again returns the order, reserving
what is not reserved yet."""
order = _store_new(caller, order_id, customer, lines, currency, owner)
if not order["reserved"]:
order = _reserve(caller, order)
if "replay_bumps" in DEFECTS:
with repository() as repo:
order = repo.get(order_id)
order["version"] = repo.save(order, expected_version=int(order["version"]))
return _view(order)
@ORDERS.use_case(kind="query", route="GET /orders")
def list_orders(caller: dict) -> dict:
"""The orders the caller may see: their own, or every one for a manager."""
with repository() as repo:
return {"orders": repo.ids(auth.owner_scope(caller))}
@ORDERS.use_case(kind="query", route="GET /orders/{order_id}")
def get_order(caller: dict, order_id: str) -> dict:
"""One order with its lines, total and due date, if the caller may see it."""
with repository() as repo:
return _view(_visible(repo, caller, order_id))
@ORDERS.use_case(kind="command", route="POST /orders/{order_id}/note", permission="orders:order:write")
def update_note(caller: dict, order_id: str, note: str, expected_version: int) -> dict:
"""Change the order's note. Pass the version you opened: if someone saved since, this is refused 409."""
if "retry_conflict" in DEFECTS: # the defect BRICKS.md forbids: retrying a Conflict = last writer wins
for _ in range(20):
try:
with repository() as repo:
o = _visible(repo, caller, order_id)
o["note"] = note
o["version"] = repo.save(o, expected_version=int(o["version"]))
return _view(o)
except Exception as exc: # noqa: BLE001
if platform_repository.as_conflict(exc) is None:
raise
with repository() as repo:
o = _visible(repo, caller, order_id)
o["note"] = note
o["version"] = repo.save(o, expected_version=expected_version)
return _view(o)
@ORDERS.use_case(kind="command", route="POST /orders/{order_id}/approve", permission="orders:order:approve",
idempotent=True)
def approve_order(caller: dict, order_id: str) -> dict:
"""Approve an open or overdue order. Approving again changes nothing."""
with repository() as repo:
o = _visible(repo, caller, order_id)
if o["status"] == "approved":
return _view(o)
if o["status"] not in ("open", "overdue"):
raise Refused(409, f"a {o['status']} order cannot be approved")
o["status"] = "approved"
o["version"] = repo.save(o, expected_version=int(o["version"]))
return _view(o)
@ORDERS.use_case(kind="command", route="POST /orders/{order_id}/confirmation", permission="orders:order:write",
idempotent=True)
def send_confirmation(caller: dict, order_id: str) -> dict:
"""Send the customer their order confirmation — by mail.send when it is provided, otherwise a task for a
person with exactly what to send."""
with repository() as repo:
o = _view(_visible(repo, caller, order_id))
key = f"confirm|{order_id}"
message = {"from_email": "orders@shop.example", "to": [f"{o['customer'].lower()}@customer.example"], "cc": [],
"subject": f"Order {order_id}", "body_text": f"Your order {order_id}: {o['total']} {o['currency']}.",
"body_html": None}
with repository() as repo:
task = repo.task(key)
# A step a person was given stays theirs: a provider wired (or reconnected) later must not do it a second time.
r = MAIL.call(key, message, actor=caller.get("email", "")) if task is None else None
if task is not None or isinstance(r, ManualTask):
with repository() as repo:
task = repo.task(key)
if task is None:
row = r.as_row()
task = {"task_key": key, "order_id": order_id, "capability": row["capability"],
"input_json": row["input_json"], "plan": row["plan"], "reason": row["reason"], "state": "open"}
repo.add_task(task)
return {"order_id": order_id, "delivery": "manual",
"task": {"capability": task["capability"], "input": json.loads(task["input_json"]),
"plan": task["plan"], "reason": task["reason"], "state": task["state"]}}
if r.status == "done":
return {"order_id": order_id, "delivery": "sent", "message_id": r.output["message_id"]}
return {"order_id": order_id, "delivery": r.status, "retryable": r.retryable, "message": r.message}
@ORDERS.use_case(kind="query", route="GET /tasks", permission="orders:order:write")
def list_tasks(caller: dict) -> dict:
"""The manual tasks on the orders the caller may see."""
with repository() as repo:
rows = repo.tasks_for(repo.ids(auth.owner_scope(caller)))
return {"tasks": [{"order_id": t["order_id"], "capability": t["capability"], "plan": t["plan"],
"state": t["state"]} for t in rows]}
@ORDERS.use_case(kind="command", route="POST /orders/{order_id}/approval-link", permission="orders:order:write")
def request_customer_approval(caller: dict, order_id: str, customer_email: str) -> dict:
"""Make the link the customer approves the order with — shown ONCE, to copy into a mail. A new link for
the same person replaces the old one."""
with repository() as repo:
_visible(repo, caller, order_id)
issued = ORDERS.links.issue("order", order_id, customer_email, "APPROVE", expires_in=timedelta(days=7),
actor=caller.get("email", ""))
return {"order_id": order_id, "link_id": issued.link_id, "url": issued.url}
@ORDERS.use_case(kind="command", route="POST /tick", permission="orders:tick:run", idempotent=True)
def tick(caller: dict) -> dict:
"""The scheduler's tick: open orders past their due time become overdue. Returns those it changed now."""
now = work.now()
changed = []
with repository() as repo:
due = repo.open_due(now)
for order_id in due:
with repository() as repo:
o = repo.get(order_id)
if o["status"] == "open":
o["status"] = "overdue"
o["version"] = repo.save(o, expected_version=int(o["version"]))
changed.append(order_id)
return {"overdue": changed}
# ── the customer's link ────────────────────────────────────────────────────────────────────
def customer_approves(caller: dict, order_id: str) -> dict:
"""The customer approves through their link: only the order the link was issued for."""
if public.subject_of(caller) != ("order", order_id, "APPROVE"):
raise Refused(404, "not found")
with repository() as repo:
o = repo.get(order_id)
if o is None:
raise Refused(404, "not found")
if o["status"] == "customer_approved":
return _view(o)
o["status"] = "customer_approved"
o["version"] = repo.save(o, expected_version=int(o["version"]))
return _view(o)
@ORDERS.public("APPROVE", subject_kind="order")
def approval_page(link, req):
with repository() as repo:
o = repo.get(link.subject_id)
total = _view(o)["total"] if o else ""
return public.Page(f"Order {public.esc(link.subject_id)}",
f"<p>Total {public.esc(str(total))}</p><form method='post'><button>Approve</button></form>")
@approval_page.act
def approve_by_link(link, req):
o = customer_approves(link.actor, link.subject_id)
return public.Page("Approved", f"<p>Thank you — order {public.esc(o['order_id'])} is approved.</p>")