dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/trace.py
"""Where a lot or serial came from and went to, and holding a lot (INV-40, INV-41): derived from the moves, never stored."""
import uuid
from decimal import Decimal
from erp_platform.bricks import Refused
from brick_inventory.application import docs, ledger
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import need_read, qstr
from brick_inventory.application.inputs import Code, Reason
from brick_inventory.domain import location as L
from brick_inventory.domain.errors import BadInput
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import BLOCKED, UNRESTRICTED, ZERO
def _doc(unit, cache: dict, document_id: str) -> dict:
if document_id not in cache:
cache[document_id] = unit.repo.get("inv_document", document_id=document_id) or {}
return cache[document_id]
def _history(unit, product_id: str, code: str) -> list:
cache: dict = {}
out = []
for m in unit.repo.moves_of(product_id=product_id, lot_code=code):
d = _doc(unit, cache, m["document_id"])
to = m["to_location_id"]
kind = ("received" if d.get("kind") == "receipt" and not m["reverses_move_id"] else
"delivered" if to.endswith(":CUSTOMER") else "scrapped" if to.endswith(":LOSS") else
"returned" if d.get("kind") in ("return", "customer_return") else "moved")
out.append({"event": kind, "move_id": m["move_id"], "on": m["moved_on"].isoformat() if m["moved_on"] else None,
"document_id": m["document_id"], "document_number": d.get("number"), "from_location_id": m["from_location_id"],
"to_location_id": to, "quantity_base": qstr(m["quantity_base"]), "bp_id": d.get("bp_id"),
"source_ref": d.get("source_ref")})
rank = {"received": 0, "moved": 1, "delivered": 2, "scrapped": 2, "returned": 3}
return sorted(out, key=lambda h: (h["on"] or "", rank[h["event"]])) # same-day moves read in the order a stock's life runs
@INV.use_case(kind="query", route="GET /serials/{product_id}/{serial}")
def get_serial(caller: dict, product_id: Code, serial: Code) -> dict:
"""The life of one serial number: received (supplier, receipt, date), moved, delivered to a customer and returned. A serial is
on hand in at most one place with quantity 1; a delivered serial is on hand nowhere."""
need_read(caller)
with U.read() as unit:
hist = _history(unit, product_id, serial)
if not hist:
raise Refused(404, f"no serial {serial} of {product_id}")
where = ledger.serial_where(unit, product_id, serial)
return {"product_id": product_id, "serial": serial, "on_hand_at": where or None, "history": hist}
def _links(unit, product_id: str, lot_code: str) -> tuple:
"""v6 (INV-59): the lots this lot was made from (back to the raw material) and the lots made from it (forward to every finished lot),
walking inv_lot_link both ways; each carries the customers its own deliveries reached."""
def walk(back: bool) -> list:
out, seen, frontier = [], {(product_id, lot_code)}, [(product_id, lot_code)]
while frontier:
nxt = []
for p, lot in frontier:
rows = (unit.repo.find("inv_lot_link", input_product_id=p, input_lot_code=lot) if not back
else unit.repo.find("inv_lot_link", product_id=p, lot_code=lot))
for r in rows:
key = (r["input_product_id"], r["input_lot_code"]) if back else (r["product_id"], r["lot_code"])
if key in seen:
continue
seen.add(key)
nxt.append(key)
entry = {"product_id": key[0], "lot_code": key[1], "via": r["source_id"], "quantity_base": qstr(r["input_quantity"])}
if not back:
got = {}
for h in _history(unit, key[0], key[1]):
if h["event"] == "delivered":
got[h["bp_id"]] = got.get(h["bp_id"], ZERO) + Decimal(h["quantity_base"])
entry["delivered_to"] = [{"bp_id": b, "quantity_base": qstr(q)} for b, q in got.items()]
out.append(entry)
frontier = nxt
return out
return walk(True), walk(False)
@INV.use_case(kind="query", route="GET /lots/{product_id}/{lot_code}/trace")
def trace_lot(caller: dict, product_id: Code, lot_code: Code) -> dict:
"""A recall in one answer: which supplier and receipt a lot came from, which customers got how much (and on which delivery),
what was scrapped, what is still on hand and where, and whether anything is unaccounted for."""
need_read(caller)
with U.read() as unit:
lot = unit.repo.get("inv_lot", product_id=product_id, lot_code=lot_code)
if lot is None:
raise Refused(404, f"no lot {lot_code} of {product_id}")
hist = _history(unit, product_id, lot_code)
received = sum((Decimal(h["quantity_base"]) for h in hist if h["event"] == "received"), ZERO)
origin = [{"supplier_bp_id": h["bp_id"], "receipt": h["document_number"] or h["document_id"], "source_ref": h["source_ref"],
"quantity_base": h["quantity_base"], "on": h["on"]} for h in hist if h["event"] == "received"]
customers: dict = {}
for h in hist:
if h["event"] == "delivered":
c = customers.setdefault(h["bp_id"], {"bp_id": h["bp_id"], "quantity_base": ZERO, "deliveries": []})
c["quantity_base"] += Decimal(h["quantity_base"])
c["deliveries"].append(h["document_number"] or h["document_id"])
scrapped = sum((Decimal(h["quantity_base"]) for h in hist if h["event"] == "scrapped"), ZERO)
returned = sum((Decimal(h["quantity_base"]) for h in hist if h["event"] == "returned"), ZERO)
rows = [r for r in ledger.stock_rows(unit, product_id) if r.lot_code == lot_code and r.quantity_base > 0 and r.kind == "internal"]
on_hand = sum((r.quantity_base for r in rows), ZERO)
delivered = sum((c["quantity_base"] for c in customers.values()), ZERO)
made_from, made_into = _links(unit, product_id, lot_code)
return {"product_id": product_id, "lot_code": lot_code, "expires_on": lot["expires_on"].isoformat() if lot["expires_on"] else None,
"blocked": bool(lot["blocked"]), "received_base": qstr(received), "from": origin, "made_from": made_from, "made_into": made_into,
"to": [{**c, "quantity_base": qstr(c["quantity_base"])} for c in customers.values()],
"scrapped_base": qstr(scrapped), "returned_base": qstr(returned),
"on_hand": [{"location_id": r.location_id, "stock_type": r.stock_type, "quantity_base": qstr(r.quantity_base)} for r in rows],
"unaccounted_base": qstr(received + returned - delivered - scrapped - on_hand)}
def _hold(unit, product_id: str, lot_code: str, to_type: str, reason: str, doc_id: str) -> list:
moved = []
n = 0
for r in ledger.stock_rows(unit, product_id):
if r.lot_code != lot_code or r.quantity_base <= 0 or r.kind != "internal":
continue
frm = r.stock_type
if (to_type == BLOCKED and frm == BLOCKED) or (to_type == UNRESTRICTED and frm != BLOCKED):
continue
n += 1
loc = unit.location(r.location_id)
L.accepts_type(loc, to_type)
spec = MoveSpec(doc_id, n, product_id, r.quantity_base, r.location_id, r.location_id, unit.today, "", None, 1, 1, lot_code, frm, to_type,
r.owner, r.owner, reason=reason, releasing=to_type == UNRESTRICTED)
pm = ledger.post_move(unit, spec)
moved.append({"location_id": r.location_id, "from_type": frm, "to_type": to_type, "quantity_base": qstr(r.quantity_base),
"move_id": pm.row["move_id"]})
return moved
def _set_lot(caller: dict, product_id: str, lot_code: str, block: bool, reason: str) -> dict:
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to block or release a lot")
lot = unit.repo.get("inv_lot", product_id=product_id, lot_code=lot_code)
if lot is None:
raise Refused(404, f"no lot {lot_code} of {product_id}")
wh = next((unit.location(r.location_id) for r in ledger.stock_rows(unit, product_id) if r.lot_code == lot_code and r.kind == "internal"), None)
doc_id = f"lot:{product_id}:{lot_code}:{uuid.uuid4().hex[:8]}"
moved = []
if wh is not None:
docs.stage_document(unit, doc_id, "type_change", wh["warehouse_id"], wh["company_code"], lifecycle="done", reason=reason,
done_on=unit.today, done_by=U.who(caller), source_kind="lot_hold", source_ref=lot_code,
number=docs.claim_number(unit, wh["warehouse_id"], "type_change", doc_id))
moved = _hold(unit, product_id, lot_code, BLOCKED if block else UNRESTRICTED, reason, doc_id)
unit.put("inv_lot", {**lot, "blocked": block}, ("product_id", "lot_code"))
return {"product_id": product_id, "lot_code": lot_code, "blocked": block, "reason": reason, "changed": moved}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")
@INV.use_case(kind="command", route="POST /lots/{product_id}/{lot_code}/block", permission="inv:stock_type:write")
def block_lot(caller: dict, product_id: Code, lot_code: Code, reason: Reason) -> dict:
"""Hold a lot for a recall: its stock becomes blocked wherever it stands (a stock-type posting, no movement, no value change),
so it is never reserved, picked or promised until released."""
return _set_lot(caller, product_id, lot_code, True, reason)
@INV.use_case(kind="command", route="POST /lots/{product_id}/{lot_code}/release", permission="inv:stock_type:write")
def release_lot(caller: dict, product_id: Code, lot_code: Code, reason: Reason) -> dict:
"""Release a blocked lot: its blocked stock becomes unrestricted again where it stands."""
return _set_lot(caller, product_id, lot_code, False, reason)