dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/reservations.py
"""Stock held for a sales order line (INV-22, INV-13), and what can be promised by date (INV-49).
A reservation is an append-only row per change: a reserve is positive, a release negative, a delivery consumes (negative).
Reserved per line = the sum of its rows. Every reservation claims the warehouse root with a plain INSERT, so a reservation and
an unreserved delivery collide too: reserved never exceeds what is available (unless negative stock is allowed there).
"""
from decimal import Decimal
from typing import Literal, Optional, Union
from erp_platform.bricks import Refused
from brick_inventory.application import ledger, products
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import qstr
from brick_inventory.application.inputs import Code, Reason, Ref, ReservationLineIn, Seq, SignedQuantity
from brick_inventory.domain.errors import BadInput, GateBroken, RuleBroken
from brick_inventory.domain.values import ZERO, to_base_exact
def default_warehouse(unit, warehouse_id: Optional[str]) -> dict:
if warehouse_id:
return unit.warehouse(warehouse_id)
rows = unit.repo.find("inv_warehouse", order=("warehouse_id",))
named = {r["key"].split(":", 1)[1]: r for r in unit.repo.setting_rows() if r["key"].startswith("default_warehouse:") and r["value"]}
if named and len(named) == 1:
return unit.warehouse(next(iter(named.values()))["value"])
if len(rows) == 1:
return rows[0]
raise GateBroken("warehouse", "the order names no warehouse and the company has no default one", "name a warehouse, or set default_warehouse", 422)
def held(unit, sales_order_id: str, line_no: int, warehouse_id: Optional[str] = None) -> Decimal:
return sum((r["quantity_base"] for r in unit.reservation_rows(sales_order_id, line_no)
if r["kind"] != "marker" and (warehouse_id is None or r["warehouse_id"] == warehouse_id)), ZERO)
def consumed(unit, sales_order_id: str, line_no: int) -> Decimal:
return -sum((r["quantity_base"] for r in unit.reservation_rows(sales_order_id, line_no)
if r["kind"] == "consume"), ZERO)
def available_at(unit, product_id: str, warehouse_id: str) -> Decimal:
"""Usable (unrestricted, unexpired, internal) stock at a warehouse less what is reserved (this request's plan included)."""
rows = ledger.stock_rows(unit, product_id, warehouse_id=warehouse_id)
usable = sum((r.quantity_base for r in rows if r.usable(unit.today)), ZERO)
planned = unit.reserved_delta.get((warehouse_id, product_id), ZERO)
k = ("reserved", warehouse_id, product_id)
if k not in unit.memo: # committed reservations, read once per request (the plan is reserved_delta)
unit.memo[k] = unit.repo.reserved(warehouse_id, product_id)
return usable - unit.memo[k] - planned
def add_row(unit, sales_order_id: str, line_no: int, kind: str, warehouse_id: str, product_id: str, quantity: Decimal, *,
revision=None, schedule_version=None, confirmed_on=None, priority=None, reason: str = "") -> None:
key = (sales_order_id, line_no)
last = unit.reservation_seq.get(key)
seq = (max((int(r["seq"]) for r in unit.reservation_rows(sales_order_id, line_no)), default=0) if last is None else last) + 1
unit.reservation_seq[key] = seq
unit.reserved_delta[(warehouse_id, product_id)] = unit.reserved_delta.get((warehouse_id, product_id), ZERO) + quantity
unit.race("inv_reservation", {"sales_order_id": sales_order_id, "so_line_no": line_no, "seq": seq, "kind": kind,
"warehouse_id": warehouse_id, "product_id": product_id, "quantity_base": quantity,
"revision": revision, "schedule_version": schedule_version, "confirmed_on": confirmed_on,
"priority": priority, "reason": reason, "by_email": U.who(unit.caller),
"reserved_at": unit.now_utc.replace(tzinfo=None)})
if quantity:
ledger.claim_root(unit, product_id, warehouse_id, f"{sales_order_id}:{line_no}:{seq}", abs(quantity), reservation=True)
def set_line(unit, sales_order_id: str, line_no: int, product_id: str, warehouse_id: str, wanted: Decimal, *, revision=None,
schedule_version=None, priority=None, reason: str = "") -> tuple:
"""Make the line's held quantity `wanted` (as much as is available when raising). Returns (held, short)."""
current = held(unit, sales_order_id, line_no)
if wanted < current:
add_row(unit, sales_order_id, line_no, "release", warehouse_id, product_id, wanted - current, revision=revision,
schedule_version=schedule_version, priority=priority, reason=reason or "order amended")
return wanted, ZERO
gain = wanted - current
if gain > 0:
free = available_at(unit, product_id, warehouse_id)
take = min(gain, max(free, ZERO))
if take > 0:
add_row(unit, sales_order_id, line_no, "reserve", warehouse_id, product_id, take, revision=revision,
schedule_version=schedule_version, priority=priority, reason=reason or "order confirmed")
return current + take, gain - take
return current, ZERO
def _gate(short: Decimal, held_: Decimal, unit_code: str) -> list:
if short > 0:
return [{"gate": "stock_reserved", "met": False,
"reason": f"{qstr(short)} {unit_code} of the line is not in stock to hold (held {qstr(held_)})",
"fix": "receive more stock, or release it from a lower-priority order"}]
return [{"gate": "stock_reserved", "met": True, "reason": "", "fix": ""}]
PRIORITY_RANK = {"rush": 1, "normal": 2}
@INV.use_case(kind="command", route="POST /reservations/sales-orders/{sales_order_id}", permission="inv:reservation:intake",
idempotent=True, tool=False)
def reserve_order(caller: dict, sales_order_id: Ref, revision: int, lines: list[ReservationLineIn], warehouse_id: Optional[Code] = None,
schedule_version: Optional[int] = None, priority: Optional[Union[Literal["normal", "rush"], int]] = None) -> dict:
"""The sales order asks this module to hold stock for its goods lines, on confirm and on every amendment. Keyed by (order,
revision, schedule version): a repeat changes nothing; a later revision sets each line's reservation to its new quantity
(what is already delivered is not held again). As much as is available is held, the rest is short, and each line answers its
readiness {gate: stock_reserved, met, reason, fix}. A service line is ignored and named.
priority is sales-order's (normal | rush, SO-36), kept as its rank (rush 1, normal 2); the delivering site is the request's
warehouse_id, else the one the promised lines name (SO-36 sends it per line with the schedule), else the default."""
priority = PRIORITY_RANK.get(priority, priority) if isinstance(priority, str) else priority
sites = {ln.warehouse_id for ln in lines if ln.warehouse_id}
if warehouse_id is None and len(sites) > 1:
raise Refused(422, "the lines name more than one warehouse: one reservation is held in one warehouse")
warehouse_id = warehouse_id or next(iter(sites), None)
@U.decide
def body(unit):
wh = default_warehouse(unit, warehouse_id)
marker_line = 0
sv = schedule_version or 0
markers = [r for r in unit.reservation_rows(sales_order_id, marker_line) if r["kind"] == "marker"
if r["revision"] == revision and (r["schedule_version"] or 0) == sv]
out_lines = []
for ln in lines:
info = unit.product(ln.product_id)
if info.get("kind") == "service":
out_lines.append({"line_no": ln.line_no, "ignored": "a service line is not stocked: nothing to reserve",
"reserved_base": "0", "short_base": "0", "base_unit_code": info.get("base_unit_code"), "readiness": []})
continue
products.stocked(unit, ln.product_id, for_new_reservation=True)
if ln.unit_numerator and ln.unit_denominator:
base = to_base_exact(ln.quantity, ln.unit_numerator, ln.unit_denominator)
else:
base = products.convert(unit, ln.product_id, ln.quantity, ln.unit_code, what="the ordered quantity").base
delivered = consumed(unit, sales_order_id, ln.line_no)
wanted = max(base - delivered, ZERO)
if markers:
have = held(unit, sales_order_id, ln.line_no)
h, short = have, max(wanted - have, ZERO)
else:
h, short = set_line(unit, sales_order_id, ln.line_no, ln.product_id, wh["warehouse_id"], wanted, revision=revision,
schedule_version=schedule_version, priority=priority)
out_lines.append({"line_no": ln.line_no, "reserved_base": qstr(h), "short_base": qstr(short),
"base_unit_code": info.get("base_unit_code"), "readiness": _gate(short, h, info.get("base_unit_code") or "")})
if not markers:
add_row(unit, sales_order_id, marker_line, "marker", wh["warehouse_id"], "", ZERO, revision=revision, schedule_version=sv)
return {"sales_order_id": sales_order_id, "revision": revision, "lines": out_lines, "replay": bool(markers)}
return U.run(caller, body, lost_message=lambda repo, row=None: "another reservation took the stock at the same moment")
@INV.use_case(kind="command", route="POST /reservations/sales-orders/{sales_order_id}/release", permission="inv:reservation:intake",
idempotent=True, tool=False)
def release_order(caller: dict, sales_order_id: Ref, reason: Optional[Reason] = None) -> dict:
"""The order was cancelled: release what is held and not yet delivered. A repeat releases nothing more."""
@U.decide
def body(unit):
rows = [r for r in unit.repo.find("inv_reservation", sales_order_id=sales_order_id) if r["kind"] != "marker"]
by_line: dict = {}
for r in rows:
e = by_line.setdefault(r["so_line_no"], {"qty": ZERO, "wh": r["warehouse_id"], "p": r["product_id"]})
e["qty"] += r["quantity_base"]
out = []
for line_no, e in sorted(by_line.items()):
if e["qty"] > 0:
add_row(unit, sales_order_id, line_no, "release", e["wh"], e["p"], -e["qty"], reason=reason or "order cancelled")
out.append({"line_no": line_no, "quantity_base": qstr(e["qty"])})
return {"sales_order_id": sales_order_id, "released_base_by_line": out}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /reservations/hand", permission="inv:reservation:write")
def reserve_by_hand(caller: dict, sales_order_id: Ref, line_no: Seq, quantity_base: SignedQuantity, reason: Reason, warehouse_id: Optional[Code] = None,
product_id: Optional[Code] = None) -> dict:
"""The warehouse manager holds stock for an important order (a positive quantity in the base unit) or releases it (a negative
one), with a reason; both are appended with who and why. Reserving more than is available is refused, naming what is."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to reserve or release by hand")
q = quantity_base
from brick_inventory.domain.values import check_quantity
check_quantity(abs(q), "the quantity")
wh = default_warehouse(unit, warehouse_id)
pid = product_id
if not pid:
from brick_inventory.infrastructure import seams
try:
order = seams.sales_order(sales_order_id)
except seams.Missing as m:
raise Refused(502, f"the sales order could not be read: {m.text}") from None
line = next((l for l in order.get("lines", []) if l["line_no"] == line_no), None)
if line is None:
raise RuleBroken(f"order {sales_order_id} has no line {line_no}")
pid = line["product_id"]
info = products.stocked(unit, pid)
have = held(unit, sales_order_id, line_no)
if q > 0:
products.stocked(unit, pid, for_new_reservation=True)
free = available_at(unit, pid, wh["warehouse_id"])
if q > free:
raise RuleBroken(f"{qstr(q)} {info['base_unit_code']} asked, only {qstr(max(free, ZERO))} available at {wh['warehouse_id']}")
add_row(unit, sales_order_id, line_no, "reserve", wh["warehouse_id"], pid, q, reason=reason)
else:
if -q > have:
raise RuleBroken(f"only {qstr(have)} {info['base_unit_code']} is held for {sales_order_id} line {line_no}: {qstr(-q)} cannot be released")
add_row(unit, sales_order_id, line_no, "release", wh["warehouse_id"], pid, q, reason=reason)
return {"sales_order_id": sales_order_id, "line_no": line_no, "held_base": qstr(have + q), "reason": reason}
return U.run(caller, body, lost_message=lambda repo, row=None: "another reservation took the stock at the same moment")