dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/application/fulfilment.py
"""Who receives an order, what is missing, what is on hold and how it may be delivered (SO-37 to SO-40).
* SO-40: a customer's ship-tos, bill-to and payer are dated partner rows (from tomorrow); an order copies them when it is raised, the credit
check reads the PAYER, tax the ship-to's country, and invoicing receives the payer (whose receivable it is) with the sold-to, ship-to and bill-to.
* SO-38: dated incompletion rules per order type say which empty field blocks confirmation, delivery or billing.
* SO-39: a delivery or billing block is an appended row (who, why), released by a later row; only so:order:block sets or releases one.
* SO-37: the fulfiller's view of an order (no prices): the delivery rules copied from the customer's policy at confirmation, the schedule and
whether delivery is held.
"""
import json
from typing import Annotated, Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId
from pydantic import Field
from brick_sales_order.application import use_cases as uc
from brick_sales_order.application.inputs import Code, Day, Pct, Reason, Ref, TransitIn
from brick_sales_order.domain import rules as rl
from brick_sales_order.domain.errors import RuleBroken
SO, READ, WRITE, PRICING, BLOCK, SETTINGS_WRITE = uc.SO, uc.READ, uc.WRITE, uc.PRICING, uc.BLOCK, uc.SETTINGS_WRITE
# ── configuration: partners, incompletion rules, block reasons ────────────────────────────────
@SO.use_case(kind="command", route="POST /customer-partners", permission=PRICING, idempotent=True)
def set_customer_partner(caller: dict, sold_to_bp_id: BigId, function: Literal["ship_to", "bill_to", "payer"], partner_bp_id: BigId, valid_from: Day,
is_default: bool = False, status: Literal["active", "ended"] = "active", transit_days: Optional[list[TransitIn]] = None) -> dict:
"""Name one of a customer's ship-tos, its bill-to or its payer, from a date AFTER today (a row is never edited: a change, or ending a partner,
is a new row from a later date). The default row of a function is what a new order takes; transit_days says how far a ship-to is from each
warehouse, for the promised date. The partner must be a business partner (checked when an order is confirmed)."""
with uc.repository() as repo, uc._decide():
on = uc._future(repo, valid_from)
row = rl.CustomerPartner(int(sold_to_bp_id), function, int(partner_bp_id), on, bool(is_default), json.dumps({t.warehouse_id: t.days for t in transit_days or []},
sort_keys=True), status, uc._who(caller))
old = next((r for r in repo.config.partners() if (r.sold_to_bp_id, r.function, r.partner_bp_id, r.valid_from) ==
(row.sold_to_bp_id, row.function, row.partner_bp_id, on)), None)
if old is not None and old == row:
return uc._plain(dict(old.__dict__))
repo.config.add_partner(row)
return uc._plain(dict(row.__dict__))
@SO.use_case(kind="command", route="POST /incompletion-rules", permission=PRICING, idempotent=True)
def add_incompletion_rule(caller: dict, order_type_code: Code, field: Code, valid_from: Day, blocks: Literal["confirm", "delivery", "billing"],
condition: Literal["always", "export_ship_to", "customer_policy_requires"], reason_text: Ref) -> dict:
"""Say that an order of a type needs a field before confirmation, delivery or billing, from a date after today: always, only for an export
order (a ship-to outside the company's country), or where the customer's policy asks. The fields it knows: delivery_terms, customer_po_ref,
return_reason, reference_invoice."""
if field not in rl.FIELDS:
raise Refused(422, f"an incompletion rule names one of {', '.join(sorted(rl.FIELDS))}, not {field}")
with uc.repository() as repo, uc._decide():
on = uc._future(repo, valid_from)
repo.config.add_incompletion_rule(rl.IncompletionRule(order_type_code, field, on, blocks, condition, reason_text))
return {"order_type_code": order_type_code, "field": field, "valid_from": on, "blocks": blocks, "condition": condition}
@SO.use_case(kind="command", route="POST /block-reasons", permission=SETTINGS_WRITE, idempotent=True)
def add_block_reason(caller: dict, block_reason_code: Code, kind: Literal["delivery", "billing"], name: Ref) -> dict:
"""Add a reason an order admin can give when blocking delivery or billing (CHECK_TERMS, PRICE_DISPUTE ...). APPROVAL is seeded."""
with uc.repository() as repo, uc._decide():
repo.config.add_block_reason(rl.BlockReason(block_reason_code, kind, name))
return {"block_reason_code": block_reason_code, "kind": kind, "name": name}
# ── blocks (SO-39) ────────────────────────────────────────────────────────────────────────────
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/blocks", permission=BLOCK, if_match="version")
def set_order_block(caller: dict, sales_order_id: Ref, kind: Literal["delivery", "billing"], reason_code: Code, note: Reason,
version: Optional[int] = None) -> dict:
"""Block the delivery or the billing of a confirmed order, with a reason code of that kind and a note. A block is an appended row (who, why):
inventory still reserves, but the delivery does not reach the pick list (the fulfilment answer says delivery_blocked) and the invoice
request is refused (gate billing_block) until the release."""
def unit():
with uc.repository() as repo, uc._decide():
o, current = uc._load(repo, caller, sales_order_id)
uc._same_version(f"sales order {sales_order_id}", current, version)
reason = next((r for r in repo.config.block_reasons() if r.block_reason_code == reason_code), None)
if reason is None or reason.kind not in (kind, "any"):
raise Refused(422, f"{reason_code} is not a {kind} block reason: the reasons are "
f"{', '.join(sorted(r.block_reason_code for r in repo.config.block_reasons() if r.kind in (kind, 'any')))}")
if reason_code == "APPROVAL":
raise Refused(422, "APPROVAL is the order type's own block: it starts on confirmation and is released by its approval")
o.set_block(kind, reason_code, note, by_email=(caller.get("email") or "").lower(), by_sub=(caller.get("sub") or "").lower(), at=uc._bnow())
return uc._saved(repo, caller, o, current)
return uc._retrying(version, unit)
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/blocks/release", permission=BLOCK, if_match="version")
def release_order_block(caller: dict, sales_order_id: Ref, kind: Literal["delivery", "billing"], note: Reason, version: Optional[int] = None) -> dict:
"""Release the delivery or billing block of an order, with a note: a release row is appended (who, why); the block stays in the history.
An APPROVAL block is released only by its approval, never here."""
def unit():
with uc.repository() as repo, uc._decide():
o, current = uc._load(repo, caller, sales_order_id)
uc._same_version(f"sales order {sales_order_id}", current, version)
o.release_block(kind, note, by_email=(caller.get("email") or "").lower(), by_sub=(caller.get("sub") or "").lower(), at=uc._bnow())
return uc._saved(repo, caller, o, current)
return uc._retrying(version, unit)
# ── the fulfiller's view and the delivery rules (SO-37) ───────────────────────────────────────
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/fulfilment-rules", permission=WRITE, if_match="version")
def set_fulfilment_rules(caller: dict, sales_order_id: Ref, partial_delivery: Optional[Literal["allowed", "complete_line", "complete_order"]] = None,
max_partials: Optional[Annotated[int, Field(ge=0, le=99)]] = None, under_tolerance_pct: Optional[Pct] = None,
over_tolerance_pct: Optional[Pct] = None, combine_allowed: Optional[bool] = None, version: Optional[int] = None) -> dict:
"""Change how THIS confirmed order may be delivered (part deliveries, at most how many per line, under- and over-delivery tolerance,
combined delivery), copied from the customer's policy at confirmation. Allowed only until the first delivery."""
def unit():
with uc.repository() as repo, uc._decide():
o, current = uc._load(repo, caller, sales_order_id)
uc._same_version(f"sales order {sales_order_id}", current, version)
if o.lifecycle != "confirmed":
raise RuleBroken(f"the order is {o.lifecycle}: its delivery rules are changed while it is confirmed")
if o.deliveries:
raise RuleBroken("something is already delivered: the delivery rules are fixed from the first delivery")
if partial_delivery is not None:
o.partial_delivery = partial_delivery
if max_partials is not None:
o.max_partials = max_partials
if under_tolerance_pct is not None:
o.under_tolerance_pct = under_tolerance_pct
if over_tolerance_pct is not None:
o.over_tolerance_pct = over_tolerance_pct
if combine_allowed is not None:
o.combine_allowed = combine_allowed
return uc._saved(repo, caller, o, current)
return uc._retrying(version, unit)
@SO.use_case(kind="query", route="GET /sales-orders/{sales_order_id}/fulfilment", permission=READ)
def get_fulfilment(caller: dict, sales_order_id: Ref) -> dict:
"""The fulfiller's view of an order, for inventory when it raises and when it validates a delivery (never a price, a condition or a value):
the order type, each line's item category and delivering warehouse, the priority, the ship-to, the delivery rules copied from the customer's
policy at confirmation (part deliveries, at most how many, tolerances, combined delivery), the latest schedule per line, and
delivery_blocked {met, reasons: [{gate, reason, fix}]}: a delivery block or an incomplete:* delivery gate holds the delivery off the pick list."""
with uc.repository() as repo:
o, _ = uc._load(repo, caller, sales_order_id)
today = uc._today(repo)
gates = [g for g in uc._delivery_gates(repo, o, today) if not g.met]
otype = repo.config.order_type(o.order_type_code, today)
wh = repo.settings.value_on("default_warehouse_id", today) or ""
purpose = "free_of_charge" if o.order_type_code == "FREE_OF_CHARGE" else "consignment" if o.order_type_code.startswith("CONS_") else "sale"
return uc._plain({
"sales_order_id": o.sales_order_id, "order_type_code": o.order_type_code, "priority": o.priority,
"lines": [{"line_no": ln.line_no, "product_id": ln.product_id, "item_category_code": ln.item_category_code, "unit_code": ln.unit_code,
"quantity": ln.quantity, "delivered": o.delivered(ln.line_no), "closed": o.closed(ln), "warehouse_id": wh} for ln in o.lines],
"ship_to_bp_id": o.ship_to_bp_id or o.bp_id, "ship_to_address": uc._address_of(o),
"partial_delivery": o.partial_delivery, "max_partials": o.max_partials, "under_tolerance_pct": o.under_tolerance_pct,
"over_tolerance_pct": o.over_tolerance_pct, "combine_allowed": o.combine_allowed,
"special_stock": otype.special_stock if otype else "none", "purpose": purpose, "schedule": [],
"delivery_blocked": {"met": not gates, "reasons": [{"gate": g.gate, "reason": g.reason, "fix": g.fix} for g in gates]},
"min_remaining_shelf_life_days": o.min_shelf_life_days})
# ── the partners of an order (SO-40) ──────────────────────────────────────────────────────────
@SO.use_case(kind="command", route="POST /sales-orders/{sales_order_id}/partners", permission=WRITE, if_match="version")
def change_order_partners(caller: dict, sales_order_id: Ref, ship_to_bp_id: Optional[BigId] = None, bill_to_bp_id: Optional[BigId] = None,
payer_bp_id: Optional[BigId] = None, version: Optional[int] = None) -> dict:
"""Change an order's ship-to, bill-to or payer to another of the customer's partners (one in force today). Allowed on a draft, and on a
confirmed order only before the first delivery (gate partners_unchanged_after_delivery); a ship-to in another country is a new order, since
its tax was frozen at confirmation."""
def unit():
with uc.repository() as repo, uc._decide():
o, current = uc._load(repo, caller, sales_order_id)
uc._same_version(f"sales order {sales_order_id}", current, version)
if o.lifecycle not in ("draft", "confirmed"):
raise RuleBroken(f"the order is {o.lifecycle}: its partners no longer change")
if o.deliveries:
from brick_sales_order.domain.errors import GateUnmet
from brick_sales_order.domain.readiness import no
raise GateUnmet([no("partners_unchanged_after_delivery", "something is already delivered: the partners are fixed from the first delivery",
"raise a new order, or credit and re-bill")])
today, rows = uc._today(repo), repo.config.partners()
payer_before = o.payer
for fn, asked, attr in (("ship_to", ship_to_bp_id, "ship_to_bp_id"), ("bill_to", bill_to_bp_id, "bill_to_bp_id"), ("payer", payer_bp_id, "payer_bp_id")):
if asked:
got, why = rl.resolve_partner(rows, o.bp_id, fn, asked, today)
if why:
raise Refused(422, why)
setattr(o, attr, got)
if o.lifecycle == "confirmed" and o.payer != payer_before: # confirmation checked the payer; so does changing it afterwards
from brick_sales_order.domain.errors import GateUnmet
from brick_sales_order.domain.readiness import no, require
if any(r.status != "cancelled" for r in o.invoice_requests):
raise GateUnmet([no("payer_unchanged_after_invoice", "an invoice or a deposit was already requested from the payer",
"credit and re-bill, or raise a new order")])
require(uc._payer_gates(repo, caller, o, today))
return uc._saved(repo, caller, o, current)
return uc._retrying(version, unit)