"""The use cases — what a rep, the deal desk, or their agents can do with quotes.
Each is declared ONCE and served as an API route and as an MCP tool. Each does four things and no more:
1. who may the permission on the declaration (checked before it runs) and the row rules below
2. load the whole quote, one read through the repository
3. ask the rules domain/quote.py decides, and raises if a rule is broken
4. save + record one idempotent write of the aggregate, and one history row, in one transaction
No business rule lives here. If one starts creeping in, the aggregate is missing a method.
WHO SEES WHICH QUOTE (CRM spec: "a rep sees rows they own; a manager sees the team's"):
* a rep sees and changes their own quotes;
* whoever holds quote:quote:read_all (a manager) sees every quote, and quote:quote:manage changes any —
permissions, not role names (0.3.0; roles.yaml: the default roles);
* the deal desk — anyone holding quote:quote:approve — sees every quote, because nobody can approve
what they cannot see; seeing does not let them edit.
A quote the caller may not see is 404, never 403: a 403 would confirm it exists.
THE DEAL (0.2.0, the CRM audit's C1). A quote attached to a deal is the evidence that deal is won on, so the deal
is looked up through the PIPELINE seam AS THE CALLER: one they may not see is 404 (as in the pipeline), one for
another customer or in another currency is 409, a closed one is 409. One quote in flight per deal (H4): a second
open returns the one in flight with `existing: true` (add_lead's house style), and is 409 when the caller may not
see it — never a second quote silently.
"""
import time
from contextlib import contextmanager
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Literal, Optional
from uuid import uuid4
from erp_platform import auth
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import Conflict, retry_on_conflict
from erp_platform.seam import Seam
from erp_platform.money import CurrencyMismatch, InvalidMoney, Money
from erp_platform.types import Amount, BigId, IsoDate, Name, Note, Percent, Quantity, text
from brick_quote.domain.quote import (ACCEPTED, DECLINED, DiscountPolicy, Line, NeedsApproval,
PricedAt, Quote, QuoteIncomplete, QuoteRuleBroken)
DRAFTING, APPROVING, DECIDING = "quote:quote:draft", "quote:quote:approve", "quote:quote:decide"
DEAL_FACTS = "quote:opportunity:read" # held by the pipeline brick's service identity
POLICY_WRITE, PRICES_WRITE = "quote:discount_policy:write", "quote:price_book:write"
# 0.3.0: a brick decides by PERMISSION, never by role name (erp-platform 0.9.2). Seeing every owner's quotes was the
# `manager`/`analyst` ROLES, changing them the `manager` role; they are now these (roles.yaml: the default roles).
READ_ALL, MANAGE = "quote:quote:read_all", "quote:quote:manage"
Currency, Ref, Code, Month = text(8), text(64), text(64), text(7)
QUOTE = Brick("quote", version="0.3.1", ids_as_strings=True, instructions=(
"Quotes a company sends its customers. A quote has versions; a version is a draft until it is "
"sent, and frozen after. Lines are priced from the price book. A discount above the policy's "
"auto-approve level needs the deal desk's approval before sending; above the maximum it is "
"refused — per line as well as for the whole quote. Money is a string with two decimals, in the quote's "
"currency. bp_id is a 64-bit id, sent and returned as digits. A quote carries a `revision`: pass it back "
"with a change, and a 409 means someone changed it since. A quote on a deal must be for the deal's "
"customer and currency; one deal has one quote in flight."))
# 0.3.0: the process is declared (process.yaml → GET /process/bpmn), as erp-platform ≥ 0.8 requires.
QUOTE.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
from brick_quote.infrastructure.tables import QuoteRepository
with db.engine.begin() as conn:
yield QuoteRepository(conn)
repository = _repository_default # tests and the live check swap this
def _now() -> datetime:
return datetime.now(timezone.utc)
@contextmanager
def _rules():
"""A rule's refusal, in the rule's own words, with the status that tells a caller what to do."""
try:
yield
except NeedsApproval as e:
raise Refused(403, str(e)) from None
except QuoteIncomplete as e:
raise Refused(422, str(e)) from None
except (QuoteRuleBroken, CurrencyMismatch) as e:
raise Refused(409, str(e)) from None
except InvalidMoney as e: # only THIS is the caller's input; any other error is ours → 500
raise Refused(422, str(e)) from None
# ── the customer, asked of the business-partner brick ──────────────────────────────────────
#
# A quote references its customer by bp_id and never writes partner tables. Whether that customer
# exists, and may take new business (not blocked or archived), is the partner brick's to say — asked
# through its route, so its own rules and permission checks run (erp_platform.seam). Three answers:
# installed, reachable refuse an unknown customer (422) or one that cannot transact (409)
# installed, unreachable go ahead and SAY SO in the history: a module restarting must not stop a sale
# not installed the customer keeps partners elsewhere; the bp_id is taken as given
# Checked when a quote is opened and again when it is sent — a customer can be blocked in between.
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=[])
# The deal, asked AS THE CALLER (caller=): may THIS person see it, and is it this customer's, in this currency?
PIPELINE = Seam("pipeline", "PIPELINE_BASE_URL", roles=["service"], permissions=[])
def _deal_fits(caller: dict, opportunity_id: str, bp_id: int, currency: str) -> str:
"""C1: '' when the deal is the caller's to see, this customer's, in this currency and open; else refused.
No pipeline installed: the id is taken as given, and the history says so."""
if not PIPELINE.installed:
return " (deal not checked: no pipeline installed)"
r = PIPELINE.get(f"/opportunities/{opportunity_id}", caller=caller)
if r.status == 404:
raise Refused(404, f"no opportunity {opportunity_id}")
if not r.ok:
raise Refused(503, f"cannot check deal {opportunity_id} right now ({r.reason}); try again — a quote is "
f"attached only to a deal it belongs to")
d = r.data
if int(d["bp_id"]) != int(bp_id):
raise Refused(409, f"opportunity {opportunity_id} is for partner {d['bp_id']}, not {bp_id} — a quote on a "
f"deal is for the deal's customer")
if d.get("currency") != currency:
raise Refused(409, f"opportunity {opportunity_id} is in {d.get('currency')}; this quote would be in "
f"{currency}")
if d.get("is_open") is False:
raise Refused(409, f"opportunity {opportunity_id} is closed ({d.get('stage')}) — a new quote is for an "
f"open deal")
return ""
def _customer_may_transact(caller: dict, bp_id: int) -> str:
"""'' when checked and fine; a note for the history when it could not be checked."""
if not PARTNERS.installed:
return ""
r = PARTNERS.get(f"/partners/{int(bp_id)}", actor=caller.get("email", ""))
if r.status == 404:
raise Refused(422, f"there is no business partner {bp_id} — add the customer to the partner "
f"register first")
if not r.ok:
return f" (customer not checked: {r.reason})"
if not r.data.get("can_transact", False):
raise Refused(409, f"{r.data.get('legal_name', bp_id)} is {r.data.get('status')} in the partner "
f"register, so no new quote can be made or sent")
return ""
def _sees(caller: dict, q: Quote) -> bool:
return (auth.owner_scope(caller, see_all=READ_ALL) is None or q.owner == caller.get("email")
or auth.has_permission(caller, APPROVING))
def _load(repo, caller: dict, quote_id: str, *, write: bool = False) -> Quote:
q = repo.get(quote_id)
if q is None or not _sees(caller, q):
raise Refused(404, f"no quote {quote_id}")
if write and not (q.owner == caller.get("email") or auth.has_permission(caller, MANAGE)):
raise Refused(403, f"quote {quote_id} belongs to {q.owner}; only they or a manager can change it")
return q
def _at(q: Quote, revision: Optional[int]) -> Quote:
"""The quote, if it is still at the revision the caller loaded; a Conflict (409) if not."""
if revision is not None and q.revision != int(revision):
raise Conflict("quote_quotes", {"quote_id": q.quote_id}, int(revision), q.revision)
return q
def _changing(revision: Optional[int], fn):
"""With the caller's revision, a race is their 409; with none, the change is re-run on the latest quote —
eight lines added at once are eight lines (the audit kept one)."""
return fn() if revision is not None else retry_on_conflict(fn)
def _money(m: Money) -> dict:
return {"amount": m.amount, "currency": m.currency}
def _view(q: Quote, policy: DiscountPolicy, version: Optional[int] = None) -> dict:
"""The read model: what a screen or an agent needs, derived figures included, money as strings."""
v = q.version(version)
return {
"quote_id": q.quote_id, "reference": q.reference, "bp_id": q.bp_id,
"opportunity_id": q.opportunity_id, "owner": q.owner, "currency": q.currency,
"version": v.version, "versions": [x.version for x in q.versions],
"status": v.status, "state": v.state(), "valid_until": v.valid_until, "sent_at": v.sent_at,
"total": v.totals.get(q.currency, Money.zero(q.currency)).amount,
"list_total": v.list_totals.get(q.currency, Money.zero(q.currency)).amount,
"discount_pct": v.discount_pct(), "net_discount_pct": v.net_discount_pct(),
"deepest_line_discount_pct": v.deepest_line_discount_pct(), "revision": q.revision,
"needs_approval": v.version == q.latest.version and q.needs_approval(policy),
"approval": ({"approved_by": v.approval.approved_by, "approved_pct": v.approval.approved_pct}
if v.approval else None),
"lines": [{"line_no": ln.line_no, "product_id": ln.product_id, "description": ln.description,
"quantity": ln.quantity, "list_price": ln.list_price.amount,
"unit_price": ln.unit_price.amount, "line_total": ln.line_total.amount,
"discount_pct": ln.discount_pct,
"price_book_id": ln.priced_at.price_book_id, "priced_on": ln.priced_at.priced_on,
"schedule": {m: a.amount for m, a in ln.schedule().items()}}
for ln in v.lines],
}
def _changed(repo, caller: dict, q: Quote, action: str, detail: str = "",
version: Optional[int] = None) -> dict:
repo.save(q)
repo.record(q.quote_id, action, caller["email"], detail)
return _view(q, repo.policy(), version)
# ── reading ────────────────────────────────────────────────────────────────────────────────
@QUOTE.use_case(kind="query", route="GET /quotes")
def list_quotes(caller: dict, mine: bool = False, state: Optional[Code] = None) -> dict:
"""List the quotes the caller may see, newest version of each, optionally filtered by state
(DRAFT, SENT, EXPIRED, ACCEPTED, DECLINED, SUPERSEDED). `mine` narrows to the caller's own."""
with repository() as repo:
sees_all = auth.has_permission(caller, READ_ALL) or auth.has_permission(caller, APPROVING)
owner = caller.get("email") if (mine or not sees_all) else None
policy, out = repo.policy(), []
for qid in repo.ids(owner=owner):
q = repo.get(qid)
if q is None:
continue
view = _view(q, policy)
if state is None or view["state"] == state.upper():
out.append({k: view[k] for k in ("quote_id", "reference", "bp_id", "owner",
"version", "state", "total", "currency",
"discount_pct", "needs_approval", "revision")})
return {"quotes": out}
@QUOTE.use_case(kind="query", route="GET /quotes/{quote_id}", etag="revision")
def get_quote(caller: dict, quote_id: Ref, version: Optional[int] = None) -> dict:
"""One quote — its latest version unless `version` is given — with lines, totals, the discount the
policy reads (the deepest line's or the net's), whether it needs approval, each line's revenue schedule,
and its `revision` (send it back with a change)."""
with repository() as repo, _rules():
return _view(_load(repo, caller, quote_id), repo.policy(), version)
@QUOTE.use_case(kind="query", route="GET /quotes/{quote_id}/history")
def quote_history(caller: dict, quote_id: Ref) -> dict:
"""Who did what to a quote, and when — every line priced, discount approved, send and decision."""
with repository() as repo:
_load(repo, caller, quote_id)
return {"quote_id": quote_id, "events": repo.history(quote_id)}
@QUOTE.use_case(kind="query", route="GET /approvals", permission=APPROVING)
def approval_queue(caller: dict) -> dict:
"""The deal desk's queue: drafts whose discount is above the auto-approve level and not yet
covered by an approval. The quote's owner cannot approve their own."""
with repository() as repo:
policy, out = repo.policy(), []
for qid in repo.ids():
q = repo.get(qid)
if q is not None and q.needs_approval(policy):
out.append({"quote_id": q.quote_id, "owner": q.owner, "version": q.latest.version,
"discount_pct": q.latest.discount_pct(),
"total": q.latest.totals.get(q.currency, Money.zero(q.currency)).amount,
"currency": q.currency})
return {"auto_approve_pct": policy.auto_approve_pct, "max_pct": policy.max_pct,
"waiting": out}
@QUOTE.use_case(kind="query", route="GET /opportunities/{opportunity_id}/quote-facts",
permission=DEAL_FACTS)
def opportunity_quote_facts(caller: dict, opportunity_id: Ref) -> dict:
"""What a deal's quotes establish — for the pipeline brick deciding whether a deal can be won: has
any been sent, is one accepted, and the accepted total per currency; and (0.2.0) each accepted quote's
customer, currency and total, so the pipeline counts only an acceptance by the deal's own customer.
Facts only, no quote contents, so it needs its own permission rather than row visibility."""
with repository() as repo, _rules():
sent = accepted = False
totals: dict = {}
each = []
for qid in repo.ids(opportunity_id=opportunity_id):
q = repo.get(qid)
if q is None:
continue
sent = sent or any(v.sent_at for v in q.versions)
v = q.accepted()
if v is not None:
accepted = True
for cur, m in v.totals.items():
totals[cur] = totals.get(cur, Money.zero(cur)) + m
each.append({"quote_id": q.quote_id, "bp_id": q.bp_id, "currency": q.currency,
"total": v.totals.get(q.currency, Money.zero(q.currency)).amount,
"version": v.version})
return {"opportunity_id": opportunity_id, "has_sent_quote": sent or accepted,
"has_accepted_quote": accepted,
"accepted_totals": {c: m.amount for c, m in sorted(totals.items())}, "accepted": each}
# ── drafting ───────────────────────────────────────────────────────────────────────────────
@QUOTE.use_case(kind="command", route="POST /quotes", permission=DRAFTING)
def open_quote(caller: dict, bp_id: BigId, currency: Currency, opportunity_id: Optional[Ref] = None,
valid_until: Optional[IsoDate] = None, reference: Optional[Name] = None) -> dict:
"""Start a quote for a customer (a business partner's bp_id), owned by the caller. Refused if the
partner register says the customer does not exist or may not take new business. On a deal
(`opportunity_id`): the deal must be one you can see (404), for this customer and in this currency, and
open (409). One quote in flight per deal: if it already has a draft or sent quote you can see, that one is
returned with `existing: true`; one you cannot see is refused 409. A new quote says `existing: false`."""
with _rules():
cur = Money.zero(currency).currency
checked = _customer_may_transact(caller, bp_id)
if opportunity_id:
checked += _deal_fits(caller, opportunity_id, bp_id, cur)
try:
return _open_quote(caller, bp_id, cur, opportunity_id, valid_until, reference, checked)
except Exception as e: # noqa: BLE001 — re-raised unless the race
from brick_quote.infrastructure.tables import lost_deal_slot
if not lost_deal_slot(e):
raise
# Lost the race to a concurrent open for this deal (it counted the same quotes; its claim collided at the
# INSERT or at COMMIT). Answer the winner's quote, as if this open had come second.
for attempt in range(6):
with repository() as repo, _rules():
found = _in_flight(repo, caller, repo.ids(opportunity_id=opportunity_id))
if found is not None:
return found
time.sleep(0.1 * (attempt + 1))
raise Refused(409, f"a quote for {opportunity_id} is being opened by another request; try again")
def _in_flight(repo, caller: dict, quote_ids: list) -> Optional[dict]:
"""The deal's quote in flight (draft or sent), as `existing: true` — or 409 if the caller may not see it."""
for qid in quote_ids:
q = repo.get(qid)
if q is not None and q.is_open():
if not _sees(caller, q):
raise Refused(409, "this deal already has a quote in flight, owned by someone else — ask them or "
"a manager to revise it")
return {**_view(q, repo.policy()), "existing": True}
return None
def _open_quote(caller, bp_id, currency, opportunity_id, valid_until, reference, checked) -> dict:
with repository() as repo, _rules():
quote_id = f"Q-{uuid4().hex[:10].upper()}"
if opportunity_id:
on_deal = repo.ids(opportunity_id=opportunity_id) # ONE read: the in-flight check and the count
found = _in_flight(repo, caller, on_deal)
if found is not None:
return found
repo.claim_deal_slot(opportunity_id, len(on_deal) + 1, quote_id) # first: a loser writes nothing
q = Quote.open(quote_id, bp_id, currency, caller["email"], opportunity_id=opportunity_id,
reference=reference or quote_id, valid_until=valid_until)
return {**_changed(repo, caller, q, "opened", f"for partner {bp_id} in {q.currency}{checked}"),
"existing": False}
@QUOTE.use_case(kind="command", route="POST /quotes/{quote_id}/versions/{version}/lines",
permission=DRAFTING, if_match="revision")
def add_line(caller: dict, quote_id: Ref, version: int, product_id: Code, quantity: Quantity = 1,
unit_price: Optional[Amount] = None, price_book_id: Code = "standard",
priced_on: Optional[IsoDate] = None, description: Optional[Note] = None,
revenue_months: Optional[list[Month]] = None, revision: Optional[int] = None) -> dict:
"""Add a product to a draft, priced from the price book in the quote's currency as of `priced_on`
(default today). `unit_price` is the agreed price; leave it out to sell at list (above list is allowed;
no line may be further below list than the policy's maximum). `revenue_months` (YYYY-MM) spreads the line
total across the months it is recognised in. `revision` (or If-Match): the one you read."""
return _changing(revision, lambda: _add_line(caller, quote_id, version, product_id, quantity, unit_price,
price_book_id, priced_on, description, revenue_months, revision))
def _add_line(caller, quote_id, version, product_id, quantity, unit_price, price_book_id, priced_on, description,
revenue_months, revision) -> dict:
with repository() as repo, _rules():
q = _at(_load(repo, caller, quote_id, write=True), revision)
on = priced_on or date.today().isoformat()
found = repo.price(price_book_id, product_id, q.currency, on)
if found is None:
raise Refused(422, f"price book {price_book_id} has no {q.currency} price for {product_id} "
f"on {on}. Add one, or quote it from another price book.")
listed = Money.of(found["list_price"], q.currency)
line = Line(0, product_id, quantity, listed,
listed if unit_price is None else Money.of(unit_price, q.currency),
PricedAt(price_book_id, on), description or found["description"],
tuple(revenue_months or ()))
added = q.add_line(version, line, repo.policy())
return _changed(repo, caller, q, "line added",
f"v{version} line {added.line_no}: {quantity} × {product_id} at "
f"{added.unit_price} (list {added.list_price})", version)
@QUOTE.use_case(kind="command", route="PATCH /quotes/{quote_id}/versions/{version}/lines/{line_no}",
permission=DRAFTING, idempotent=True, if_match="revision")
def edit_line(caller: dict, quote_id: Ref, version: int, line_no: int,
quantity: Optional[Quantity] = None, unit_price: Optional[Amount] = None,
description: Optional[Note] = None, revenue_months: Optional[list[Month]] = None,
revision: Optional[int] = None) -> dict:
"""Change a draft line's quantity, agreed price, description or revenue months. Fields left out
are unchanged. The product and its list price cannot change — remove the line and add another."""
fields = {k: v for k, v in {"quantity": quantity, "unit_price": unit_price,
"description": description, "revenue_months": revenue_months}.items()
if v is not None}
def run():
with repository() as repo, _rules():
q = _at(_load(repo, caller, quote_id, write=True), revision)
before, after = q.edit_line(version, line_no, repo.policy(), **fields)
moved = [f"{k} {getattr(before, k)} → {getattr(after, k)}" for k in fields
if getattr(before, k) != getattr(after, k)]
return _changed(repo, caller, q, "line edited",
f"v{version} line {line_no}: " + ("; ".join(moved) or "no change"), version)
return _changing(revision, run)
@QUOTE.use_case(kind="command", route="DELETE /quotes/{quote_id}/versions/{version}/lines/{line_no}",
permission=DRAFTING, destructive=True, if_match="revision")
def remove_line(caller: dict, quote_id: Ref, version: int, line_no: int, revision: Optional[int] = None) -> dict:
"""Remove a line from a draft."""
def run():
with repository() as repo, _rules():
q = _at(_load(repo, caller, quote_id, write=True), revision)
gone = q.remove_line(version, line_no)
return _changed(repo, caller, q, "line removed",
f"v{version} line {line_no}: {gone.quantity} × {gone.product_id}", version)
return _changing(revision, run)
# ── approval, sending, the customer's answer ───────────────────────────────────────────────
@QUOTE.use_case(kind="command", route="POST /quotes/{quote_id}/versions/{version}/approve",
permission=APPROVING, idempotent=True, if_match="revision")
def approve_discount(caller: dict, quote_id: Ref, version: int, revision: Optional[int] = None) -> dict:
"""Approve the discount a draft carries now — the deepest line's or the net's, whichever is larger. The
approval covers that percentage only — if the draft is later discounted further, it needs approving again.
Nobody approves their own quote."""
def run():
with repository() as repo, _rules():
q = _at(_load(repo, caller, quote_id), revision)
approval = q.approve_discount(version, caller["email"], repo.policy())
return _changed(repo, caller, q, "discount approved",
f"v{version} at {approval.approved_pct}%", version)
return _changing(revision, run)
@QUOTE.use_case(kind="command", route="POST /quotes/{quote_id}/versions/{version}/send",
permission=DRAFTING, if_match="revision")
def send_quote(caller: dict, quote_id: Ref, version: int, revision: Optional[int] = None) -> dict:
"""Send a draft to the customer. After this it cannot change; changes are a new version. Refused
with the figures when the discount needs an approval it does not have."""
def run():
with repository() as repo, _rules():
q = _at(_load(repo, caller, quote_id, write=True), revision)
checked = _customer_may_transact(caller, q.bp_id)
q.send(version, repo.policy(), _now().isoformat())
return _changed(repo, caller, q, "sent", f"v{version}{checked}", version)
return _changing(revision, run)
@QUOTE.use_case(kind="command", route="POST /quotes/{quote_id}/versions/{version}/decision",
permission=DECIDING, idempotent=True, if_match="revision")
def record_decision(caller: dict, quote_id: Ref, version: int,
outcome: Literal["ACCEPTED", "DECLINED"], revision: Optional[int] = None) -> dict:
"""Record the customer's answer to a sent version. Accepting supersedes the quote's other open
versions. Recording the same answer again changes nothing."""
def run():
with repository() as repo, _rules():
q = _at(_load(repo, caller, quote_id, write=True), revision)
if q.version(version).status == outcome:
return _view(q, repo.policy(), version)
q.decide(version, outcome)
return _changed(repo, caller, q, outcome.lower(), f"v{version}", version)
return _changing(revision, run)
@QUOTE.use_case(kind="command", route="POST /quotes/{quote_id}/revise", permission=DRAFTING, if_match="revision")
def revise_quote(caller: dict, quote_id: Ref, valid_until: Optional[IsoDate] = None,
revision: Optional[int] = None) -> dict:
"""Make a new draft version from the latest sent one, carrying its lines forward. The sent version
stays exactly as the customer read it."""
def run():
with repository() as repo, _rules():
q = _at(_load(repo, caller, quote_id, write=True), revision)
before = q.latest.version
nxt = q.revise(valid_until)
return _changed(repo, caller, q, "revised", f"v{before} → v{nxt.version}", nxt.version)
return _changing(revision, run)
# ── configuration: the deal desk's rows ────────────────────────────────────────────────────
@QUOTE.use_case(kind="query", route="GET /discount-policy")
def get_discount_policy(caller: dict) -> dict:
"""The discount a rep may give without approval, and the maximum no approval can exceed."""
with repository() as repo:
p = repo.policy()
return {"auto_approve_pct": p.auto_approve_pct, "max_pct": p.max_pct}
@QUOTE.use_case(kind="command", route="PUT /discount-policy", permission=POLICY_WRITE,
idempotent=True)
def set_discount_policy(caller: dict, auto_approve_pct: Percent, max_pct: Percent) -> dict:
"""Set the discount policy. Applies to quotes not yet sent; a sent quote keeps the terms it went
out on."""
with repository() as repo, _rules():
p = DiscountPolicy(auto_approve_pct, max_pct)
repo.set_policy(p, caller["email"])
return {"auto_approve_pct": p.auto_approve_pct, "max_pct": p.max_pct}
@QUOTE.use_case(kind="query", route="GET /prices/{price_book_id}/{product_id}")
def find_price(caller: dict, price_book_id: Code, product_id: Code, currency: Currency,
on: Optional[IsoDate] = None) -> dict:
"""The list price of a product in a price book and currency, as of a date (default today)."""
on = on or date.today().isoformat()
with repository() as repo:
found = repo.price(price_book_id, product_id, currency.upper(), on)
if found is None:
raise Refused(404, f"no {currency.upper()} price for {product_id} in {price_book_id} on {on}")
return {"price_book_id": price_book_id, "product_id": product_id,
"currency": currency.upper(), "on": on, **found}
@QUOTE.use_case(kind="command", route="PUT /prices/{price_book_id}/{product_id}",
permission=PRICES_WRITE, idempotent=True)
def set_price(caller: dict, price_book_id: Code, product_id: Code, currency: Currency, valid_from: IsoDate,
list_price: Amount, description: Optional[Name] = None) -> dict:
"""Add a list price from a date. A price is never changed in place — a quote priced on an earlier
date must re-price the same forever — so a different price needs a later valid_from."""
with repository() as repo, _rules():
m = Money.of(list_price, currency)
existing = repo.price_at(price_book_id, product_id, m.currency, valid_from)
if existing is not None and existing != m.amount:
raise Refused(409, f"{product_id} already has a {m.currency} price of {existing} from "
f"{valid_from}; add the new price from a later date")
repo.set_price(price_book_id, product_id, m.currency, valid_from, m.amount,
description or "", caller["email"])
return {"price_book_id": price_book_id, "product_id": product_id, "currency": m.currency,
"valid_from": valid_from, "list_price": m.amount}
# keep the outcome names importable for callers that do not want string literals
OUTCOMES = (ACCEPTED, DECLINED)