dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/deliveries.py
"""Fulfil customer orders: a delivery is raised from a confirmed sales order, picked from the bins, validated (INV-07, INV-08,
INV-23, INV-35, INV-39). The sales order learns of a delivery only from here, through its own route; nothing is written here
if it refuses; two validations for the last unit: one wins the claim row, the other is refused and writes nothing."""
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import docs, ledger, postings, products, reservations
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import need_read, qstr, shaped
from brick_inventory.application.inputs import Code, PickIn, Reason, Ref
from brick_inventory.domain import location as L
from brick_inventory.domain import removal
from brick_inventory.domain.errors import BadInput, GateBroken, RuleBroken, ShortOf
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import NO_LOT, OURS, UNRESTRICTED, ZERO, fraction
from brick_inventory.infrastructure import seams
DEFAULT_FULFILMENT = {"order_type_code": None, "priority": None, "partial_delivery": "allowed", "max_partials": None,
"under_tolerance_pct": 0, "over_tolerance_pct": 0, "combine_allowed": True, "special_stock": "none",
"purpose": "sale", "delivery_blocked": {"met": True, "reasons": []}, "min_remaining_shelf_life_days": None}
PURPOSE_ACCOUNT = {"sale": "cost_of_goods_sold", "free_of_charge": "free_of_charge", "consignment": "cost_of_goods_sold"}
def order_of(sales_order_id: str) -> dict:
if not seams.sales_installed():
raise Refused(409, "sales-order is not part of this installation: deliveries are raised by hand and nothing is reserved")
try:
return seams.sales_order(sales_order_id)
except seams.Missing as m:
if m.status == 404:
raise Refused(404, f"no sales order {sales_order_id}") from None
raise Refused(502, f"the sales order {sales_order_id} could not be read: {m.text}") from None
def fulfilment_of(sales_order_id: str) -> dict:
"""The order's fulfilment answer (blocks, partial delivery, tolerances, special stock, purpose); defaults if it offers none."""
try:
return {**DEFAULT_FULFILMENT, **seams.fulfilment(sales_order_id)}
except seams.Missing as m:
if m.status in (404, 405, 501) or not m.installed:
return dict(DEFAULT_FULFILMENT)
raise Refused(502, f"the sales order's fulfilment could not be read: {m.text}") from None
def _lines(unit, doc_id: str) -> list:
"""The document's committed lines, read once per request (a delete of its lines here forgets them)."""
key = ("inv_document_line", doc_id)
if key not in unit.memo:
unit.memo[key] = unit.repo.find("inv_document_line", order=("line_no",), document_id=doc_id)
return [dict(l) for l in unit.memo[key]]
def _demand_lines(unit, doc_id: str) -> list:
return [l for l in _lines(unit, doc_id) if l["line_kind"] == "demand"]
def _pick_lines(unit, doc_id: str) -> list:
return [l for l in _lines(unit, doc_id) if l["line_kind"] == "pick"]
def _delivery(unit, document_id: str) -> dict:
d = unit.repo.get("inv_document", document_id=document_id)
if d is None or d["kind"] != "delivery":
raise Refused(404, f"no delivery {document_id}")
return d
def _delivered_before(unit, sales_order_id: str, line_no: int) -> Decimal:
"""What our done deliveries already shipped on an order line, in the line's unit."""
total = ZERO
for d in (x for x in _order_deliveries(unit, sales_order_id) if x["lifecycle"] == "done"):
for l in _demand_lines(unit, d["document_id"]):
if l["line_no"] == line_no:
total += l["quantity_done"] or ZERO
return total
def _in_unit(base: Decimal, num: int, den: int) -> Decimal:
"""A base quantity in the order line's unit, exactly, or a refusal: record_delivery takes at most four places."""
q = fraction(base) / Fraction(num, den) * 10 ** 4
if q.denominator != 1:
raise GateBroken("exact_base", f"{qstr(base)} in the base unit is not a whole number of the order line's unit (x{num}/{den}) "
"to four places", "pick a quantity that is", 409)
return Decimal(q.numerator).scaleb(-4)
def _order_deliveries(unit, sales_order_id: str) -> list:
"""The order's committed deliveries, read once per request (what this request stages is not among them, as before)."""
key = ("inv_document:delivery", sales_order_id)
if key not in unit.memo:
unit.memo[key] = unit.repo.find("inv_document", kind="delivery", source_ref=sales_order_id)
return [dict(d) for d in unit.memo[key]]
def _next_n(unit, sales_order_id: str) -> int:
ids = [d["document_id"] for d in _order_deliveries(unit, sales_order_id)]
return len(ids) + 1
def _view(unit, d: dict, caller: dict) -> dict:
demand, picks = _demand_lines(unit, d["document_id"]), _pick_lines(unit, d["document_id"])
lines = []
for l in demand:
mine = [p for p in picks if p["parent_line_no"] == l["line_no"]]
lines.append({"line_no": l["line_no"], "product_id": l["product_id"], "unit_code": l["unit_code"],
"quantity_expected": qstr(l["quantity_expected"]), "quantity_base": qstr(l["quantity_base"]),
"quantity_done": qstr(l["quantity_done"]),
"reserved_base": qstr(reservations.held(unit, d["source_ref"], l["line_no"])) if d["lifecycle"] != "done" else None,
"picks": [{"line_no": p["line_no"], "location_id": p["from_location_id"], "lot_code": p["lot_code"] or None,
"quantity": qstr(p["quantity_done"]), "unit_code": p["unit_code"], "quantity_base": qstr(p["quantity_base"])}
for p in mine]})
view = {"document_id": d["document_id"], "number": d["number"], "sales_order_id": d["source_ref"], "bp_id": d["bp_id"],
"warehouse_id": d["warehouse_id"], "lifecycle": d["lifecycle"], "purpose": d["purpose"], "back_order_of": d["back_order_of"],
"pick_batch_id": d["pick_batch_id"], "reason": d["reason"], "lines": lines,
"delivery_ref": f"inv:{d['number']}" if d["number"] else None, "readiness": _readiness(unit, d, demand, picks)}
return view
def _readiness(unit, d: dict, demand: list, picks: list) -> list:
out = [{"gate": "something_picked", "met": bool(picks), "reason": "" if picks else "nothing is picked yet",
"fix": "" if picks else "pick from the bins"}]
if unit.settings.flag("delivery_needs_weights", unit.today):
unknown = [p for p in picks if p["gross_weight"] is None and p["net_weight"] is None]
out.append({"gate": "weights_known", "met": not unknown,
"reason": "" if not unknown else f"line {unknown[0]['line_no']} ({unknown[0]['product_id']}) has no weight",
"fix": "" if not unknown else "give the unit its weight in the product catalogue"})
if unit.settings.value("delivery_steps", unit.today) == "pick_pack_ship":
packed = _packed_base(unit, d["document_id"])
loose = [p for p in picks if packed.get((p["product_id"], p["lot_code"] or ""), ZERO) < p["quantity_base"]]
out.append({"gate": "all_packed", "met": not loose, "reason": "" if not loose else f"{loose[0]['product_id']} is not in any package",
"fix": "" if not loose else "pack every picked item"})
return out
def _packed_base(unit, document_id: str) -> dict:
import json
packed: dict = {}
for p in unit.repo.find("inv_package", document_id=document_id):
for c in json.loads(p["contents"] or "[]"):
k = (c["product_id"], c.get("lot_code") or "")
packed[k] = packed.get(k, ZERO) + Decimal(c["quantity_base"])
return packed
@INV.use_case(kind="command", route="POST /deliveries", permission="inv:delivery:validate")
def create_delivery(caller: dict, sales_order_id: Ref, warehouse_id: Optional[Code] = None) -> dict:
"""Raise the delivery of a confirmed sales order (read through the order's own route): one demand line per goods line for what
is not yet delivered, in the line's own unit; stock already reserved for the order is shown. A delivery block on the order
keeps it waiting. The order's special stock and purpose decide where the goods go and which account carries the cost."""
@U.decide
def body(unit):
order = order_of(sales_order_id)
if order.get("lifecycle") not in ("confirmed", "partly_delivered", "delivered_partly", "open"):
raise RuleBroken(f"order {sales_order_id} is {order.get('lifecycle')}: only a confirmed order is delivered")
ful = fulfilment_of(sales_order_id)
wh = reservations.default_warehouse(unit, warehouse_id)
open_ = [d for d in _order_deliveries(unit, sales_order_id) if d["lifecycle"] in ("waiting", "ready")]
if open_:
raise RuleBroken(f"{open_[0]['document_id']} is already open for {sales_order_id}: use it")
n = _next_n(unit, sales_order_id)
doc_id = f"dlv:so:{sales_order_id}:{n}"
purpose = {"consignment": "consignment_fill" if ful["special_stock"] == "to_consignment" else "consignment_issue"}.get(
ful.get("purpose"), ful.get("purpose") or "sale")
blocked = not (ful["delivery_blocked"] or {}).get("met", True)
lines = []
for l in order.get("lines", []):
info = unit.product(l["product_id"])
if info.get("kind") == "service":
continue
num, den = int(l.get("unit_numerator") or 1), int(l.get("unit_denominator") or 1)
ordered = Decimal(str(l["quantity"]))
remaining = ordered - _delivered_before(unit, sales_order_id, l["line_no"])
if remaining <= 0:
continue
base = fraction(remaining) * Fraction(num, den)
lines.append((l, num, den, remaining, Decimal(base.numerator) / Decimal(base.denominator)))
if not lines:
raise RuleBroken(f"order {sales_order_id} has nothing left to deliver")
state = "waiting" if blocked else _initial_state(unit, sales_order_id, lines, ful)
reason = "; ".join(r.get("reason", "") for r in (ful["delivery_blocked"] or {}).get("reasons", [])) if blocked else None
docs.stage_document(unit, doc_id, "delivery", wh["warehouse_id"], wh["company_code"], stored=False, bp_id=ful.get("ship_to_bp_id") or order.get("bp_id"),
source_kind="sales_order", source_ref=sales_order_id, purpose=purpose, lifecycle=state, reason=reason)
for l, num, den, remaining, base in lines:
docs.stage_line(unit, doc_id, l["line_no"], stored=False, line_kind="demand", product_id=l["product_id"], unit_code=l["unit_code"],
unit_numerator=num, unit_denominator=den, quantity_expected=remaining, quantity_base=base,
source_line_ref=str(l["line_no"]))
return {"document_id": doc_id, "sales_order_id": sales_order_id, "lifecycle": state, "purpose": purpose,
"blocked": reason, "lines": [{"line_no": l["line_no"], "product_id": l["product_id"], "unit_code": l["unit_code"],
"quantity": qstr(rem), "quantity_base": qstr(b),
"reserved_base": qstr(reservations.held(unit, sales_order_id, l["line_no"]))}
for l, _n, _d, rem, b in lines]}
return U.run(caller, body)
def partial_allowed(ful: dict) -> bool:
"""sales-order's partial_delivery is allowed | complete_line | complete_order (SO-37); a string is always truthy, so
complete_order read as 'allowed' and a part-reserved order shipped in parts."""
v = ful.get("partial_delivery", "allowed")
return v is True or v == "allowed"
def _initial_state(unit, sales_order_id: str, lines: list, ful: dict) -> str:
"""stock_arrives: ready once every line is reserved in full, or the order accepts part shipments (the setting)."""
if unit.settings.flag("part_shipments", unit.today) and partial_allowed(ful):
return "ready"
full = all(reservations.held(unit, sales_order_id, l["line_no"]) >= b for l, _n, _d, _r, b in lines)
return "ready" if full else "waiting"
@INV.use_case(kind="query", route="GET /deliveries/{document_id}")
def get_delivery(caller: dict, document_id: Ref) -> dict:
"""A delivery with its demand lines, what is reserved, what is picked, and its readiness gates {gate, met, reason, fix}."""
need_read(caller)
with U.read() as unit:
d = _delivery(unit, document_id)
if d["lifecycle"] == "waiting":
d = _refresh_waiting(unit, d)
return shaped(caller, _view(unit, d, caller))
def _refresh_waiting(unit, d: dict) -> dict:
return d
@INV.use_case(kind="query", route="GET /deliveries")
def list_deliveries(caller: dict, lifecycle: Optional[Code] = None, warehouse_id: Optional[Code] = None) -> dict:
"""Deliveries, newest first: waiting, ready, done or cancelled; of one warehouse or all."""
need_read(caller)
with U.read() as unit:
where = {k: v for k, v in (("kind", "delivery"), ("lifecycle", lifecycle), ("warehouse_id", warehouse_id)) if v}
rows = unit.repo.find("inv_document", order=("-document_id",), limit=200, **where)
return {"deliveries": [{"document_id": d["document_id"], "number": d["number"], "sales_order_id": d["source_ref"],
"lifecycle": d["lifecycle"], "warehouse_id": d["warehouse_id"], "bp_id": d["bp_id"]} for d in rows]}
def candidates(unit, product_id: str, warehouse_id: str) -> list:
rows = ledger.stock_rows(unit, product_id, warehouse_id=warehouse_id)
lots = {r["lot_code"]: r for r in unit.repo.find("inv_lot", product_id=product_id)}
out = []
for r in rows:
if r.kind != "internal" or r.quantity_base <= 0:
continue
loc = unit.location(r.location_id)
lot = lots.get(r.lot_code) or {}
out.append(removal.Candidate(r.location_id, r.lot_code, r.stock_type, r.owner, r.quantity_base, r.expires_on,
lot.get("received_on"), loc.get("pick_sequence") or 0))
return out
@INV.use_case(kind="query", route="GET /deliveries/{document_id}/pick-list")
def get_pick_list(caller: dict, document_id: Ref) -> dict:
"""What to pick, bin by bin in walking order (pick sequence), naming the lot by the removal strategy (FEFO or FIFO, with the
customer's minimum remaining shelf life), proposed in the warehouse unit first (5 PK24 and 6 C62, not 94.5 litres). An
expired lot is never proposed."""
need_read(caller)
with U.read() as unit:
d = _delivery(unit, document_id)
ful = fulfilment_of(d["source_ref"]) if d["source_kind"] == "sales_order" else dict(DEFAULT_FULFILMENT)
return {"document_id": document_id, "lines": pick_proposal(unit, d, ful)}
def pick_proposal(unit, d: dict, ful: dict) -> list:
out = []
for l in _demand_lines(unit, d["document_id"]):
info = unit.product(l["product_id"])
strategy = unit.tracking(l["product_id"])["removal_strategy"]
got, short = removal.propose(candidates(unit, l["product_id"], d["warehouse_id"]), l["quantity_base"], strategy, unit.today,
ful.get("min_remaining_shelf_life_days"), unit.settings.flag("allow_expired_pick", unit.today))
from brick_inventory.application.stock import display
wu = products.warehouse_unit(info)
out.append({"line_no": l["line_no"], "product_id": l["product_id"], "need_base": qstr(l["quantity_base"]),
"order_unit": l["unit_code"], "warehouse_unit": wu, "removal_strategy": strategy,
"warehouse_unit_set": bool((info.get("roles") or {}).get("warehouse")),
"proposals": [{"location_id": c.location_id, "lot_code": c.lot_code or None, "quantity_base": qstr(q),
"expires_on": c.expires_on.isoformat() if c.expires_on else None, "pick_sequence": c.pick_sequence,
"shown": display(info, q)} for c, q in got],
"short_base": qstr(short)})
return out
def _check_pick_stock(unit, d: dict, ln: dict, ful: dict, pick: PickIn, base: Decimal, override: Optional[str]) -> None:
loc = unit.location(pick.location_id)
at_customer = loc["kind"] == "consignment_customer" and ful.get("special_stock") == "from_consignment" and str(loc.get("bp_id")) == str(d["bp_id"])
if not at_customer and (loc["kind"] != "internal" or loc["warehouse_id"] != d["warehouse_id"]):
raise RuleBroken(f"{pick.location_id} is not a bin of warehouse {d['warehouse_id']}")
if not at_customer and ledger.is_goods_in(unit, loc):
raise RuleBroken(f"{pick.location_id} is goods-in: put it away first, then pick it from its bin")
stype = pick.stock_type or UNRESTRICTED
if stype != UNRESTRICTED:
raise RuleBroken(f"{pick.location_id} stock is in {stype}: release it first")
if loc["stock_status"] != UNRESTRICTED:
raise RuleBroken(f"{pick.location_id} is in {'quality inspection' if loc['stock_status'] == 'quality' else 'blocked status'}: release it first")
tr = unit.tracking(ln["product_id"])
lot = pick.lot_code or NO_LOT
if tr["tracking"] != "none" and not lot:
raise RuleBroken(f"{ln['product_id']} is tracked by {tr['tracking']}: name the {'serial number' if tr['tracking'] == 'serial' else 'lot'}")
if lot:
row = unit.repo.get("inv_lot", product_id=ln["product_id"], lot_code=lot) or {}
exp = row.get("expires_on")
if exp is not None and exp <= unit.today and not (unit.settings.flag("allow_expired_pick", unit.today) and override):
raise GateBroken("shelf_life", f"{lot} expired on {exp.isoformat()}: it is never picked", "scrap it, or ask the owner to allow expired picks", 409)
need = ful.get("min_remaining_shelf_life_days")
if need and exp is not None and (exp - unit.today).days < int(need) and not override:
raise GateBroken("shelf_life", f"{lot} expires {exp.isoformat()}: {d['bp_id']} needs {need} days", "pick another lot, or a warehouse manager allows it with a reason", 409)
@INV.use_case(kind="command", route="POST /deliveries/{document_id}/pick", permission="inv:delivery:validate")
def pick_goods(caller: dict, document_id: Ref, picks: list[PickIn], override_reason: Optional[Reason] = None) -> dict:
"""Record what was picked from which bin (and lot or serial), in any unit of the product. The picks listed for a line replace
its earlier ones. Each bin must hold what is picked from it (unless negative stock is allowed there) and be unrestricted
stock: a quality or blocked bin is refused ("WH1/QC is in quality inspection: release it first"). A lot with less shelf life
left than the customer accepts is refused (gate shelf_life); a warehouse manager may allow it with an override_reason, which
is recorded."""
@U.decide
def body(unit):
d = _delivery(unit, document_id)
if d["lifecycle"] not in ("waiting", "ready"):
raise RuleBroken(f"{document_id} is {d['lifecycle']}: nothing more is picked on it")
override = override_reason if (override_reason and U.can(caller, "inv:reservation:write")) else None
if override_reason and not override:
raise Refused(403, "missing permission: inv:reservation:write (only a warehouse manager overrides the shelf-life rule)")
ful = fulfilment_of(d["source_ref"]) if d["source_kind"] == "sales_order" else dict(DEFAULT_FULFILMENT)
rows = _record_picks(unit, d, picks, ful, override)
return {"document_id": document_id,
"picks": [{"line_no": r["parent_line_no"], "location_id": r["from_location_id"], "lot_code": r["lot_code"] or None,
"quantity": qstr(r["quantity_done"]), "unit_code": r["unit_code"], "quantity_base": qstr(r["quantity_base"])}
for r in rows],
"readiness": _readiness(unit, d, _demand_lines(unit, document_id), rows)}
return U.run(caller, body)
def _record_picks(unit, d: dict, picks: list, ful: dict, override: Optional[str]) -> list:
demand = {l["line_no"]: l for l in _demand_lines(unit, d["document_id"])}
existing = _pick_lines(unit, d["document_id"])
next_no = max([l["line_no"] for l in existing] + [1000]) + 1
touched = {p.line_no for p in picks}
for ln_no in touched:
if ln_no not in demand:
raise RuleBroken(f"{d['document_id']} has no line {ln_no}")
if touched: # the touched lines' earlier picks go, in one statement
unit.repo.delete_where("inv_document_line", document_id=d["document_id"], line_kind="pick", parent_line_no=sorted(touched))
unit.memo.pop(("inv_document_line", d["document_id"]), None)
planned: dict = {}
out = []
for p in picks:
ln = demand[p.line_no]
conv = products.convert(unit, ln["product_id"], p.quantity, p.unit_code, document_ref=d["document_id"], what="the picked quantity")
_in_unit(conv.base, ln["unit_numerator"], ln["unit_denominator"])
k = (p.location_id, p.lot_code or "", ln["product_id"])
_check_pick_stock(unit, d, ln, ful, p, conv.base, override)
planned[k] = planned.get(k, ZERO) + conv.base
key = ledger.stock_rules.key(ln["product_id"], p.location_id, p.lot_code or "", p.stock_type or UNRESTRICTED, OURS)
have = unit.stock(key)
key_c = ledger.stock_rules.key(ln["product_id"], p.location_id, p.lot_code or "", UNRESTRICTED, "")
_ = key_c
if have < planned[k] and not unit.settings.negative_allowed(d["warehouse_id"], unit.today) and not _owned_by_supplier(unit, ln, p, planned[k]):
raise RuleBroken(f"{p.location_id} holds {qstr(max(have, ZERO))} {unit.base_unit(ln['product_id'])} of {ln['product_id']}"
f"{' lot ' + p.lot_code if p.lot_code else ''}, {qstr(planned[k])} picked")
info = unit.product(ln["product_id"])
row = docs.stage_line(unit, d["document_id"], next_no, stored=False, line_kind="pick", parent_line_no=p.line_no, product_id=ln["product_id"],
unit_code=p.unit_code, unit_numerator=conv.numerator, unit_denominator=conv.denominator,
quantity_done=conv.quantity, quantity_base=conv.base, from_location_id=p.location_id,
lot_code=p.lot_code or NO_LOT, stock_type=p.stock_type or UNRESTRICTED, note=override, **products.measures(info, p.unit_code))
out.append(row)
next_no += 1
return out
def _check_not_reserved_for_others(unit, d: dict, demand: dict, pick_rows: list) -> None:
"""What a delivery takes beyond its own order's reservation must be unreserved stock (SC-INV-02): usable less ALL reservations."""
if unit.settings.negative_allowed(d["warehouse_id"], unit.today):
return
for p in pick_rows: # a bin that does not hold its pick is refused by the move itself, naming the bin
if unit.stock(ledger.stock_rules.key(p["product_id"], p["from_location_id"], p["lot_code"] or "", p["stock_type"] or UNRESTRICTED, OURS)) < p["quantity_base"]:
return
excess: dict = {}
for line_no in sorted({p["parent_line_no"] for p in pick_rows}):
ln = demand[line_no]
picked = sum((p["quantity_base"] for p in pick_rows if p["parent_line_no"] == line_no), ZERO)
own = reservations.held(unit, d["source_ref"], line_no) if d["source_kind"] == "sales_order" else ZERO
excess[ln["product_id"]] = excess.get(ln["product_id"], ZERO) + max(picked - max(own, ZERO), ZERO)
for product_id, need in excess.items():
if need <= 0:
continue
free = reservations.available_at(unit, product_id, d["warehouse_id"])
usable = sum((r.quantity_base for r in ledger.stock_rows(unit, product_id, warehouse_id=d["warehouse_id"]) if r.usable(unit.today)), ZERO)
if need > max(free, ZERO) and usable >= need: # a plain shortage is the move's to refuse; this is stock held for others
reserved = unit.repo.reserved(d["warehouse_id"], product_id)
raise ShortOf(f"{qstr(need)} {unit.base_unit(product_id)} of {product_id} beyond this order's reservation, only "
f"{qstr(max(free, ZERO))} is unreserved at {d['warehouse_id']}: {qstr(reserved)} is reserved for other orders",
d["warehouse_id"], free)
def _owned_by_supplier(unit, ln, p, qty) -> bool:
"""Supplier-consignment stock in the bin (we may use it): enough of it counts as what the bin holds."""
from brick_inventory.domain.values import OURS as _o
rows = unit.repo.stock(product_id=ln["product_id"], locations={p.location_id}, lot_code=p.lot_code or "")
total = sum((v for k, v in rows.items() if k[3] == UNRESTRICTED), ZERO)
_ = _o
return total >= qty and any(k[4] != "" and v > 0 for k, v in rows.items())
@INV.use_case(kind="command", route="POST /deliveries/{document_id}/validate", permission="inv:delivery:validate", idempotent=True)
def validate_delivery(caller: dict, document_id: Ref, picks: Optional[list[PickIn]] = None, override_reason: Optional[Reason] = None) -> dict:
"""Validate the delivery: the goods leave their bins for the customer at the average (or standard) cost, the reservation is
consumed, the sales order records the delivery (record_delivery, keyed inv:<number>), the cost of goods is posted, and what
is left goes to a back-order delivery. Picks given here are recorded first. Two validations for the last unit: one wins the
claim row, the other is refused and writes nothing - not the move, not the call to the sales order. A refusal by the order
refuses the whole validation. Validating a done delivery returns the same answer."""
@U.decide
def body(unit):
d = _delivery(unit, document_id)
if d["lifecycle"] == "done":
return {**_view(unit, d, caller), "replay": True}
if d["lifecycle"] == "cancelled":
raise RuleBroken(f"{document_id} was cancelled")
override = override_reason if (override_reason and U.can(caller, "inv:reservation:write")) else None
ful = fulfilment_of(d["source_ref"]) if d["source_kind"] == "sales_order" else dict(DEFAULT_FULFILMENT)
blocked = not (ful["delivery_blocked"] or {}).get("met", True)
if blocked:
reasons = (ful["delivery_blocked"] or {}).get("reasons", [{}])
first = reasons[0] if reasons else {}
raise GateBroken("not_blocked", first.get("reason") or "the order has a delivery block", first.get("fix") or "clear the block on the order", 409)
if picks:
pick_rows = _record_picks(unit, d, picks, ful, override)
else:
pick_rows = _pick_lines(unit, document_id)
if not pick_rows:
raise GateBroken("something_picked", "nothing is picked yet", "pick from the bins", 409)
demand = {l["line_no"]: l for l in _demand_lines(unit, document_id)}
for g in _readiness_rows(unit, d, pick_rows):
if not g["met"]:
raise GateBroken(g["gate"], g["reason"], g["fix"], 409)
wh = unit.warehouse(d["warehouse_id"])
company = wh["company_code"]
special = ful.get("special_stock", "none")
if special != "from_consignment":
_check_not_reserved_for_others(unit, d, demand, pick_rows)
number = docs.claim_number(unit, d["warehouse_id"], "delivery", document_id)
delivery_ref = f"inv:{number}"
purpose = d["purpose"] or "sale"
postings_value = ZERO
posted = []
by_line: dict = {}
for i, p in enumerate(sorted(pick_rows, key=lambda r: r["line_no"]), start=1):
ln = demand[p["parent_line_no"]]
customer = L.virtual_id(company, "CUSTOMER")
target = customer
source = p["from_location_id"]
if special == "to_consignment":
from brick_inventory.application.warehouses import special_location
target = special_location(unit, company, "consignment_customer", int(d["bp_id"]))
elif special == "from_consignment":
from brick_inventory.application.warehouses import special_location
source = special_location(unit, company, "consignment_customer", int(d["bp_id"]))
owner_from = OURS
spec = MoveSpec(document_id, p["line_no"], p["product_id"], p["quantity_base"], source, target, unit.today, p["unit_code"],
p["quantity_done"], p["unit_numerator"], p["unit_denominator"], p["lot_code"] or NO_LOT, p["stock_type"] or UNRESTRICTED,
UNRESTRICTED, owner_from, OURS, n=1, reason=override or "")
withdrawn = _withdraw_consignment(unit, d, p, spec) if source == p["from_location_id"] else None
pm = ledger.post_move(unit, spec)
posted.append((p, pm))
by_line[ln["line_no"]] = by_line.get(ln["line_no"], ZERO) + p["quantity_base"]
postings_value += pm.row["value_amount"]
_ = withdrawn
unit.flush_races() # the claims are won: only now is anyone told
lines_out = []
recorded = []
if special != "to_consignment" and seams.sales_installed():
for line_no, base in sorted(by_line.items()):
ln = demand[line_no]
qty = _in_unit(base, ln["unit_numerator"], ln["unit_denominator"])
body_ = {"line_no": line_no, "quantity": qstr(qty), "source": "inventory", "delivery_ref": delivery_ref,
"delivered_on": unit.today.isoformat()}
try:
seams.record_delivery(d["source_ref"], body_)
except seams.Missing as m:
raise GateBroken("order_accepts", f"the sales order refused the delivery: {m.text}",
"deliver no more than the order says, and only while it is confirmed", 409) from None
recorded.append((line_no, qty))
lines_out.append({"line_no": line_no, "quantity": qstr(qty), "unit_code": ln["unit_code"]})
unit.compensations.append(lambda rec=list(recorded), so=d["source_ref"], ref=delivery_ref, today=unit.today: _undo_order(so, ref, rec, today))
for line_no, base in by_line.items():
reservations_consume(unit, d["source_ref"], line_no, base, wh["warehouse_id"], demand[line_no]["product_id"])
for line_no, base in by_line.items():
ln = demand[line_no]
qty = _in_unit(base, ln["unit_numerator"], ln["unit_denominator"])
docs.stage_line(unit, document_id, line_no, stored=True, quantity_done=qty) # the demand lines this request read
docs.stage_document(unit, document_id, "delivery", d["warehouse_id"], company, stored=True, number=number, lifecycle="done", done_on=unit.today,
done_by=U.who(caller), source_kind=d["source_kind"], source_ref=d["source_ref"], bp_id=d["bp_id"], purpose=purpose)
back = _back_order(unit, d, demand, by_line, ful, wh, special)
if special != "to_consignment" and postings_value:
account = PURPOSE_ACCOUNT.get(purpose if purpose in PURPOSE_ACCOUNT else "sale", "cost_of_goods_sold")
by_class: dict = {}
for p, pm in posted:
vc = unit.product(p["product_id"]).get("valuation_class") or "*"
by_class[vc] = by_class.get(vc, ZERO) + pm.row["value_amount"]
lines = []
for vc, v in by_class.items():
lines += postings.movement_lines(v, account, vc, memo=f"delivery {number}")
postings.stage(unit, document_id, "delivery", company, unit.today, f"delivery {number}", _merge(lines))
view = {"document_id": document_id, "number": number, "delivery_ref": delivery_ref, "lifecycle": "done",
"sales_order_id": d["source_ref"], "lines": lines_out, "back_order": back, "replay": False,
"moves": [{"move_id": pm.row["move_id"], "from_location_id": pm.row["from_location_id"],
"to_location_id": pm.row["to_location_id"], "quantity_base": qstr(pm.row["quantity_base"]),
"lot_code": pm.row["lot_code"] or None, "value_amount": f"{pm.row['value_amount']:.2f}"} for _p, pm in posted]}
return shaped(caller, view)
return U.run(caller, body, lost_message=_lost_message(document_id))
def _lost_message(document_id: str):
def message(repo, row=None) -> str:
if row and row.get("table") != "x" and row.get("stock_type"):
have = repo.stock_one((row["product_id"], row["location_id"], row["lot_code"], row["stock_type"], row["owner"]))
unit = U.Unit(repo, {})
return (f"{row['location_id']} holds {qstr(max(have, ZERO))} {unit.base_unit(row['product_id'])} of {row['product_id']} now: "
"another delivery took it")
return "another delivery took the stock at the same moment"
return message
def _merge(lines: list) -> list:
"""Journal lines of one purpose and side are one line (no two cost-of-goods lines for the same delivery)."""
out: dict = {}
for l in lines:
if not l:
continue
k = (l["purpose"], l["side"], l["valuation_class"], l["cost_centre"])
if k in out:
out[k] = {**out[k], "amount": f"{Decimal(out[k]['amount']) + Decimal(l['amount']):.2f}"}
else:
out[k] = l
return list(out.values())
def _readiness_rows(unit, d, picks):
return _readiness(unit, d, _demand_lines(unit, d["document_id"]), picks)
def _withdraw_consignment(unit, d: dict, p: dict, spec: MoveSpec):
"""Supplier-consignment stock picked is withdrawn into our ownership first (slice 2: see consignment.py)."""
from brick_inventory.application import consignment
return consignment.withdraw_if_needed(unit, d, p, spec)
def reservations_consume(unit, sales_order_id: str, line_no: int, base: Decimal, warehouse_id: str, product_id: str) -> None:
have = reservations.held(unit, sales_order_id, line_no)
take = min(have, base)
if take > 0:
reservations.add_row(unit, sales_order_id, line_no, "consume", warehouse_id, product_id, -take, reason="delivered")
def _undo_order(sales_order_id: str, delivery_ref: str, recorded: list, today: date) -> None:
for line_no, qty in recorded:
try:
seams.record_delivery(sales_order_id, {"line_no": line_no, "quantity": qstr(-qty), "source": "inventory",
"delivery_ref": f"{delivery_ref}:undo", "delivered_on": today.isoformat(),
"corrects_ref": delivery_ref})
except seams.Missing:
pass
def _back_order(unit, d: dict, demand: dict, by_line: dict, ful: dict, wh: dict, special: str) -> Optional[dict]:
"""What is left goes to a back-order delivery, unless the line closed within the order's tolerance."""
left = []
for line_no, ln in demand.items():
done = by_line.get(line_no, ZERO)
remaining = ln["quantity_base"] - done
if remaining <= 0:
continue
pct = (remaining / ln["quantity_base"] * 100) if ln["quantity_base"] else ZERO
if pct <= Decimal(str(ful.get("under_tolerance_pct") or 0)):
continue
left.append((ln, remaining))
if not left:
return None
n = _next_n(unit, d["source_ref"])
bo = f"dlv:so:{d['source_ref']}:{n}"
docs.stage_document(unit, bo, "delivery", d["warehouse_id"], wh["company_code"], stored=False, bp_id=d["bp_id"], source_kind="sales_order",
source_ref=d["source_ref"], purpose=d["purpose"], lifecycle="ready", back_order_of=d["document_id"])
out = []
for ln, remaining in left:
qty = _in_unit(remaining, ln["unit_numerator"], ln["unit_denominator"])
docs.stage_line(unit, bo, ln["line_no"], stored=False, line_kind="demand", product_id=ln["product_id"], unit_code=ln["unit_code"],
unit_numerator=ln["unit_numerator"], unit_denominator=ln["unit_denominator"], quantity_expected=qty,
quantity_base=remaining, source_line_ref=ln["source_line_ref"])
out.append({"line_no": ln["line_no"], "quantity": qstr(qty), "unit_code": ln["unit_code"]})
return {"document_id": bo, "lines": out}
@INV.use_case(kind="command", route="POST /deliveries/{document_id}/cancel", permission="inv:delivery:validate", idempotent=True)
def cancel_delivery(caller: dict, document_id: Ref, reason: Reason) -> dict:
"""Cancel a delivery nothing of which was validated, with a reason: its reservations are released. A delivery in a pick batch
leaves the batch first."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to cancel a delivery")
d = _delivery(unit, document_id)
if d["lifecycle"] == "cancelled":
return {"document_id": document_id, "lifecycle": "cancelled", "replay": True}
if d["lifecycle"] == "done":
raise RuleBroken(f"{document_id} is done: a delivery is undone only by a return receipt")
if d["pick_batch_id"]:
raise RuleBroken(f"{document_id} is in pick batch {d['pick_batch_id']}: take it out of the batch first")
for l in _demand_lines(unit, document_id):
have = reservations.held(unit, d["source_ref"], l["line_no"])
if have > 0:
reservations.add_row(unit, d["source_ref"], l["line_no"], "release", d["warehouse_id"], l["product_id"], -have,
reason=f"delivery cancelled: {reason}")
unit.repo.delete_where("inv_document_line", document_id=document_id, line_kind="pick")
unit.memo.pop(("inv_document_line", document_id), None)
docs.stage_document(unit, document_id, "delivery", d["warehouse_id"], d["company_code"], lifecycle="cancelled", reason=reason,
source_kind=d["source_kind"], source_ref=d["source_ref"])
return {"document_id": document_id, "lifecycle": "cancelled", "reason": reason, "replay": False}
return U.run(caller, body)