"""The use cases — what a person (or their agent) can do with quotes.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and
nowhere else:
* **who may** — the `permission` on the declaration, checked before the use case runs;
* **which rows** — `_visible()`, from the caller. A quote the caller may not see is a 404, never a
403: a 403 would confirm it exists.
Each use case opens one unit of work: load the aggregate, let the domain decide, save it whole.
"""
from contextlib import contextmanager
from decimal import Decimal
from typing import Optional
from erp_platform import auth
from erp_platform import repository as repository_
from erp_platform.bricks import Brick, Refused
from sqlalchemy.exc import IntegrityError
from brick_example.domain.quote import Quote, RuleBroken
READ_ALL = "example:quote:read_all" # who sees every owner's quotes: a permission, never a role name
EXAMPLE = Brick("example", version="0.1.0",
instructions="Quotes a company sends its customers: draft them, add lines, send, "
"and record the customer's acceptance.")
# The process view (BRICKS.md, "The process view"): the quote workflow as BPMN, GET /process/bpmn.
EXAMPLE.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
"""A duplicate key — but not a version claim, which the doors already answer as 'changed since you opened it'."""
if repository_.as_conflict(exc) is not None:
return False
text = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a racer's duplicate key (two people creating the same natural key at once) can
surface at COMMIT, after the use case has returned — so it is caught here, around the commit, and answered as
the retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_example.infrastructure.tables import QuoteRepository
try:
with engine.begin() as conn:
yield QuoteRepository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
repository = _repository_default # tests swap in an in-memory one
def _visible(repo, caller: dict, quote_id: str) -> Quote:
q, scope = repo.get(quote_id), auth.owner_scope(caller, see_all=READ_ALL)
if q is None or (scope is not None and q.owner != scope):
raise Refused(404, f"no quote {quote_id}")
return q
def _view(q: Quote) -> dict:
return {"quote_id": q.quote_id, "customer": q.customer, "owner": q.owner,
"status": q.status, "currency": q.currency, "total": q.total,
"lines": [{"line_no": ln.line_no, "description": ln.description,
"quantity": ln.quantity, "unit_price": ln.unit_price,
"amount": ln.amount} for ln in q.lines]}
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words, through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
@EXAMPLE.use_case(kind="query", route="GET /quotes")
def list_quotes(caller: dict, mine: bool = False) -> dict:
"""List the quotes the caller may see. Managers see every quote; everyone else sees their own."""
with repository() as repo:
return {"quotes": repo.ids_owned_by(auth.owner_scope(caller, mine, see_all=READ_ALL))}
@EXAMPLE.use_case(kind="query", route="GET /quotes/{quote_id}")
def get_quote(caller: dict, quote_id: str) -> dict:
"""One quote with its lines and total, if the caller may see it."""
with repository() as repo:
return _view(_visible(repo, caller, quote_id))
@EXAMPLE.use_case(kind="command", route="POST /quotes", permission="example:quote:write",
idempotent=True)
def create_quote(caller: dict, quote_id: str, customer: str, currency: str = "EUR") -> dict:
"""Start a draft quote for a customer. The caller owns it. Creating the same id again returns it."""
with repository() as repo:
existing = repo.get(quote_id)
if existing is not None:
if existing.owner != caller["email"]:
raise Refused(409, f"quote id {quote_id} is taken")
return _view(existing)
q = Quote(quote_id=quote_id, customer=customer, owner=caller["email"], currency=currency)
repo.save(q)
return _view(q)
@EXAMPLE.use_case(kind="command", route="POST /quotes/{quote_id}/lines",
permission="example:quote:write")
def add_line(caller: dict, quote_id: str, description: str, quantity: int,
unit_price: Decimal) -> dict:
"""Add a line to a draft quote. Prices are exact amounts in the quote's currency."""
with repository() as repo, _decide():
q = _visible(repo, caller, quote_id)
q.add_line(description, quantity, unit_price)
repo.save(q)
return _view(q)
@EXAMPLE.use_case(kind="command", route="DELETE /quotes/{quote_id}/lines/{line_no}",
permission="example:quote:write", idempotent=False, destructive=True)
def remove_line(caller: dict, quote_id: str, line_no: int) -> dict:
"""Remove a line from a draft quote."""
with repository() as repo, _decide():
q = _visible(repo, caller, quote_id)
q.remove_line(line_no)
repo.save(q)
return _view(q)
@EXAMPLE.use_case(kind="command", route="POST /quotes/{quote_id}/send", permission="example:quote:send")
def send_quote(caller: dict, quote_id: str) -> dict:
"""Send a draft quote to the customer. After this its lines cannot change."""
with repository() as repo, _decide():
q = _visible(repo, caller, quote_id)
q.send()
repo.save(q)
return _view(q)
@EXAMPLE.use_case(kind="command", route="POST /quotes/{quote_id}/accept",
permission="example:quote:accept", idempotent=True)
def accept_quote(caller: dict, quote_id: str) -> dict:
"""Record that the customer accepted a sent quote. Accepting again changes nothing."""
with repository() as repo, _decide():
q = _visible(repo, caller, quote_id)
q.accept()
repo.save(q)
return _view(q)