dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/application/use_cases.py
"""The use cases — what a person (or their agent, or a neighbouring brick) can do with sales orders.
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 (a PERMISSION on the declaration, never a role's name) and WHICH ROWS (`_can_see`, from the caller: an order
the caller may not see is a 404, never a 403). Each command is one unit of work: load, let the domain decide, save whole
at the version it was loaded (`If-Match` for a screen).
Row visibility (module.yaml): a caller sees the orders they own; so:sales_order:read_all sees every order; a fulfiller
sees confirmed orders, without prices. Prices are projected out for a caller holding neither so:order:write nor
so:sales_order:read_all.
The order is a commitment, separate from the quote. It asks product for every price (find_price), tax for every tax
(compute), payments for the credit hold, and invoicing for every invoice; it keeps none of their rules. Fulfilment and
invoice status are derived on read.
"""
import contextvars
import dataclasses
import json
import re
import uuid
from contextlib import contextmanager
from datetime import date, datetime, timezone
from decimal import Decimal
from fractions import Fraction
from typing import Annotated, Literal, Optional, Union
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from erp_platform import after_response, auth, work
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import retry_on_conflict
from erp_platform.types import BigId, IsoDate, Limit, text
from pydantic import Field
from sqlalchemy.exc import IntegrityError
from brick_sales_order.application import odoo_migration, promising
from brick_sales_order.application.inputs import (
AddressIn,
Amt,
ChangeIn,
Code,
Currency,
Day,
DecisionIn,
DeductionIn,
DifferenceIn,
LineIn,
LineNo,
Offset,
Pct,
PoSuggestionIn,
Qty0,
QuoteLineIn,
Rate,
Reason,
RebateTierIn,
Ref,
ScaleTierIn,
day,
)
from brick_sales_order.domain import gates as gate_rules
from brick_sales_order.domain import pricing as pr
from brick_sales_order.domain import rules as rl
from brick_sales_order.domain.errors import GateAnswer, GateUnmet, OutOfBounds, RuleBroken
from brick_sales_order.domain.readiness import no, require
from brick_sales_order.domain.rebate import RebateAgreement
from brick_sales_order.domain.sales_order import LineSpec, SalesOrder, priced_quantity
from brick_sales_order.domain.values import (
AI_KEYS,
CONTROL_KEYS,
MEMO_TYPES,
RETURN_TYPES,
SETTING_KINDS,
SETTINGS,
money,
num,
render_number,
)
from brick_sales_order.infrastructure import ai_steps, business_time
from brick_sales_order.infrastructure.seams import Missing, RealPartners, ports, price_answer
from brick_sales_order.infrastructure.tables import Duplicate
QuoteVersion = Annotated[int, Field(ge=1, le=100000)]
Mark = Annotated[int, Field(ge=0, le=2_000_000_000)]
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
READ, WRITE, CONFIRM, APPROVE, CANCEL = "so:order:read", "so:order:write", "so:order:confirm", "so:order:approve", "so:order:cancel"
DELIVER, INVOICE = "so:order:deliver", "so:order:invoice"
SETTINGS_WRITE, CONTROLS, READ_ALL = "so:settings:write", "so:settings:controls", "so:sales_order:read_all"
INTAKE, SUGGEST, FINISH = "so:order:intake", "so:order:suggest", "so:order:finish" # machine
DEMAND_READ = "so:demand:read" # machine: manufacturing's MRP-lite (SO-50)
CLAIM, IMPORT = "so:number_series:claim", "so:order:import" # machine
RESCHEDULE, ACCRUE = "so:order:reschedule", "so:rebate:accrue" # machine
PRICING, REBATE, BLOCK = "so:pricing:write", "so:rebate:write", "so:order:block"
MEMO, INSPECT = "so:memo:approve", "so:return:inspect" # a second person's approval; inventory's inspection (machine)
# The spec this build implements: erp-method modules/sales-order v10, signed by seemo 2026-10-25 (erp-method 0.1.38): one POST /prices/batch per
# order (product v7 PRD-41) and, from v9, SO-51 (timesheets delivers approved hours on a service line).
SPEC_VERSION = 10
SO = Brick("sales-order", version="0.2.0", ids_as_strings=True, machine_permissions=(INTAKE, SUGGEST, FINISH, CLAIM, IMPORT, RESCHEDULE, ACCRUE, INSPECT, DEMAND_READ), instructions=(
"Sales orders: what a customer has committed to buy. An accepted quote becomes a draft order (never re-priced); a "
"rep confirms it once every readiness gate is met (get_order_readiness answers {gate, met, reason, fix}); then "
"record deliveries, ask invoicing for what is due (request_invoice, raise_deposit), amend with a reason, or cancel. "
"Lines are priced only by the product module and taxed only by the tax module. Fulfilment and invoice status are "
"derived from quantities, never stored. The order asks invoicing for an invoice; it never writes one."))
SO.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:
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 write conflict on a claim row can surface at COMMIT (the loser's COMMIT raises 23505), after the use
case has returned - so it is caught here, around the commit, and answered as the same retry-safe 409 a conflict seen mid-use-case
gets, never a 500 and never a second number."""
from brick_sales_order.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(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 one on SQLite
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words (a gate: its reason and fix), through both doors."""
try:
yield
except OutOfBounds as broken:
raise Refused(422, str(broken)) from None
except GateAnswer as broken:
g = broken.gates[0]
raise Refused(409, g.reason, gate=g.gate, fix=g.fix) from None
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
except Duplicate as dup:
raise Refused(409, f"{dup}: a correction is a new row from a later date, never an edit") from None
def _who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _ids(caller: dict) -> tuple:
"""Every identity the caller has (email and sub, lower-case): the same person under either is the same person."""
return tuple(dict.fromkeys(x.strip().lower() for x in (caller.get("email"), caller.get("sub")) if x and x.strip()))
def _scope(caller: dict, mine: bool = False) -> Optional[str]:
"""Whose rows the caller may see: None (all) ONLY with read_all and not narrowing to mine; otherwise their own identity,
the one orders are stamped with (email, else sub). A token with no email is scoped by its sub — never to everyone."""
if not mine and auth.has_permission(caller, READ_ALL):
return None
return _who(caller) or "\x00nobody"
def _today(repo) -> date:
return business_time.today(repo.settings)
def _tz(repo) -> str:
return business_time.timezone_name(repo.settings)
def _stamp() -> datetime:
"""Audit stamps are real time (UTC, as the columns hold it); business dates come from _today."""
return datetime.now(timezone.utc).replace(tzinfo=None)
def _bnow() -> datetime:
"""The business instant (a demo may move it), as the columns hold it: UTC."""
return work.now().astimezone(timezone.utc).replace(tzinfo=None)
def _retrying(version: Optional[int], unit):
"""A caller who sent NO version asked for "this change on whatever is there": a race lost to another writer is
re-run on the latest. One who sent a version is told it changed (409) and reloads."""
return unit() if version is not None else retry_on_conflict(unit)
def _same_version(what: str, current: int, asked: Optional[int]) -> None:
if asked is not None and int(asked) != current:
raise Refused(409, f"{what} changed since you opened it (you have version {asked}, it is at {current}) — load it again")
# ── who sees what ─────────────────────────────────────────────────────────────────────────────
def _can_see(caller: dict, o: SalesOrder) -> bool:
scope = _scope(caller)
if scope is None or o.owner == scope.lower():
return True
return o.lifecycle == "confirmed" and auth.has_permission(caller, DELIVER) # a fulfiller sees what is to deliver
def _sees_prices(caller: dict) -> bool:
return auth.has_permission(caller, WRITE) or auth.has_permission(caller, READ_ALL)
def _load(repo, caller: dict, sales_order_id: str) -> tuple:
found = repo.sales_orders.get(sales_order_id)
if found is None or not _can_see(caller, found[0]):
raise Refused(404, f"no sales order {sales_order_id}")
return found
# ── asking the neighbours ─────────────────────────────────────────────────────────────────────
def _get(repo, sales_order_id: str) -> tuple:
"""An order the SYSTEM continues work on (a claimed request being completed, a return credited): found, whoever the caller was."""
found = repo.sales_orders.get(sales_order_id)
if found is None:
raise Refused(404, f"no sales order {sales_order_id}")
return found
def _refuse(m: Missing, what: str) -> Refused:
if not m.installed:
return Refused(409, f"{what}: that module is not part of this installation")
if m.status in (400, 404, 409, 422):
return Refused(422, f"{what}: {m.reason}")
return Refused(502, f"{what}: {m.reason} — try again")
def _base_quantity(caller: dict, ln, q: Decimal) -> Decimal:
"""A delivered quantity in the product's base unit, EXACT (SO-26): quantity x numerator / denominator, never rounded. Product
answers the same conversion rounded half up to the base unit's decimals; they must agree, or the quantity is refused
(gate exact_base) and nothing is recorded. The sign follows the quantity (a correction is negative)."""
try:
ans = ports.product.convert(caller, product_id=ln.product_id, quantity=abs(q), unit_code=ln.unit_code)
except Missing as m:
raise _refuse(m, f"product {ln.product_id}") from None
rounded = Decimal(str(ans["in_base_unit"]))
base_code, places = ans.get("base_unit_code") or "the base unit", max(0, -rounded.as_tuple().exponent)
if Fraction(abs(q)) * ln.unit_numerator / ln.unit_denominator != Fraction(rounded):
raise GateUnmet([no("exact_base", f"{num(abs(q))} {ln.unit_code} is not an exact quantity of {base_code} to {places} decimals",
f"deliver a quantity that converts exactly (1 {ln.unit_code} = {ln.unit_numerator}/{ln.unit_denominator} {base_code})")])
return rounded if q > 0 else -rounded
def _reserve(o: SalesOrder, wh: str = "") -> None:
"""With inventory installed: ask it to hold the order's lines at the order's revision (confirm: 1; every amendment: n+1) and keep
each line's stocked flag and what it holds from the ANSWER (a service line comes back ignored: stocked false). Called after the
order's own checks and before it is saved; keyed (order, revision, schedule version) there, so a retry of any step changes nothing.
The promised dates (SO-35) ride along where inventory was asked: the schedule version, the priority, each line's warehouse and
schedule. A short reservation is shown on the line and never blocks anything; an inventory that does not answer does."""
if not ports.inventory_installed():
return
lines, version = [], 0
for ln in o.lines:
if ln.item_category_code == "THIRD_PARTY": # the supplier sends it: nothing is held in inventory for it
continue
row = {"line_no": ln.line_no, "product_id": ln.product_id, "unit_code": ln.unit_code, "quantity": num(ln.quantity),
"unit_numerator": ln.unit_numerator, "unit_denominator": ln.unit_denominator}
checked = [r for r in o.in_force(ln.line_no) if r.confirmed_on and "not checked" not in (r.reason or "")]
if checked and wh:
version = o.schedule_version()
row |= {"warehouse_id": wh, "schedule": [{"confirmed_on": r.confirmed_on.isoformat(), "quantity_base": num(r.confirmed_base)} for r in checked]}
lines.append(row)
if not lines:
return
try:
got = ports.inventory.reserve(sales_order_id=o.sales_order_id, revision=o.revision, lines=lines, schedule_version=version, priority=o.priority)
except Missing as m:
raise _refuse(m, "inventory (reserve stock)") from None
answer = {int(x["line_no"]): x for x in got.get("lines") or []}
for ln in o.lines:
a = answer.get(ln.line_no) or {}
ln.stocked = bool(a) and not a.get("ignored")
ln.reserved_base = Decimal(str(a.get("reserved_base") or 0)) if ln.stocked else Decimal(0)
ln.short_base = Decimal(str(a.get("short_base") or 0)) if ln.stocked else Decimal(0)
def _currency(code: str) -> str:
if len(code) != 3 or not code.isalpha():
raise Refused(422, f"{code!r} is not an ISO 4217 currency code such as GBP")
return code.upper()
# v10 (product v7 PRD-41): the lines a use case is about to price are asked in ONE POST /prices/batch; _price then reads its
# answer from here (the same answer find_price gives). Per use-case attempt: set by _prefetched, reset when it ends.
_PRICED: contextvars.ContextVar = contextvars.ContextVar("so_priced", default=None)
def _price_key(product_id: str, quantity, unit_code: Optional[str], bp_id, currency: str, on: date) -> tuple:
return (product_id, Decimal(str(quantity)).normalize(), unit_code or None, bp_id, currency, on)
@contextmanager
def _prefetched(caller: dict, o: SalesOrder, asks: list, today: date):
"""Price `asks` [(product_id, quantity, unit_code)] in one call for the _price calls inside the block. A product module that is
not installed or does not answer leaves nothing here: each _price then asks (and refuses) exactly as before."""
found: dict = {}
asks = list(dict.fromkeys((p, Decimal(str(q)), u or None) for p, q, u in asks))
if asks:
try:
got = ports.product.price_lines(caller, bp_id=o.bp_id, currency=o.currency, priced_on=today,
lines=[{"line_no": i, "product_id": p, "quantity": q, "unit_code": u}
for i, (p, q, u) in enumerate(asks)])
for (p, q, u), ans in zip(asks, got):
found[_price_key(p, q, u, o.bp_id, o.currency, today)] = ans["price"]
except Missing:
found = {}
token = _PRICED.set(found)
try:
yield
finally:
_PRICED.reset(token)
def _ask_price(caller: dict, o: SalesOrder, product_id: str, quantity, unit_code: Optional[str], today: date) -> dict:
"""find_price's answer for one line: from the batch when _prefetched asked it, else from GET /products/{id}/price. Raises Missing
with the status and reason the one-line route would have answered."""
hit = (_PRICED.get() or {}).get(_price_key(product_id, quantity, unit_code, o.bp_id, o.currency, today))
if hit is None:
return ports.product.find_price(caller, product_id=product_id, quantity=quantity, unit_code=unit_code, bp_id=o.bp_id,
currency=o.currency, priced_on=today)
if "status" in hit and "unit_price" not in hit:
raise Missing(f"product {product_id}", str(hit.get("reason") or ""), int(hit.get("status") or 0))
return hit
def _price(caller: dict, o: SalesOrder, product_id: str, quantity: Decimal, unit_code: Optional[str], today: date,
base=None) -> LineSpec:
"""One line, priced by product's find_price with priced_on = today; priced_from names what the answer used.
`base`: an existing line whose copies (description, tax category, policy, billing) are kept (an amendment).
Inside _prefetched the answer comes from the batch (the same answer, product v7 PRD-41)."""
try:
raw = _ask_price(caller, o, product_id, quantity, unit_code, today)
except Missing as m:
raise _refuse(m, f"product {product_id}") from None
a = price_answer(raw)
keep = base is not None
return LineSpec(product_id, base.description if keep else (a.description or product_id), a.unit_code, quantity,
a.numerator, a.denominator, a.unit_price, a.unit_price, Decimal(0), today, a.priced_from,
base.tax_category if keep else a.tax_category, base.invoicing_policy if keep else a.invoicing_policy,
base.billing_kind if keep else a.billing_kind, base.billing_plan if keep else a.billing_plan,
pricing_quantity=a.pricing_quantity, pricing_unit_code=a.pricing_unit_code, pricing_numerator=a.pricing_numerator,
pricing_denominator=a.pricing_denominator, category_code=a.category_code,
product_type_code=base.product_type_code or a.product_type_code if keep else a.product_type_code,
usage=base.usage if keep else "normal", item_category_code=base.item_category_code if keep else "",
gross_weight=a.gross_weight, weight_unit_code=a.weight_unit_code, measure_valid_from=a.measure_valid_from, procedural=True)
def _categorise(repo, o: SalesOrder, spec: LineSpec, today: date, usage: str = "normal") -> None:
"""The item category a line is determined with (SO-29): from the order's type, the product's type and the line's usage, by the rule in
force today; the line keeps it. A product that answers no type (a product module that predates PRD-31) leaves it empty."""
spec.usage = usage
if spec.product_type_code:
spec.item_category_code = pr.item_category(repo.config.rules(), o.order_type_code, spec.product_type_code, usage, today)
def _reprice(repo, o: SalesOrder, today: date, only=None) -> set:
"""Run the pricing procedure over the lines (all of them for a draft) and spread the header conditions; the lines whose net changed."""
t = repo.config.order_type(o.order_type_code, today)
cats = {c.item_category_code: c for c in repo.config.categories()}
stat = frozenset(ln.line_no for ln in o.lines if (cats.get(ln.item_category_code) or pr.ItemCategory("")).pricing == "statistical")
return o.reprice(only=[ln.line_no for ln in o.lines] if only is None else only, types=repo.config.condition_types(),
steps=repo.config.procedure(t.pricing_procedure_code if t else "STANDARD"), access=repo.config.access(),
records=repo.config.records(), on=today, revision=o.revision, statistical=stat)
_NEVER_BILLED = ("FREE_OF_CHARGE", "CONS_FILL", "CONS_PICKUP", "CONS_RETURN", "RETURNS", "CREDIT_REQ", "DEBIT_REQ")
def _static_skip(o: SalesOrder) -> frozenset:
"""What the read model skips when it derives the invoice status: the seeded types and categories that never bill (the use cases read the
configured ones with `_nonbilling`)."""
if o.order_type_code in _NEVER_BILLED:
return frozenset(ln.line_no for ln in o.lines)
return frozenset(ln.line_no for ln in o.lines if ln.item_category_code in ("FREE", "CONS_FILL"))
def _nonbilling(repo, o: SalesOrder, today: date) -> frozenset:
"""The lines that are never invoiced: every line of an order type that does not bill (free of charge, a consignment fill, a return or a
memo request, whose money goes back through a credit or a debit), and a line whose item category is not billing-relevant."""
t = repo.config.order_type(o.order_type_code, today)
if t is not None and t.billing != "invoice":
return frozenset(ln.line_no for ln in o.lines)
cats = {c.item_category_code: c for c in repo.config.categories()}
return frozenset(ln.line_no for ln in o.lines if ln.item_category_code in cats and not cats[ln.item_category_code].billing_relevant)
def _multiples(caller: dict, asks: list, on: date, currency: str) -> Optional[dict]:
"""product's delivery-multiple answer for each (line, quantity, unit) asked (SO-28): {line_no: answer | "unanswered" | None (no rule to
read)}; None altogether when product is not part of this installation."""
out = {}
if not asks:
return out
try: # v10: every line in ONE POST /prices/batch (product v7 PRD-41)
got = ports.product.price_lines(caller, bp_id=None, currency=currency, priced_on=on, lines=[
{"line_no": ln.line_no, "product_id": ln.product_id, "quantity": qty, "unit_code": unit, "price": False,
"rounding_role": "delivery"} for ln, qty, unit in asks])
except Missing as m:
if not m.installed:
return None
return {ln.line_no: "unanswered" for ln, _, _ in asks}
for (ln, _, _), ans in zip(asks, got):
r = ans.get("rounding")
if isinstance(r, dict) and "status" in r and "met" not in r:
out[ln.line_no] = None if int(r.get("status") or 0) == 404 else "unanswered"
else:
out[ln.line_no] = r
return out
def _entered_tax(li: LineIn, spec: LineSpec) -> None:
"""Where the tax module is not installed, tax is entered per line; where it is, a typed tax is refused."""
if li.tax_code or li.tax_rate_pct is not None:
if ports.tax_installed():
raise Refused(422, "tax is computed by the tax module: leave tax_code and tax_rate_pct out")
spec.tax_code = li.tax_code or "ENTERED"
spec.tax_amount = money(money(spec.quantity * spec.unit_price) * (li.tax_rate_pct or Decimal(0)) / 100)
def _apply_tax(caller: dict, o: SalesOrder, line_nos=None) -> str:
"""Ask the tax module for the order's tax (tax point = the order date) and store its answer as the order's own copy.
Returns a note when it could not (the readiness gate lines_taxed says why); never guesses a rate."""
pending = [ln for ln in o.lines if (line_nos is None or ln.line_no in line_nos) and not (ln.tax_code and not ports.tax_installed())]
if not pending:
return ""
if not ports.tax_installed():
return "the tax module is not part of this installation: enter tax per line"
try:
ans = ports.tax.compute(caller, lines=[{"ref": ln.line_no, "product_id": ln.product_id, "tax_category": ln.tax_category,
"net_amount": num(ln.net_amount, 2)} for ln in pending],
bp_id=o.bp_id, tax_point=o.ordered_on, currency=o.currency, document_ref=f"so:{o.sales_order_id}@r{o.revision}",
ship_to_country=o.ship_to_country)
except Missing as m:
return f"tax could not be computed: {m.reason}"
for x in ans.get("lines", []):
o.set_tax(int(x["ref"]), x["tax_code"], Decimal(str(x["tax_amount"])))
return ""
def _issued(o: SalesOrder, strict: bool = False) -> dict:
"""Quantities on ISSUED invoices, read back from invoicing (never stored here)."""
if o.lifecycle in ("draft", "cancelled") or not ports.invoicing_installed():
return {}
try:
return {int(k): Decimal(str(v)) for k, v in ports.invoicing.billed(o.sales_order_id).items()}
except Missing as m:
if strict:
raise Refused(502, f"invoicing did not answer ({m.reason}), so what is already invoiced cannot be confirmed — try again") from None
return {}
def _issued_deposits(o: SalesOrder) -> tuple:
"""The deposit requests whose invoice invoicing has ISSUED (a deposit has no order line, so `billed` never shows it)."""
deps = [r for r in o.invoice_requests if r.kind == "deposit" and r.status == "requested" and r.invoice_id]
sources = [f"so:{o.sales_order_id}:{r.request_no}" for r in deps]
if not sources or not ports.invoicing_installed():
return ()
try:
by_invoice = ports.invoicing.issued_sources([r.invoice_id for r in deps]) # ar keys /invoice-totals by ITS invoice_id, never by our source id
got = {f"so:{o.sales_order_id}:{r.request_no}": by_invoice.get(r.invoice_id) for r in deps}
except Missing as m:
raise Refused(502, f"invoicing did not answer ({m.reason}), so whether a deposit invoice is issued cannot be confirmed — try again") from None
return tuple(s for s in sources if got.get(s))
def _facts(repo, caller: dict, o: SalesOrder, today: date, strict: bool = False) -> gate_rules.Facts:
f = gate_rules.Facts(invoicing_installed=ports.invoicing_installed(), issued=_issued(o, strict))
if o.lifecycle == "draft":
seen: dict = {} # each partner is asked once per confirmation (sold-to is often payer and bill-to)
try:
f.customer_active = bool(_asked(seen, "customer", o.bp_id, lambda: ports.partners.customer(caller, o.bp_id)).get("can_transact", True))
f.customer_note = "" if f.customer_active else "the customer cannot transact on the partner register"
except Missing as m:
f.customer_active = False if m.status == 404 else None
f.customer_note = (f"there is no business partner {o.bp_id}" if m.status == 404 else f"the partner register did not answer ({m.reason})")
otype = repo.config.order_type(o.order_type_code, today)
f.delivery_mode = repo.settings.value_on("delivery_multiple_mode", today) or "warn"
f.multiples = _multiples(caller, [(ln, ln.quantity, ln.unit_code) for ln in o.lines], o.ordered_on or today, o.currency)
if ports.payments_installed() and (otype is None or otype.credit_check):
try:
f.credit = ports.payments.credit_status(caller, o.payer) # the PAYER's hold, never the sold-to's
except Missing as m:
f.credit = {"hold": True, "reason": f"payments did not answer ({m.reason})", "mode": "block"}
pol = repo.policies.in_force(o.bp_id, today)
f.po_required = bool(pol and pol["po_required"])
f.extra = [_partners_gate(caller, o, seen), *_incomplete(repo, o, "confirm", today), *_consignment_gates(caller, o)]
return f
def _partner_country(caller: dict, bp_id: int) -> str:
"""v7 (the business-partner seam): the ship-to's country, copied onto the order as ship_to_country - what delivery terms and tax's
place of supply read. A register that does not answer leaves it empty (the partners_valid gate names that at confirmation)."""
try:
return str(ports.partners.partner(caller, int(bp_id)).get("country") or "")
except Missing:
return ""
def _asked(seen: Optional[dict], what: str, bp_id, ask):
"""The neighbour's answer for one partner, asked once per `seen` (a refusal is remembered and raised again)."""
if seen is None:
return ask()
key = ("partner" if isinstance(ports.partners, RealPartners) else what, int(bp_id)) # the real port answers both from GET /partners/{id}
if key not in seen:
try:
seen[key] = ("ok", ask())
except Missing as m:
seen[key] = ("missing", m)
kind, got = seen[key]
if kind == "missing":
raise got
return got
def _partners_gate(caller: dict, o: SalesOrder, seen: Optional[dict] = None):
"""partners_valid (SO-40): sold-to, ship-to, bill-to and payer are each an ACTIVE business partner with the role they play (customer; a
ship-to may be a ship_to or a customer). A partner the register cannot answer for is unmet, never assumed."""
bad = []
for fn, bp, roles in (("sold-to", o.bp_id, ("customer",)), ("ship-to", o.ship_to_bp_id or o.bp_id, ("ship_to", "customer")),
("bill-to", o.bill_to, ("customer",)), ("payer", o.payer, ("customer",))):
try:
p = _asked(seen, "partner", bp, lambda bp=bp: ports.partners.partner(caller, bp))
except Missing as m:
bad.append(f"{fn} {bp}: " + ("there is no such business partner" if m.status == 404 else f"the partner register did not answer ({m.reason})"))
continue
if p.get("status") != "active":
bad.append(f"{fn} {bp} is {p.get('status') or 'not active'}")
elif not set(roles) & set(p.get("roles") or []):
bad.append(f"{fn} {bp} is not a {' or '.join(roles)} on the partner register")
return no("partners_valid", "; ".join(bad), "Business partner > the partner's status and roles, or Order > its partners") if bad else gate_rules.ok("partners_valid")
def _payer_gates(repo, caller: dict, o: SalesOrder, today: date) -> list:
"""What confirmation asked of the payer, asked again when a confirmed order's payer changes (SO-40): every partner active with its role (partners_valid)
and the NEW payer's credit hold (never the old payer's override)."""
gates = [_partners_gate(caller, o)]
otype = repo.config.order_type(o.order_type_code, today)
if ports.payments_installed() and (otype is None or otype.credit_check):
try:
c = ports.payments.credit_status(caller, o.payer)
except Missing as m:
c = {"hold": True, "reason": f"payments did not answer ({m.reason})", "mode": "block"}
if c.get("hold") and (c.get("mode") or "block") != "warn":
gates.append(no("credit", f"payer {o.payer} is on hold: {c.get('reason', 'credit hold')}", "ask an order admin to override, or choose another payer"))
return gates
def _consignment_gates(caller: dict, o: SalesOrder) -> list:
"""consignment_on_hand (SO-45): a CONS_ISSUE or CONS_PICKUP takes no more than the customer holds on consignment, read from inventory (the stock is its,
never kept here). An inventory that does not answer is unmet, never assumed."""
if o.order_type_code not in ("CONS_ISSUE", "CONS_PICKUP") or not ports.inventory_installed():
return []
want: dict = {}
for ln in o.lines:
want[ln.product_id] = want.get(ln.product_id, Decimal(0)) + Decimal(ln.quantity) * ln.unit_numerator / ln.unit_denominator
bad = []
for pid, qty in want.items():
try:
held = ports.inventory.consignment_stock(bp_id=o.bp_id, product_id=pid)
except Missing as m:
bad.append(f"{pid}: inventory did not answer ({m.reason})")
continue
row = next((x for x in held.get("products") or [] if x.get("product_id") == pid), None)
on_hand = Decimal(str(row["on_hand_base"])) if row else Decimal(0)
if qty > on_hand:
bad.append(f"{o.bp_id} holds {num(on_hand)} {(row or {}).get('base_unit_code') or 'of the base unit'} of {pid} on consignment: {num(qty)} is more")
return [no("consignment_on_hand", "; ".join(bad), "take no more than the customer holds, or fill the consignment first")] if bad else [gate_rules.ok("consignment_on_hand")]
def _incomplete(repo, o: SalesOrder, stage: str, today: date) -> list:
"""The incomplete:<field> gates of a stage (confirm | delivery | billing): the dated rules of the order's type (SO-38)."""
pol = repo.policies.in_force(o.bp_id, today)
return rl.incomplete_gates(o, repo.config.incompletion_rules(), stage, on=today, company_country=repo.settings.value_on("company_country", today) or "",
policy_requires_po=bool(pol and pol["po_required"]))
def _foc_value(o: SalesOrder):
return sum((ln.statistical_value for ln in o.lines), Decimal(0))
def _delivery_gates(repo, o: SalesOrder, today: date) -> list:
gates = [rl.block_gate(o, "delivery"), *_incomplete(repo, o, "delivery", today)]
if o.order_type_code == "FREE_OF_CHARGE" and o.active_block("delivery") is not None and o.active_block("delivery").reason_code == "APPROVAL":
limit = Decimal(str(repo.settings.value_on("free_of_charge_limit", today) or "250.00"))
gates.append(no("free_of_charge_approval", f"{num(_foc_value(o), 2)} is above the limit of {num(limit, 2)} of statistical value: a controller approves it",
"a holder of so:memo:approve other than the requester (another email AND another sign-in)"))
return gates
def _billing_gates(repo, o: SalesOrder, today: date) -> list:
t = repo.config.order_type(o.order_type_code, today)
never = t is not None and t.billing != "invoice"
return [rl.block_gate(o, "billing"), *_incomplete(repo, o, "billing", today),
no("billing_relevant", f"a {o.order_type_code} order is never invoiced" + (": its approval raises the credit or debit" if o.order_type_code in MEMO_TYPES else
": its credit follows its inspection" if o.order_type_code in RETURN_TYPES else ""),
"nothing to request") if never else gate_rules.ok("billing_relevant")]
def _needs_approval(repo, today: date) -> bool:
v = repo.settings.value_on("amendment_needs_approval", today)
return True if v is None else bool(v)
def _readiness(repo, caller: dict, o: SalesOrder, today: date, f: Optional[gate_rules.Facts] = None) -> dict:
f = f or _facts(repo, caller, o, today)
if o.lifecycle == "draft":
return {"confirm": [g.as_dict() for g in gate_rules.confirm_gates(o, f)]}
if o.lifecycle == "confirmed":
skip = _nonbilling(repo, o, today)
return {"amend": [g.as_dict() for g in gate_rules.amend_gates(o, f, needs_approval=_needs_approval(repo, today))],
"cancel": [g.as_dict() for g in gate_rules.cancel_gates(o, f)],
"deposit": [g.as_dict() for g in gate_rules.deposit_gates(o)],
"invoice": [g.as_dict() for g in gate_rules.invoice_gates(o, f, skip)],
"delivery": [g.as_dict() for g in _delivery_gates(repo, o, today)],
"billing": [g.as_dict() for g in _billing_gates(repo, o, today)]}
return {}
# ── the read model ────────────────────────────────────────────────────────────────────────────
_PRICED_ROOT = ("subtotal", "tax_total", "total", "deposit_pct", "credit_override_by", "credit_override_reason")
_PRICED_LINE = ("statistical_value", "list_price", "unit_price", "discount_pct", "priced_on", "priced_from", "tax_code", "net_amount", "tax_amount", "conditions",
"quantity_priced")
def _line_view(o: SalesOrder, ln, issued: dict) -> dict:
inv = o.invoiced_qty(ln.line_no, issued)
return {**{k: getattr(ln, k) for k in ln.__dataclass_fields__}, "delivered": o.delivered(ln.line_no), "invoiced": inv,
"to_deliver": max(Decimal(0), ln.quantity - o.delivered(ln.line_no)), "to_invoice": o.due_qty(ln.line_no, issued),
"quantity_priced": ln.quantity_priced,
"schedule": [{k: getattr(r, k) for k in ("schedule_version", "seq", "requested_on", "requested_quantity", "confirmed_on", "material_available_on",
"confirmed_quantity", "reason")} for r in o.in_force(ln.line_no)],
"promise": promising.line_summary(o, ln),
"conditions": [{k: getattr(c, k) for k in ("step", "condition_type_code", "rate", "amount", "basis", "source", "statistical",
"header_share_of", "revision")} for c in o.current_conditions(ln.line_no)
if c.source != "no longer applies"]}
_MONEYISH = frozenset({"subtotal", "tax_total", "total", "net_amount", "tax_amount", "list_price", "unit_price", "amount", "deducted", "basis"})
def _plain(v, key: str = ""):
"""Numbers as a person reads them (12, 24.00), the same whether they are in memory or just loaded: a replay of a
command answers byte for byte what the first call did."""
if isinstance(v, Decimal):
return num(v, 2 if key in _MONEYISH else 0)
if isinstance(v, dict):
return {k: _plain(x, k) for k, x in v.items()}
if isinstance(v, list):
return [_plain(x, key) for x in v]
return v
def _view(o: SalesOrder, version: int, caller: dict, issued: Optional[dict] = None, prices: Optional[bool] = None) -> dict:
"""The order as a screen or an agent needs it: derived statuses included, nothing stored for them."""
issued = issued if issued is not None else {}
prices = _sees_prices(caller) if prices is None else prices
lines = [_line_view(o, ln, issued) for ln in o.lines]
out = {k: getattr(o, k) for k in ("sales_order_id", "order_no", "bp_id", "owner", "currency", "price_list_id", "lifecycle",
"revision", "customer_po_ref", "ordered_on", "confirmed_at", "confirmed_on", "delivery_terms",
"payment_terms_id", "source_kind", "source_id", "cancel_reason", "order_type_code", "ship_to_bp_id",
"ship_to_country", "price_group", "ref_sales_order_id", "ref_line_no", "ref_invoice_id", "freight_unknown",
"bill_to_bp_id", "payer_bp_id", "requested_delivery_on", "priority", "partial_delivery", "max_partials",
"under_tolerance_pct", "over_tolerance_pct", "combine_allowed", "min_shelf_life_days", "return_reason",
"deposit_deduction_mode", "approved_by_email", "approved_by_sub", "approved_at", *_PRICED_ROOT)}
out["ship_to_address"] = _address_of(o) if o.ship_to_address_lines else None
out.update(version=version, fulfilment_status=o.fulfilment_status(), invoice_status=o.invoice_status(issued, _static_skip(o)), lines=lines,
warnings=list(json.loads(o.warnings_json or "{}").values()),
header_conditions=[{k: getattr(h, k) for k in ("condition_type_code", "step", "rate", "amount", "source")}
for h in o.current_header_conditions() if h.source != "no longer applies"],
deliveries=[dict(d.__dict__) for d in o.deliveries], sends=[dict(s.__dict__) for s in o.sends],
revisions=[{k: v for k, v in r.__dict__.items()} for r in o.revisions],
suggestions=[{**{k: v for k, v in s.__dict__.items() if k not in ("output_json", "decisions_json")},
"output": s.output, "decisions": s.decisions} for s in o.suggestions],
invoice_requests=[{k: v for k, v in r.__dict__.items()} for r in o.invoice_requests],
blocks=[dict(b.__dict__) for b in o.blocks], schedule_version=o.schedule_version(),
milestones=[{k: v for k, v in m.__dict__.items() if prices or k != "amount"} for m in o.milestones],
inspections=[dict(i.__dict__) for i in o.inspections])
for ln in out["lines"]:
if o.order_type_code in RETURN_TYPES:
acc, rej = o.inspected(ln["line_no"])
ln.update(received=o.delivered(ln["line_no"]), accepted=acc, rejected=rej, credited=o.credited_qty(ln["line_no"]))
out = _plain(out)
if not prices:
out.pop("header_conditions", None)
for k in _PRICED_ROOT:
out.pop(k, None)
for ln in out["lines"]:
for k in _PRICED_LINE:
ln.pop(k, None)
out.pop("invoice_requests", None)
return out
def _fresh(caller: dict, sales_order_id: str, **kw) -> dict:
"""The order as it now stands, after a call made following the commit changed it."""
with repository() as repo:
o, v = _get(repo, sales_order_id)
return _view(o, v, caller, _issued(o), **kw)
def _saved(repo, caller: dict, o: SalesOrder, current: int, **kw) -> dict:
new = repo.sales_orders.save(o, current)
return _view(o, new, caller, _issued(o), **kw)
# ═══ reading ═══════════════════════════════════════════════════════════════════════════════════
@SO.use_case(kind="query", route="GET /demand", permission=DEMAND_READ, aliases={"date_from": "from", "date_to": "to"})
def get_demand(caller: dict, product_id: Ref, warehouse_id: Code, date_from: Day, date_to: Day, limit: Limit = 500, offset: Offset = 0) -> dict:
"""SO-50: the open quantity of a product across CONFIRMED orders' goods lines delivered from a warehouse, dated by the latest schedule line's
confirmed date, else the requested date; open = ordered - delivered (a line closed within its tolerance is open for nothing). No price,
customer or amount is answered (manufacturing's MRP-lite never reads so_ tables). Drafts, quotes, cancelled orders and THIRD_PARTY lines are not
demand. Ordered by needed_on, sales_order_id, line_no; the delivering warehouse is the setting default_warehouse_id."""
try:
lo, hi = day(date_from, "from"), day(date_to, "to")
except ValueError as e:
raise Refused(422, str(e)) from None
if lo is None or hi is None or lo > hi:
raise Refused(422, "from and to are dates, from not after to")
rows = []
with repository() as repo:
today = _today(repo)
if (repo.settings.value_on("default_warehouse_id", today) or "").strip() == warehouse_id:
for sid in repo.sales_orders.confirmed_with_product(product_id):
found = repo.sales_orders.get(sid) # no owner scope: the machine permission is the gate, and no customer or price leaves
if found is None:
continue
o = found[0]
for ln in o.lines:
if ln.product_id != product_id or ln.item_category_code == "THIRD_PARTY":
continue
open_q = Decimal(0) if o.closed(ln) else max(Decimal(0), ln.quantity - o.delivered(ln.line_no))
if open_q <= 0:
continue
dated = [r.confirmed_on for r in o.in_force(ln.line_no) if r.confirmed_on]
needed = max(dated) if dated else (o.requested_delivery_on or o.ordered_on)
if lo <= needed <= hi:
base = Fraction(open_q) * Fraction(ln.unit_numerator or 1, ln.unit_denominator or 1)
rows.append({"needed_on": needed, "sales_order_id": sid, "line_no": ln.line_no, "open_quantity": num(open_q),
"unit_code": ln.unit_code, "open_base": num(Decimal(base.numerator) / Decimal(base.denominator)),
"confirmed": bool(dated)})
rows.sort(key=lambda r: (r["needed_on"], r["sales_order_id"], r["line_no"]))
return {"product_id": product_id, "warehouse_id": warehouse_id, "from": lo, "to": hi, "demand": rows[offset:offset + limit], "more": offset + limit < len(rows)}
@SO.use_case(kind="query", route="GET /sales-orders", permission=READ)
def list_sales_orders(caller: dict, mine: bool = False, state: Optional[Code] = None, bp_id: Optional[BigId] = None,
ordered_from: Optional[IsoDate] = None, ordered_to: Optional[IsoDate] = None, limit: Limit = 50,
offset: Offset = 0) -> dict:
"""One list of the orders the caller may see, with what is still to deliver and to invoice (both derived): each row
shows the number, the customer, the total, the fulfilment status and the invoice status. Filter by state (draft,
confirmed, done, cancelled), customer and date; `mine` narrows to the caller's own. Paged."""
try:
frm, to = day(ordered_from, "ordered_from"), day(ordered_to, "ordered_to")
except ValueError as e:
raise Refused(422, str(e)) from None
scope = _scope(caller, mine)
with repository() as repo:
page, total = repo.sales_orders.page(owner=scope.lower() if scope else None,
also_confirmed=bool(scope) and not mine and auth.has_permission(caller, DELIVER),
lifecycle=state, bp_id=bp_id, ordered_from=frm, ordered_to=to, limit=limit, offset=offset)
prices = _sees_prices(caller)
issued = {}
live = [o.sales_order_id for o in page if o.lifecycle in ("confirmed", "done")]
if live and ports.invoicing_installed():
try:
issued = ports.invoicing.billed_page(live)
except Missing:
issued = {}
rows = []
for o in page:
row = {"sales_order_id": o.sales_order_id, "order_no": o.order_no, "bp_id": o.bp_id, "owner": o.owner,
"currency": o.currency, "lifecycle": o.lifecycle, "ordered_on": o.ordered_on,
"fulfilment_status": o.fulfilment_status(), "invoice_status": o.invoice_status(issued.get(o.sales_order_id, {}), _static_skip(o))}
if prices:
row["total"] = o.total
rows.append(row)
return {"orders": rows, "total": total, "limit": limit, "offset": offset}
@SO.use_case(kind="query", route="GET /sales-orders/{sales_order_id}", permission=READ)
def get_sales_order(caller: dict, sales_order_id: Ref) -> dict:
"""One sales order, whole, with its version, its derived fulfilment and invoice status, and each line's delivered,
invoiced and to-invoice quantity. A fulfiller sees no prices."""
with repository() as repo:
o, version = _load(repo, caller, sales_order_id)
return _view(o, version, caller, _issued(o))
@SO.use_case(kind="query", route="GET /sales-orders/{sales_order_id}/readiness", permission=READ)
def get_order_readiness(caller: dict, sales_order_id: Ref) -> dict:
"""What stands between this order and its next steps, as a list of {gate, met, reason, fix} per step (a draft: confirm;
a confirmed order: amend, cancel, deposit, invoice). A refusal names the same gate, reason and fix."""
with repository() as repo:
o, _ = _load(repo, caller, sales_order_id)
return {"sales_order_id": o.sales_order_id, "lifecycle": o.lifecycle, "readiness": _readiness(repo, caller, o, _today(repo))}
# ═══ the draft ═════════════════════════════════════════════════════════════════════════════════
def _need_type(repo, code: str, today: date) -> pr.OrderType:
t = repo.config.order_type(code, today)
if t is None or t.status != "active":
raise Refused(422, f"there is no active order type {code} today: the types are {', '.join(sorted({x.order_type_code for x in repo.config.order_types()}))}")
return t
def _need_reference(t: pr.OrderType, order_id, line_no, invoice_id) -> None:
"""A type that returns or corrects something names it (gate reference_required): refused before anything is made."""
ask = {"order_line": (bool(order_id and line_no), "a return names the order line it returns", "give ref_sales_order_id and ref_line_no"),
"invoice": (bool(invoice_id), "a credit or debit request names the invoice it is about", "give ref_invoice_id"),
"consignment": (bool(order_id), "a consignment movement names the consignment order it belongs to", "give ref_sales_order_id")}.get(t.reference)
if ask and not ask[0]:
raise GateUnmet([no("reference_required", ask[1], ask[2])])
def _policy_defaults(repo, bp_id: int, today: date) -> tuple:
pol = repo.policies.in_force(bp_id, today)
default = repo.settings.value_on("default_deposit_pct", today) or 0
pct = pol["deposit_pct"] if pol and pol["deposit_pct"] is not None else Decimal(default)
return Decimal(pct), (pol["delivery_terms"] if pol and pol["delivery_terms"] else "")
def _key_return(oid: str, line_no: int) -> str:
return f"ret:{oid}:{line_no}"
def _key_invoice(invoice_id: str, line_no: int) -> str:
return f"inv:{invoice_id}:{line_no}"
def _return_reason(repo, code: str):
return next((r for r in repo.config.return_reasons() if r.return_reason_code.upper() == (code or "").strip().upper()), None)
def _return_spec(repo, caller: dict, otype, li: LineIn, ref: dict, today: date, taking: dict) -> tuple:
"""A return's line copies the ORIGINAL line's price, unit and tax category (never re-priced) and names the original order line, its delivery
and its invoice (gate reference_required); no more than delivered less already returned on that line (gate within_delivered). -> (spec, gates)."""
gates = []
oid, lno, inv, ilno, dref = ref["order"], ref["line"], ref["invoice"], ref["invoice_line"], ref["delivery"]
if not (oid and lno):
return None, [no("reference_required", "a return names the order line it returns", "give ref_sales_order_id and ref_line_no")]
orig, _ = _load(repo, caller, oid)
ol = orig.line(lno)
if ol is None:
return None, [no("reference_required", f"order {oid} has no line {lno}", "name a line of the original order")]
if ol.product_id != li.product_id:
raise Refused(422, f"line {lno} of {oid} is {ol.product_id}, not {li.product_id}: a return is of what was delivered")
if li.unit_code and li.unit_code != ol.unit_code:
raise Refused(422, f"a return is counted in the original line's unit ({ol.unit_code}), not {li.unit_code}")
refs = sorted({d.delivery_ref for d in orig.deliveries if d.line_no == lno and d.kind != "correction" and d.quantity > 0})
dref = dref or (refs[0] if len(refs) == 1 else "")
if not dref or dref not in refs:
gates.append(no("reference_required", f"a return names the delivery it comes from ({', '.join(refs) or 'nothing was delivered on that line'})",
"give ref_delivery_ref"))
if not (inv and ilno):
gates.append(no("reference_required", "a return names the invoice line the goods were billed on", "give ref_invoice_id and ref_invoice_line_no"))
key = _key_return(oid, lno)
delivered, returned = orig.delivered(lno), repo.claims.returned(key) + taking.get(key, Decimal(0))
allowed = max(Decimal(0), delivered - returned)
if li.quantity > allowed:
gates.append(no("within_delivered", f"{num(delivered)} delivered, {num(returned)} returned: at most {num(allowed)}", f"return {num(allowed)} or fewer"))
if gates:
return None, gates
taking[key] = taking.get(key, Decimal(0)) + li.quantity
spec = LineSpec(ol.product_id, ol.description, ol.unit_code, li.quantity, ol.unit_numerator, ol.unit_denominator, ol.list_price, ol.unit_price,
ol.discount_pct, ol.priced_on, f"return of {orig.order_no or oid} line {lno}", ol.tax_category, "ordered",
pricing_quantity=ol.pricing_quantity, pricing_unit_code=ol.pricing_unit_code, pricing_numerator=ol.pricing_numerator,
pricing_denominator=ol.pricing_denominator, category_code=ol.category_code, product_type_code=ol.product_type_code,
ref_sales_order_id=oid, ref_line_no=lno, ref_delivery_ref=dref, ref_invoice_id=inv, ref_invoice_line_no=ilno)
return spec, []
def _memo_spec(caller: dict, o: SalesOrder, li: LineIn, ref: dict, today: date) -> LineSpec:
"""A credit or debit memo request STATES its amount (the unit price): the product only names what it is for. A credit says which invoice line."""
if li.unit_price is None:
raise Refused(422, "a credit or debit request states its amount: give unit_price on each line")
if o.order_type_code == "CREDIT_REQ" and not ref["invoice_line"]:
raise GateUnmet([no("reference_required", "a credit request names the invoice line it corrects", "give ref_invoice_line_no")])
base = _price(caller, o, li.product_id, li.quantity, li.unit_code, today)
return dataclasses.replace(base, unit_price=li.unit_price, list_price=li.unit_price, discount_pct=Decimal(0), priced_from=f"typed on the request by {_who(caller)}",
description=li.description or base.description, pricing_quantity=1, pricing_unit_code="", pricing_numerator=1,
pricing_denominator=1, procedural=False, ref_invoice_id=ref["invoice"], ref_invoice_line_no=ref["invoice_line"])
@SO.use_case(kind="command", route="POST /sales-orders", permission=WRITE)
def draft_by_hand(caller: dict, bp_id: BigId, currency: Currency, lines: list[LineIn], customer_po_ref: Optional[Ref] = None,
ordered_on: Optional[IsoDate] = None, payment_terms_id: Optional[Code] = None,
sales_order_id: Optional[Ref] = None, order_type_code: Optional[Code] = None, ship_to_bp_id: Optional[BigId] = None,
ship_to_country: Optional[text(2, min_length=2)] = None, ref_sales_order_id: Optional[Ref] = None,
ref_line_no: Optional[LineNo] = None, ref_invoice_id: Optional[Ref] = None, bill_to_bp_id: Optional[BigId] = None,
payer_bp_id: Optional[BigId] = None, requested_delivery_on: Optional[IsoDate] = None, return_reason: Optional[Ref] = None,
priority: Optional[Literal["normal", "rush"]] = None, ref_delivery_ref: Optional[Ref] = None,
ref_invoice_line_no: Optional[LineNo] = None, ship_to_address: Optional[AddressIn] = None) -> dict:
"""Raise an order by hand when there was no quote. The order is dated today in business_timezone (it cannot be dated
earlier); give only products, units and quantities: every line is priced by the product module's find_price (priced_on
today), stores the price, the unit, its conversion and priced_from, and is taxed by the tax module. A unit the product
does not sell in is refused. The order has a TYPE (default STANDARD; RUSH, CASH, RETURNS, CREDIT_REQ, DEBIT_REQ, FREE_OF_CHARGE,
CONS_*): it decides each line's item category and the series the number comes from, and a type that needs a reference (a return
names the order line it returns, a memo request the invoice) is refused without it (gate reference_required). Each line is then run
through the pricing procedure: product's price (BASE), then the discounts and surcharges the customer's agreements give, the header
conditions and freight to the ship-to's country. The ship-to, bill-to and payer default from the customer's partner rows in force on the order's
date (a ship-to you name must be one of the customer's: "9999 is not a ship-to of 7311") and are copied onto the order. Pass the same
sales_order_id again to get the same order back.
A RETURNS order (raise_return) names, per line or on the header, the original order line, its delivery and its invoice line and a
return_reason; it copies the original's price, unit and tax category and is refused beyond what was delivered less what was already returned
(gate within_delivered). A CREDIT_REQ or DEBIT_REQ states each line's unit_price and names the invoice (a credit, the invoice line). A line may
carry milestones (billed by value, they must total 100% to confirm) or instalments (a schedule at invoicing). Samples and replacements are a
FREE_OF_CHARGE order: every line is priced for statistics only.
v8: `ship_to_address` is the postal address object {lines, city, postcode, region, country} (shared.yaml conventions.address), typed on the
order; its country is the ship-to country. A drop-ship line sends it to purchasing and cannot be ordered without it."""
sid = sales_order_id or uuid.uuid4().hex
try:
on = day(ordered_on, "ordered_on")
except ValueError as e:
raise Refused(422, str(e)) from None
def unit():
with repository() as repo, _decide():
if repo.sales_orders.exists(sid):
o, v = _load(repo, caller, sid)
return _view(o, v, caller)
today = _today(repo)
pct, terms = _policy_defaults(repo, int(bp_id), today)
otype = _need_type(repo, order_type_code or "STANDARD", today)
refs = [{"order": li.ref_sales_order_id or ref_sales_order_id or "", "line": li.ref_line_no or ref_line_no or 0,
"invoice": li.ref_invoice_id or ref_invoice_id or "", "invoice_line": li.ref_invoice_line_no or ref_invoice_line_no or 0,
"delivery": li.ref_delivery_ref or ref_delivery_ref or ""} for li in lines]
for r in refs:
_need_reference(otype, r["order"], r["line"], r["invoice"])
code = otype.order_type_code
if return_reason and code in RETURN_TYPES and not _return_reason(repo, return_reason):
raise GateUnmet([no("return_reason", f"{return_reason!r} is not a return reason: {', '.join(r.return_reason_code for r in repo.config.return_reasons())}",
"give one of them (an empty reason is the incompletion rule's to refuse at authorisation)")])
pol = repo.policies.in_force(int(bp_id), today)
partners, rows = {}, repo.config.partners()
for fn, asked in (("ship_to", ship_to_bp_id), ("bill_to", bill_to_bp_id), ("payer", payer_bp_id)):
partners[fn], why = rl.resolve_partner(rows, int(bp_id), fn, asked, on or today)
if why:
raise Refused(422, why)
try:
want_on = day(requested_delivery_on, "requested_delivery_on")
except ValueError as e:
raise Refused(422, str(e)) from None
country = (ship_to_address.country if ship_to_address else (ship_to_country or "").upper()) or _partner_country(caller, partners["ship_to"] or int(bp_id))
first = refs[0] if refs else {"order": "", "line": 0, "invoice": ""}
o = SalesOrder.draft(sid, owner=_who(caller), bp_id=int(bp_id), currency=_currency(currency), ordered_on=on, today=today,
customer_po_ref=customer_po_ref or "", delivery_terms=terms, payment_terms_id=payment_terms_id or "",
deposit_pct=pct, order_type_code=code, ship_to_bp_id=partners["ship_to"],
bill_to_bp_id=partners["bill_to"], payer_bp_id=partners["payer"], requested_delivery_on=want_on,
return_reason=return_reason or "",
ship_to_country=country, price_group=(pol or {}).get("price_group") or "",
ref_sales_order_id=first["order"], ref_line_no=first["line"], ref_invoice_id=first["invoice"])
o.priority = priority or "normal"
if ship_to_address:
o.ship_to_address_lines, o.ship_to_address_city = json.dumps(ship_to_address.lines), ship_to_address.city
o.ship_to_address_postcode, o.ship_to_address_region = ship_to_address.postcode or "", ship_to_address.region or ""
o.owner_sub = (caller.get("sub") or "").strip().lower()
o.deposit_deduction_mode = (pol or {}).get("deposit_deduction_mode") or repo.settings.value_on("default_deposit_deduction_mode", today) or "proportional"
taking, unmet = {}, []
asks = [] if code in RETURN_TYPES else [(li.product_id, li.quantity, li.unit_code) for li in lines]
with _prefetched(caller, o, asks, today): # v10: every line priced in ONE call (product v7 PRD-41)
for li, ref in zip(lines, refs):
if code not in MEMO_TYPES and (li.unit_price is not None or li.description):
raise Refused(422, "the price comes from the product module: only a credit or debit request states its own amount and text")
if code in RETURN_TYPES:
spec, why = _return_spec(repo, caller, otype, li, ref, today, taking)
if why:
unmet += why
continue
elif code in MEMO_TYPES:
spec = _memo_spec(caller, o, li, ref, today)
else:
spec = _price(caller, o, li.product_id, li.quantity, li.unit_code, today)
spec.ref_sales_order_id, spec.ref_line_no, spec.ref_invoice_id = ref["order"], ref["line"], ref["invoice"]
spec.ref_invoice_line_no, spec.ref_delivery_ref = ref["invoice_line"], ref["delivery"]
if li.instalments:
spec.billing_kind, spec.billing_plan, spec.instalments = "instalment", "monthly", li.instalments
_entered_tax(li, spec)
_categorise(repo, o, spec, today, li.usage or "normal")
line = o.add_line(spec)
if li.milestones:
o.set_milestones(line.line_no, [m.model_dump() for m in li.milestones])
if unmet:
raise GateUnmet(unmet)
o.price_list_id = o.lines[0].priced_from.split(" ")[0] if o.lines else ""
_reprice(repo, o, today)
_apply_tax(caller, o)
require(_consignment_gates(caller, o))
for key, qty in taking.items():
repo.claims.claim(key, "return", sid, quantity=qty) # the returned quantity is claimed with the draft: a second return sees it
return _saved(repo, caller, o, 0)
return _retrying(None, unit)
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/lines", permission=WRITE, if_match="version")
def add_order_line(caller: dict, sales_order_id: Ref, line: LineIn, version: Optional[int] = None) -> dict:
"""Add a line to a DRAFT order: priced by find_price (priced_on today), taxed by the tax module. A confirmed order
changes only by an amendment."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
today = _today(repo)
spec = _price(caller, o, line.product_id, line.quantity, line.unit_code, today)
_entered_tax(line, spec)
_categorise(repo, o, spec, today, line.usage or "normal")
added = o.add_line(spec)
changed = _reprice(repo, o, today)
_apply_tax(caller, o, {added.line_no} | changed)
return _saved(repo, caller, o, current)
@SO.use_case(kind="command", route="DELETE /sales-orders/{sales_order_id}/lines/{line_no}", permission=WRITE, destructive=True,
if_match="version")
def remove_order_line(caller: dict, sales_order_id: Ref, line_no: LineNo, version: Optional[int] = None) -> dict:
"""Remove a line from a DRAFT order."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
o.remove_line(line_no)
changed = _reprice(repo, o, _today(repo))
_apply_tax(caller, o, changed)
return _saved(repo, caller, o, current)
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/recalculate-tax", permission=WRITE, idempotent=True,
if_match="version")
def recalculate_tax(caller: dict, sales_order_id: Ref, version: Optional[int] = None) -> dict:
"""Ask the tax module again for a DRAFT order's tax (after a tax change, or when it was unreachable). The answer is
stored as the order's own copy; confirmation freezes it."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
if o.lifecycle != "draft":
raise Refused(409, f"the order is {o.lifecycle}: its tax is frozen")
for ln in o.lines:
ln.tax_code = ""
note = _apply_tax(caller, o)
out = _saved(repo, caller, o, current)
return {**out, "tax_note": note} if note else out
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/customer-po", permission=WRITE, idempotent=True, if_match="version")
def record_customer_po(caller: dict, sales_order_id: Ref, customer_po_ref: Ref, version: Optional[int] = None) -> dict:
"""Record the customer's purchase order number on a DRAFT order, so their accounts-payable team can match our invoice:
every invoice for the order carries it. Where the customer's policy requires one, confirming without it is refused."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
if o.customer_po_ref == customer_po_ref.strip():
return _view(o, current, caller, _issued(o))
o.record_customer_po(customer_po_ref)
return _saved(repo, caller, o, current)
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/override-credit-hold", permission=APPROVE, idempotent=True,
if_match="version")
def override_credit_hold(caller: dict, sales_order_id: Ref, reason: Reason, version: Optional[int] = None) -> dict:
"""An order admin overrides the customer's credit hold on this draft, with a reason (recorded on the order). The rep then
confirms. Where the hold mode is only a warning, no override is needed."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
if o.credit_override_by and o.credit_override_reason == reason.strip():
return _view(o, current, caller, _issued(o))
o.override_credit(reason, _who(caller))
return _saved(repo, caller, o, current)
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/cancel-draft", permission=WRITE, idempotent=True, if_match="version")
def cancel_draft_order(caller: dict, sales_order_id: Ref, version: Optional[int] = None) -> dict:
"""Cancel a draft order. A repeat changes nothing."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
if not o.cancel_draft():
return _view(o, current, caller)
if o.order_type_code in RETURN_TYPES: # the quantity the draft held is released by a negative claim row
for ln in o.lines:
repo.claims.claim(_key_return(ln.ref_sales_order_id, ln.ref_line_no), "return", o.sales_order_id, quantity=-ln.quantity)
return _saved(repo, caller, o, current)
# ═══ the accepted quote ════════════════════════════════════════════════════════════════════════
@SO.use_case(kind="command", route="POST /sales-orders/from-quote", permission=INTAKE, idempotent=True, tool=False)
def order_from_quote(caller: dict, quote_id: Ref, quote_version: QuoteVersion, bp_id: BigId, currency: Currency,
lines: list[QuoteLineIn], owner: Optional[Ref] = None, customer_po_ref: Optional[Ref] = None) -> dict:
"""An accepted quote version becomes a draft order in one step (SO-01), keyed sales_order_id = quote:<quote_id>@v<version>,
so handing it over twice returns the same order. The composition's adapter reads brick-quote's get_quote for the accepted
version and passes its lines as they arrived: product, quantity, list_price, unit_price, unit_code (the product's sales
unit when the quote line predates units), the quote's price book and priced_on. The accepted prices are COPIED, never
re-priced; priced_from names the quote line. Tax is computed by the tax module for the order date. Where the control
confirm_on_acceptance is on, the order is confirmed at once through the same readiness gates as a person."""
sid = f"quote:{quote_id}@v{quote_version}"
def unit():
with repository() as repo, _decide():
found = repo.sales_orders.get(sid)
if found is not None:
return _view(found[0], found[1], caller, _issued(found[0]), prices=True)
today = _today(repo)
for ql in lines: # a price from a second price book never reaches an order: refuse BEFORE any other work
if ql.priced_from is None or not ql.priced_from.price_list_id:
raise GateUnmet([no("priced_by_product", f"quote line {ql.line_no} ({ql.product_id}) is not priced by a product price list",
"re-price the quote from product before it is sent")])
pct, terms = _policy_defaults(repo, int(bp_id), today)
o = SalesOrder.draft(sid, owner=(owner or _who(caller)).strip().lower(), bp_id=int(bp_id), currency=_currency(currency),
ordered_on=None, today=today, source_kind="quote", source_id=sid,
customer_po_ref=customer_po_ref or "", delivery_terms=terms, deposit_pct=pct)
with _prefetched(caller, o, [(ql.product_id, ql.quantity, ql.unit_code) for ql in lines], today): # v10: one batch (PRD-41)
for ql in lines:
try: # the unit and its conversion are the product's; the PRICE is the quote's
got = price_answer(_ask_price(caller, o, ql.product_id, ql.quantity, ql.unit_code, today))
except Missing as m:
raise _refuse(m, f"product {ql.product_id} (quote line {ql.line_no})") from None
pf, pday = ql.priced_from, day(ql.priced_on) or today
where = (f"quote:{quote_id}@v{quote_version}#{ql.line_no} ({pf.price_list_id}"
+ (f" rule {pf.rule_seq}" if pf.rule_seq is not None else "") + f" on {pday})")
pu = ql.pricing_unit_code if ql.pricing_unit_code and ql.pricing_unit_code != got.unit_code else ""
pn, pd = 1, 1
if pu: # priced in a unit other than the line's: its conversion is product's own answer
if got.pricing_unit_code != pu:
raise Refused(422, f"quote line {ql.line_no}: {ql.product_id} is priced per {pu} on the quote but product now answers "
f"per {got.pricing_unit_code or got.unit_code} — re-price the quote from product")
pn, pd = got.pricing_numerator, got.pricing_denominator
spec = LineSpec(ql.product_id, ql.description or got.description or ql.product_id, got.unit_code, ql.quantity,
got.numerator, got.denominator, ql.list_price, ql.unit_price, ql.discount_pct or Decimal(0),
pday, where, got.tax_category, got.invoicing_policy, got.billing_kind,
got.billing_plan, quote_ref=f"{quote_id}|{quote_version}|{ql.line_no}",
pricing_quantity=ql.pricing_quantity or 1, pricing_unit_code=pu, pricing_numerator=pn,
pricing_denominator=pd, category_code=got.category_code, product_type_code=got.product_type_code,
gross_weight=got.gross_weight, weight_unit_code=got.weight_unit_code, measure_valid_from=got.measure_valid_from)
_categorise(repo, o, spec, today)
o.add_line(spec)
_apply_tax(caller, o)
if repo.settings.value_on("confirm_on_acceptance", today):
_confirm_in(repo, caller, o, 0, _who(caller) or "system", must=False)
return _saved(repo, caller, o, 0, prices=True)
out = _retrying(None, unit)
if _complete_sends(sid):
out = _fresh(caller, sid, prices=True)
return out
# ═══ confirm ═══════════════════════════════════════════════════════════════════════════════════
def _copy_delivery_rules(repo, o: SalesOrder, otype, today: date) -> None:
"""At confirmation the order takes the customer's delivery rules as they are that day (SO-37); the rep may change them on the order
until the first delivery (set_fulfilment_rules)."""
pol = repo.policies.in_force(o.bp_id, today) or {}
o.partial_delivery = pol.get("partial_delivery") or "allowed"
o.max_partials = int(pol.get("max_partials") or 0)
o.under_tolerance_pct = Decimal(pol.get("under_tolerance_pct") or 0)
o.over_tolerance_pct = Decimal(pol.get("over_tolerance_pct") or 0)
o.combine_allowed = bool(pol.get("combine_allowed"))
o.min_shelf_life_days = int(pol.get("min_shelf_life_days") or 0)
o.priority = "rush" if "rush" in (o.priority, otype.priority) else "normal" # the order's own rush, or its type's
def _series_id(repo, today: date) -> str:
return f"SO-{today.year}" if repo.settings.value_on("number_restarts_yearly", today) is not False else "SO"
def _other_marks(repo, series_id: str) -> list:
"""High-water marks of the other series, so a series started by switching the yearly restart off continues after them."""
return [m for sid, m in repo.number_series.marks() if sid != series_id]
def _stage_send(repo, sales_order_id: str, event: str, payload: dict) -> None:
"""Owe a neighbour a call that cannot refuse the action (a billing schedule, a purchase request, a draft cancellation): the row commits WITH the
action and the call is made after it (_complete_sends), so a refused action leaves nothing in the neighbour and a failed call is retried."""
repo.postings.add({"document_id": sales_order_id, "event": event, "journal_id": "", "status": "pending", "reason": "", "attempts": 0,
"rebate_agreement_id": "", "payload_json": json.dumps(payload)})
def _address_of(o: SalesOrder) -> dict:
"""The ship-to as the one address object (v8). An order with no typed address answers the country alone (inventory's delivery check
reads only that); purchasing needs the whole object (_stage_purchase refuses without it)."""
if not o.ship_to_address_lines:
return {"country": o.ship_to_country}
return {"lines": json.loads(o.ship_to_address_lines), "city": o.ship_to_address_city, "postcode": o.ship_to_address_postcode or None,
"region": o.ship_to_address_region or None, "country": o.ship_to_country}
def _stage_purchase(repo, o: SalesOrder, ln, replaces: str = "") -> None:
"""A third-party line is ordered from purchasing once per revision (request_ref so:{order}:{line}@r{revision}); an amendment's new request
replaces (cancels) the previous one when it is sent."""
if not ports.purchasing_installed():
raise Refused(409, "purchasing: that module is not part of this installation, so a third-party line cannot be ordered")
if not o.ship_to_address_lines:
raise Refused(422, f"line {ln.line_no} is shipped by the supplier: the delivery address is needed (ship_to_address)", gate="ship_to_address",
fix="type the delivery address on the order")
_stage_send(repo, o.sales_order_id, f"send:purchase:{ln.line_no}@r{o.revision}", {
"op": "purchase_request", "replaces": replaces, "args": {
"request_ref": f"so:{o.sales_order_id}:{ln.line_no}@r{o.revision}", "product_id": ln.product_id, "quantity": num(ln.quantity),
"unit_code": ln.unit_code, "needed_by": (o.requested_delivery_on or _today(repo)).isoformat(), "ship_to_bp_id": o.ship_to_bp_id or o.bp_id,
"ship_to_address": _address_of(o), "sales_order_id": o.sales_order_id, "so_line_no": ln.line_no}})
def _complete_sends(sales_order_id: str) -> bool:
"""Make every call the order's committed actions still owe (rows send:% in so_posting), then mark each posted; a purchase request's id is
saved on its line. Each call is idempotent on its key, so a retry after a failure completes the row without sending twice. A call a neighbour
refuses stays pending with its reason; the action it belongs to is not undone. -> True when anything was owed."""
with repository() as repo:
todo = repo.postings.pending_sends(sales_order_id)
if not todo:
return False
done, failed = {}, {}
for p in todo:
pl, key = json.loads(p["payload_json"]), p["event"]
try:
if pl["op"] == "plan_billing":
ports.invoicing.plan_billing(**pl["args"])
done[key] = ""
elif pl["op"] == "purchase_request":
if not ports.purchasing_installed():
raise Missing("purchasing", "that module is not part of this installation", 409, installed=False)
done[key] = str(ports.purchasing.purchase_request(**pl["args"]).get("request_id") or "")
if pl.get("replaces"):
ports.purchasing.cancel_purchase_request(request_id=pl["replaces"], reason="the line was amended")
elif pl["op"] == "cancel_drafts":
ports.invoicing.cancel_drafts(pl["drafts"])
done[key] = ""
except Missing as m:
done.pop(key, None)
failed[key] = m.reason
def mark():
with repository() as repo, _decide():
o, current = _get(repo, sales_order_id)
changed = False
for p in todo:
key, pl = p["event"], json.loads(p["payload_json"])
row = {"document_id": sales_order_id, "event": key, "journal_id": "", "reason": "", "rebate_agreement_id": "", "payload_json": p["payload_json"]}
if key in done:
if pl["op"] == "purchase_request" and done[key] and o.line(pl["args"]["so_line_no"]) is not None:
o.line(pl["args"]["so_line_no"]).purchase_request_id, changed = done[key], True
repo.postings.add({**row, "status": "posted", "attempts": int(p.get("attempts") or 0) + 1})
elif key in failed:
repo.postings.add({**row, "status": "pending", "reason": failed[key], "attempts": int(p.get("attempts") or 0) + 1})
if changed:
repo.sales_orders.save(o, current)
retry_on_conflict(mark)
return True
def _foc_above_limit(repo, o: SalesOrder, today: date) -> bool:
return _foc_value(o) > Decimal(str(repo.settings.value_on("free_of_charge_limit", today) or "250.00"))
def _confirm_in(repo, caller: dict, o: SalesOrder, current: int, by: str, must: bool = True) -> bool:
"""Confirm inside the caller's unit of work: every readiness gate, then the next number from the series, the repeating
lines handed to invoicing as schedules, and the order's prices and tax frozen. With must=False an unmet gate leaves the
order a draft (the system confirming on acceptance never forces it). The series and the order save in one transaction."""
today = _today(repo)
gates = gate_rules.confirm_gates(o, _facts(repo, caller, o, today))
if not must and any(not g.met for g in gates):
return False
require(gates)
otype = _need_type(repo, o.order_type_code, today)
if otype.series_id == "SO": # STANDARD numbers by the owner's format and yearly restart (settings)
fmt = repo.settings.value_on("order_number_format", today) or "SO-{yyyy}-{seq:4}"
series_id = _series_id(repo, today)
else: # every other type has its own series: RE-2026-0001, CR-2026-0001, FC-...
fmt, series_id = f"{otype.series_id}-{{yyyy}}-{{seq:4}}", f"{otype.series_id}-{today.year}"
series, sver = repo.number_series.get_or_new(series_id)
if series.series_id == "SO" and sver == 0 and series.high_water_mark == 0: # the yearly restart was switched off: carry on after the yearly series
series.high_water_mark = max([0, *(s for s in _other_marks(repo, series.series_id))])
while True:
order_no = render_number(fmt, today.year, series.claim())
if repo.order_no_claims.claim(order_no, o.sales_order_id): # a number already held by another order is never reused
break
for ln in o.lines:
if ln.repeating:
try:
price, pct = ln.billed_price()
periods = ln.quantity
if ln.billing_kind == "instalment": # equal instalments of a fixed total: n periods of total / n, exactly
each = money(ln.net_amount / ln.instalments)
if each * ln.instalments != ln.net_amount:
raise Refused(422, f"line {ln.line_no}: {num(ln.net_amount, 2)} does not divide into {ln.instalments} equal instalments to the penny")
price, pct, periods = each, Decimal(0), Decimal(ln.instalments)
if not ports.invoicing_installed():
raise Refused(409, f"invoicing (billing schedule for line {ln.line_no}): that module is not part of this installation")
_stage_send(repo, o.sales_order_id, f"send:billing:{ln.line_no}", {"op": "plan_billing", "args": {
"schedule_id": f"{o.sales_order_id}|{ln.line_no}", "bp_id": o.bp_id, "currency": o.currency,
"line": {"line_no": ln.line_no, "product_id": ln.product_id, "starts_on": today.isoformat(),
"quantity": num(periods), "unit_code": ln.unit_code, "unit_price": num(price, 2),
"discount_pct": num(pct), "billing_kind": ln.billing_kind,
"billing_plan": ln.billing_plan, "description": ln.description,
"tax_category": ln.tax_category}}})
except Missing as m:
raise _refuse(m, f"invoicing (billing schedule for line {ln.line_no})") from None
o.confirm(gates=gates, order_no=order_no, now=_bnow(), today=today)
_copy_delivery_rules(repo, o, otype, today)
o.release_on_confirmation(today) # the milestones that fall due with the order
if o.order_type_code != "FREE_OF_CHARGE" or _foc_above_limit(repo, o, today): # free of charge waits for a controller only ABOVE the limit
o.default_block(otype.default_block, _bnow())
for ln in o.lines:
if ln.item_category_code == "THIRD_PARTY": # the supplier sends it: purchasing is asked once, keyed per line and revision
if not ports.purchasing_installed():
raise Refused(409, "purchasing: that module is not part of this installation, so a third-party line cannot be ordered")
_stage_purchase(repo, o, ln)
if otype.direction == "in": # goods coming back: inventory receives them against the RMA, nothing is reserved
for ln in o.lines:
ln.stocked = ports.inventory_installed()
if o.order_type_code == "RETURNS" and ports.inventory_installed():
_expect_return(repo, o, today)
elif otype.delivery_relevant:
if o.priority == "rush" and not o.requested_delivery_on:
o.requested_delivery_on = today # a RUSH order asks for a delivery today
promising.promise_order(repo, o, today, at=_bnow(), reason="promised at confirmation", exclude_own=False)
_reserve(o, promising.warehouse(repo, today)) # revision 1; the order is confirmed in memory, nothing is saved until this answers
repo.number_series.save(series, sver)
return True
def _expect_return(repo, o: SalesOrder, today: date) -> None:
"""authorise_return: inventory is told to expect the RMA (so:{sales_order_id}, idempotent): it receives a customer return only against an open one."""
wh = promising.warehouse(repo, today)
if not wh:
raise Refused(409, "inventory (expected return): no delivering warehouse is set (the setting default_warehouse_id), so it cannot be told where the return arrives")
reason = _return_reason(repo, o.return_reason)
lines = [{"line_no": ln.line_no, "product_id": ln.product_id, "unit_code": ln.unit_code, "quantity": num(ln.quantity), "unit_numerator": ln.unit_numerator,
"unit_denominator": ln.unit_denominator, "original_delivery_ref": ln.ref_delivery_ref, "serials": [], "inspect": bool(reason.inspect) if reason else False}
for ln in o.lines]
try:
ports.inventory.expect_return(rma_ref=f"so:{o.sales_order_id}", bp_id=o.bp_id, warehouse_id=wh, lines=lines)
except Missing as m:
raise _refuse(m, "inventory (expect the return)") from None
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/confirm", permission=CONFIRM, idempotent=True, if_match="version")
def confirm_order(caller: dict, sales_order_id: Ref, version: Optional[int] = None) -> dict:
"""Confirm a draft order. Every readiness gate must be met (customer_active, lines_priced, units_valid, lines_taxed,
credit, po_ref, no_open_suggestion; get_order_readiness shows them first, each {gate, met, reason, fix}); otherwise it
is refused naming the gate, its reason and its fix. A credit hold (mode block) is overridden only by an order admin, with
a reason. Confirming numbers the order from the series (SO-2026-0001), freezes its prices and tax, and hands repeating
lines to invoicing as billing schedules. A repeat changes nothing."""
def unit():
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
if o.lifecycle != "draft":
if o.lifecycle == "confirmed":
return _view(o, current, caller, _issued(o))
raise Refused(409, f"the order is {o.lifecycle}, not a draft")
_confirm_in(repo, caller, o, current, _who(caller))
return _saved(repo, caller, o, current)
out = _retrying(version, unit)
if _complete_sends(sales_order_id): # the billing schedules and purchase requests the confirmation owes, after it committed
out = _fresh(caller, sales_order_id)
if out.get("order_type_code") == "CASH" and out.get("lifecycle") == "confirmed":
try:
out = _invoice_flow(caller, sales_order_id, None, None) # a CASH order is invoiced (and issued) in the same step
except Refused as r:
if "something_due" not in r.reason:
raise
return out
# ═══ the confirmation document ═════════════════════════════════════════════════════════════════
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/send-confirmation", permission=WRITE, if_match="version")
def send_confirmation(caller: dict, sales_order_id: Ref, address: Ref, channel: Code = "email", version: Optional[int] = None) -> dict:
"""Send the order confirmation: the document (number, lines, prices, tax, delivery terms, their PO) is produced from the
FROZEN order and the send is recorded (who, where, which revision). Sending again records a second send. This brick
records and renders; delivery of the mail itself is the composition's mail capability."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
o.record_send(channel=channel, address=address, by=_who(caller), at=_stamp())
out = _saved(repo, caller, o, current)
out["document"] = {"title": f"Order confirmation {o.order_no}", "order_no": o.order_no, "revision": o.revision,
"customer_po_ref": o.customer_po_ref, "delivery_terms": o.delivery_terms, "currency": o.currency,
"lines": [{k: v for k, v in ln.items() if k in ("line_no", "description", "quantity", "unit_code", "unit_price", "net_amount", "tax_code", "tax_amount",
"promise", "schedule")}
for ln in out["lines"]],
"subtotal": o.subtotal, "tax_total": o.tax_total, "total": o.total}
return out
# ═══ deliveries ════════════════════════════════════════════════════════════════════════════════
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/deliveries", permission=DELIVER, idempotent=True, if_match="version")
def record_delivery(caller: dict, sales_order_id: Ref, line_no: LineNo, quantity: str, delivery_ref: Ref,
delivered_on: Optional[IsoDate] = None, corrects_ref: Optional[Ref] = None, source: Optional[Code] = None,
version: Optional[int] = None) -> dict:
"""Record what was delivered or done on a line, with the delivery note's reference. Recording the same delivery note
twice records it once. Delivering more than ordered, or dating a delivery after today or before the confirmation, is
refused. A mistake is corrected by a NEW row: a negative quantity with corrects_ref naming the original (never an edit,
and never below what is already invoiced). With inventory installed a STOCKED line accepts only source inventory (the call
inventory makes inside its validation, delivery_ref inv:<delivery number>; only its service client may name that source) and
a hand delivery on it is refused; a service line is delivered by hand, or by timesheets' service client (source timesheets, delivery_ref ts:<sheet_id>:r<rev>:<line_no>, SO-51: service lines only, a correction names corrects_ref). Every quantity converts EXACTLY to the product's base
unit and is recorded beside the line's own; one that does not is refused, never rounded."""
from decimal import InvalidOperation
src = (source or "manual").strip().lower()
if src not in ("manual", "inventory", "purchasing", "timesheets"):
raise Refused(422, f"a delivery's source is manual, inventory, purchasing or timesheets, not {source!r}")
if src == "timesheets" and not (delivery_ref or "").startswith("ts:"):
raise Refused(422, "a timesheets delivery's delivery_ref is ts:<sheet_id>:r<rev>:<line_no>")
if src == "timesheets" and not delivered_on:
raise Refused(422, "a timesheets delivery names delivered_on, the approval's business date")
if src in ("inventory", "purchasing", "timesheets") and caller.get("auth_source") not in ("platform", "service"):
raise Refused(403, f"only {src}'s service client records a delivery with source {src}")
try:
q = Decimal(str(quantity))
on = day(delivered_on, "delivered_on")
if not q.is_finite() or q == 0 or abs(q) >= Decimal("1e12") or q.as_tuple().exponent < -4:
raise ValueError("a quantity is an exact non-zero number with up to 4 places")
except (InvalidOperation, ValueError) as e:
raise Refused(422, str(e) if "quantity" in str(e) or "date" in str(e) else "a quantity is an exact number such as \"5\"") from None
def unit():
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
today = _today(repo)
issued = _issued(o, strict=bool(corrects_ref))
d = o.record_delivery(line_no, q, delivery_ref, on or today, today=today, by=_who(caller), issued=issued,
kind="correction" if corrects_ref else "delivery", corrects_ref=corrects_ref or "", source=src,
inventory=ports.inventory_installed(),
hold=tuple(g for g in _delivery_gates(repo, o, today) if g.gate != "delivery_block"))
if d is None:
return _view(o, current, caller, issued)
d.quantity_base = _base_quantity(caller, o.line(line_no), q) # exact, or refused before anything is saved
skip = _nonbilling(repo, o, today)
if repo.settings.value_on("lock_done_orders", today) is not False and o.can_finish(_issued(o, strict=False), skip):
o.finish(_issued(o), skip)
return _saved(repo, caller, o, current)
out = _retrying(version, unit)
if out.get("order_type_code") in RETURN_TYPES:
_credit_return(sales_order_id, "receipt") # a return credited on receipt is credited now; on inspection, when inventory inspects
with repository() as repo:
o, v = _get(repo, sales_order_id) # the return may have closed meanwhile: the caller just delivered on it, so it is theirs to see
out = _view(o, v, caller)
return out
# ═══ invoicing ═════════════════════════════════════════════════════════════════════════════════
def _deposit_state(repo, o: SalesOrder) -> list:
"""The ISSUED deposits of the order, oldest first, each with what it has left (its amount less every deduction and credit claimed against it).
A deposit still a draft at invoicing is not here: nothing is deducted from it."""
deps = sorted((r for r in o.invoice_requests if r.kind == "deposit" and r.live and r.invoice_id), key=lambda r: r.request_no)
if not deps or not ports.invoicing_installed():
return []
try:
by_invoice = ports.invoicing.issued_sources([r.invoice_id for r in deps]) # ar keys /invoice-totals by ITS invoice_id, never by our source id
got = {f"so:{o.sales_order_id}:{r.request_no}": by_invoice.get(r.invoice_id) for r in deps}
except Missing as m:
raise Refused(502, f"invoicing did not answer ({m.reason}), so whether a deposit invoice is issued cannot be confirmed — try again") from None
return [{"invoice_id": r.invoice_id, "request_no": r.request_no,
"tax_category": (r.lines[0].get("tax_category") if isinstance(r.lines, list) and r.lines else "") or "",
"amount": r.amount, "remaining": r.amount - repo.claims.deducted(r.invoice_id)}
for r in deps if got.get(f"so:{o.sales_order_id}:{r.request_no}") or r.requested_by == "migration"] # a migrated deposit was posted in Odoo
def _credit_line(x: dict, sales_order_id: str) -> dict:
"""One credit line as invoicing's spec lists it: against the original invoice's line, carrying the order's own line (never the original's)."""
out = {k: x[k] for k in ("invoice_line_no", "quantity", "unit_price", "discount_pct", "net_amount", "tax_category") if x.get(k) not in (None, "")}
if x.get("line_no"):
out |= {"sales_order_id": sales_order_id, "so_line_no": int(x["line_no"])}
return out
def _complete_pending(caller: dict, sales_order_id: str, loaded: Optional[SalesOrder] = None) -> dict:
"""Tell invoicing about every request this order claimed locally and has not yet sent (status pending), then mark each requested. Every call
is idempotent on its source (so:{sales_order_id}:{request_no}), so a retry after a failed call completes the row without sending twice."""
with repository() as repo, _decide():
o = loaded if loaded is not None else _get(repo, sales_order_id)[0] # `loaded`: the order the caller just saved, not read again (13 round trips)
todo = [r for r in o.invoice_requests if r.status == "pending"]
done = {}
for r in todo:
src = f"so:{o.sales_order_id}:{r.request_no}"
try:
if r.kind == "credit":
got = ports.invoicing.credit_note(source_id=src, bp_id=o.payer, currency=o.currency, invoice_id=r.source_ref,
reason=(f"Return {o.order_no}: {o.return_reason}" if o.order_type_code in RETURN_TYPES else
f"Credit request {o.order_no}: {o.return_reason or 'price correction'}" if o.order_type_code == "CREDIT_REQ"
else f"Unused deposit: order {o.order_no} cancelled"),
lines=[_credit_line(x, o.sales_order_id) for x in r.lines])
else:
got = ports.invoicing.raise_draft(source_id=src, bp_id=o.payer, currency=o.currency, kind="regular" if r.kind == "debit" else r.kind,
lines=[{**x, "sales_order_id": o.sales_order_id} for x in r.lines], customer_po_ref=o.customer_po_ref,
sold_to_bp_id=o.bp_id, ship_to_bp_id=o.ship_to_bp_id or o.bp_id, bill_to_bp_id=o.bill_to,
deductions=[{k: v for k, v in d.items() if k != "request_no"} for d in r.deductions],
issue_hint="cash" if o.order_type_code == "CASH" else "")
except Missing as m:
raise _refuse(m, "invoicing") from None
done[r.request_no] = str(got.get("credit_id") or got.get("invoice_id") or src)
def mark():
with repository() as repo, _decide():
o, current = _get(repo, sales_order_id)
for r in o.invoice_requests:
if r.status == "pending" and r.request_no in done:
r.status, r.invoice_id = "requested", done[r.request_no]
if todo:
if o.order_type_code in MEMO_TYPES and o.lifecycle == "confirmed" and not any(r.status == "pending" for r in o.invoice_requests):
o.finish_memo()
if o.order_type_code in RETURN_TYPES and o.lifecycle == "confirmed" and _return_done(repo, o):
o.close_return()
return _saved(repo, caller, o, current)
return _view(o, current, caller, _issued(o))
return retry_on_conflict(mark)
def _return_done(repo, o: SalesOrder) -> bool:
"""close_return: every line received, inspected where its reason inspects, and credited for what is accepted (or, uninspected, received)."""
reason = _return_reason(repo, o.return_reason)
inspect = bool(reason and reason.inspect)
for ln in o.lines:
got = o.delivered(ln.line_no)
acc, rej = o.inspected(ln.line_no)
if got <= 0 or (inspect and acc + rej < got) or o.credited_qty(ln.line_no) < (acc if inspect else got):
return False
return True
def _invoice_flow(caller: dict, sales_order_id: str, version: Optional[int], manual: Optional[list]) -> dict:
"""Ask invoicing for what is due, in three steps so nothing is claimed twice and nothing is lost: (1) one transaction validates, claims every
deposit deduction (plain INSERT per deposit), appends the request as PENDING and saves; (2) invoicing is asked, idempotent on the request's key;
(3) the request is marked requested. A call that fails leaves the pending request; asking again completes THAT request, it never makes another."""
with repository() as repo, _decide():
o, current = _load(repo, caller, sales_order_id)
_same_version(f"sales order {sales_order_id}", current, version)
if not ports.invoicing_installed():
raise Refused(409, "invoicing: that module is not part of this installation")
if not any(r.status == "pending" and r.kind in ("regular", "final") for r in o.invoice_requests):
today = _today(repo)
require(_billing_gates(repo, o, today)) # a block or a missing field is refused before anything is asked of invoicing
skip = _nonbilling(repo, o, today)
issued = _issued(o, strict=True)
deposits = _deposit_state(repo, o)
typed = {}
for d in manual or []:
if not any(x["invoice_id"] == d.deposit_invoice_id for x in deposits):
raise GateUnmet([no("deposit_remaining", f"deposit {d.deposit_invoice_id} is not an issued deposit of this order (a draft is deducted once issued)",
"name an issued deposit")])
typed[d.deposit_invoice_id] = typed.get(d.deposit_invoice_id, Decimal(0)) + d.amount
plan = o.plan_invoice(issued, skip=skip, deposits=deposits, manual=typed)
req = o.add_invoice_request(plan["kind"], plan["lines"], plan["net"] - plan["deduct"], plan["deduct"], by=_who(caller), now=_stamp(),
status="pending", deductions=[{k: v for k, v in d.items() if k != "request_no"} for d in plan["deductions"]])
for d in plan["deductions"]:
repo.claims.claim_deposit(d["deposit_invoice_id"], "deduct", o.sales_order_id, req.request_no, Decimal(d["net_amount"]), d["tax_category"])
repo.sales_orders.save(o, current)
saved = o
return _complete_pending(caller, sales_order_id, loaded=saved)
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/request-invoice", permission=INVOICE, if_match="version")
def request_invoice(caller: dict, sales_order_id: Ref, deductions: Optional[list[DeductionIn]] = None, version: Optional[int] = None) -> dict:
"""Ask invoicing for what is due on a confirmed order: each line's ordered quantity (policy ordered) or delivered
quantity (policy delivered) not yet invoiced, a milestone line's released stages by value, less the deposits' deductions (per deposit, oldest
first, in the deposit's tax category: proportional to the share of the order this invoice bills, full_first, or manual: give `deductions`
[{deposit_invoice_id, amount}]; the final invoice takes all that is left). Invoicing receives ONE draft request keyed
on the order and the request number and owns the invoice; the order never writes one. With nothing new due the call
is refused (gate something_due) and creates nothing. A free-of-charge order, a return and a memo