dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/receipts.py
"""Goods in: purchasing posts each received line here (INV-20) and each return to the supplier (INV-28); the scanner records a
delivery THROUGH purchasing (INV-21, the one way a receipt is recorded) and attaches the lots and serials; subcontract
provisions and valuation adjustments arrive from purchasing too (INV-47, INV-57)."""
import re
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import docs, ledger, postings, 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,
CostIn,
InputLotIn,
IsoDate,
ProvidedIn,
Quantity,
Rate,
Ref,
ScanLine,
day,
text,
)
from brick_inventory.domain import location as L
from brick_inventory.domain.errors import BadInput, GateBroken, NotFound, RuleBroken, ShortOf
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import (
NO_LOT,
OURS,
UNRESTRICTED,
ZERO,
cost_in_fractions,
money,
)
from brick_inventory.application.source_ops import SOURCE_RECEIPTS, receive_from_source
from brick_inventory.infrastructure import seams
MOVE_REF = re.compile(r"^pur:(.+):(\d{1,9})$")
BigBp = int
def parse_ref(move_ref: str, source_id: Optional[str] = None) -> tuple:
m = MOVE_REF.match(move_ref)
if not m:
raise BadInput(f"move_ref looks like pur:<receipt_id>:<line_no>, not {move_ref!r}")
if source_id is not None and source_id != m.group(1):
raise BadInput(f"move_ref names {m.group(1)} but source_id is {source_id}")
return m.group(1), int(m.group(2))
def _internal(unit, location_id: str) -> dict:
loc = unit.repo.get("inv_location", location_id=location_id)
if loc is None:
raise GateBroken("location_internal", f"there is no location {location_id}", "pick a bin of a warehouse", 422)
if loc["kind"] != "internal" or loc["status"] != "active":
raise GateBroken("location_internal", f"{location_id} is not an internal, active location", "pick a bin of a warehouse", 422)
unit._locations[location_id] = loc
return loc
def _ensure_lot(unit, product_id: str, lot_code: str, tracking: dict, received_on: date, expires_on: Optional[date], supplier) -> None:
if not lot_code:
return
existing = unit.repo.get("inv_lot", product_id=product_id, lot_code=lot_code)
if existing:
return
exp = expires_on
if exp is None and tracking.get("shelf_life_days"):
exp = received_on + timedelta(days=int(tracking["shelf_life_days"]))
unit.put("inv_lot", {"product_id": product_id, "lot_code": lot_code, "kind": tracking["tracking"], "expires_on": exp,
"received_on": received_on, "blocked": False, "supplier_bp_id": supplier}, ("product_id", "lot_code"))
def _pending_scan(unit, purchase_order_id: Optional[str], product_id: str) -> Optional[dict]:
if not purchase_order_id:
return None
for d in unit.repo.find("inv_document", kind="receipt", lifecycle="scanned", source_ref=purchase_order_id):
for ln in docs.extra_of(d).get("lines", []):
if ln["product_id"] == product_id and not ln.get("used"):
return {"document": d, "line": ln}
return None
def _answer(rows: list, replay: bool, consumed: Optional[list] = None, unit_code: str = "") -> dict:
main = [r for r in rows if r["n"] < 100]
out = {"move_id": main[0]["move_id"], "quantity_base": qstr(sum((r["quantity_base"] for r in main), ZERO)), "replay": replay}
if len(main) > 1:
out["moves"] = [{"move_id": r["move_id"], "lot_code": r["lot_code"] or None, "quantity_base": qstr(r["quantity_base"])}
for r in main]
if consumed:
out["consumed"] = consumed
return out
def _move_rows(unit, doc_id: str, line_no: int) -> list:
rows = []
for r in unit.repo.find("inv_move", order=("move_id",), document_id=doc_id, line_no=line_no):
r["n"] = int(r["move_id"].rsplit(":", 1)[1])
rows.append(r)
return rows
@INV.use_case(kind="command", route="POST /receipts", permission="inv:receipt:intake", idempotent=True, tool=False)
def post_receipt(caller: dict, move_ref: Ref, source_id: Ref, product_id: Code, quantity: Quantity, unit_code: Code,
location_id: Code, received_on: IsoDate, source_kind: Code = "purchase_receipt",
purchase_order_id: Optional[Ref] = None, bp_id: Optional[BigBp] = None, unit_cost: Optional[Rate] = None,
pricing_unit_code: Optional[Code] = None, pricing_numerator: Optional[int] = None,
pricing_denominator: Optional[int] = None, pricing_quantity: Optional[Quantity] = None,
currency: Optional[text(3)] = None, stock_owner_bp_id: Optional[BigBp] = None,
consume_provided: Optional[list[ProvidedIn]] = None, service_fee: Optional[Rate] = None,
delivery_costs: Optional[list[CostIn]] = None, lot_code: Optional[Code] = None,
expires_on: Optional[IsoDate] = None, stock_type: Optional[Code] = None, company_code: Optional[Code] = None,
reverses_move_ref: Optional[Ref] = None, input_lots: Optional[list[InputLotIn]] = None) -> dict:
"""Purchasing posts one received line: a move from the supplier into a bin, converted exactly into the product's base unit
(the unit written beside it), valued at the line's price per pricing unit, rounded once; average cost appended; the journal
asked of the ledger. Keyed by move_ref (pur:<receipt_id>:<line_no>): a repeat returns the same move and records nothing. A
consignment line carries its owner and no value; a subcontract receipt consumes the components held at the subcontractor
and is valued at their value plus the service. A unit that does not convert exactly is refused with product's reason.
v6 (INV-59): source_kind production_receipt | by_product_receipt (manufacturing: an output lot at unit_cost per base unit, with the
input_lots it was made from) and pos_refund_restock (a restocked till refund, at the cost of the move named by reverses_move_ref) are
received through source_ops; their move_ref is the sender's own."""
if source_kind in SOURCE_RECEIPTS:
return receive_from_source(caller, move_ref=move_ref, source_kind=source_kind, source_id=source_id, company_code=company_code,
product_id=product_id, quantity=quantity, unit_code=unit_code, location_id=location_id, received_on=received_on,
lot_code=lot_code, unit_cost=unit_cost, reverses_move_ref=reverses_move_ref, input_lots=input_lots)
receipt_id, line_no = parse_ref(move_ref, source_id)
if source_kind not in ("purchase_receipt", "consignment_receipt", "subcontract_receipt"):
raise Refused(422, "source_kind is purchase_receipt, consignment_receipt, subcontract_receipt, production_receipt, by_product_receipt or pos_refund_restock")
@U.decide
def body(unit):
doc_id = f"rcpt:pur:{receipt_id}"
existing = _move_rows(unit, doc_id, line_no)
if existing:
consumed = [{"move_id": r["move_id"], "product_id": r["product_id"], "quantity_base": qstr(r["quantity_base"])}
for r in existing if r["n"] >= 100]
return _answer(existing, True, consumed)
on = day(received_on)
if on > unit.today:
raise BadInput(f"received_on {on.isoformat()} is in the future: a receipt is dated today or earlier")
loc = _internal(unit, location_id)
info = products.stocked(unit, product_id)
conv = products.convert(unit, product_id, quantity, unit_code, document_ref=doc_id, what="the received quantity")
owner = str(stock_owner_bp_id) if stock_owner_bp_id or source_kind == "consignment_receipt" else OURS
if source_kind == "consignment_receipt" and not stock_owner_bp_id:
raise GateBroken("consignment_owner", "a consignment receipt names the consigning supplier (stock_owner_bp_id)",
"send stock_owner_bp_id", 422)
supplier_loc = L.virtual_id(loc["company_code"], "SUPPLIER")
value = ZERO
paid = None
if not owner:
if unit_cost is None:
raise GateBroken("unit_cost", "a receipt of our own stock needs the line's price (unit_cost)", "send unit_cost", 422)
p_num = pricing_numerator or conv.numerator
p_den = pricing_denominator or conv.denominator
p_qty = pricing_quantity or Decimal(1)
paid = cost_in_fractions(unit_cost, p_num, p_den, p_qty, conv.base)
value = paid
arrives_as = stock_type or loc.get("stock_status") or UNRESTRICTED # a QC area takes quality stock: goods land as that
tracking = unit.tracking(product_id)
splits = _splits(unit, tracking, product_id, conv.base, purchase_order_id, lot_code, expires_on, on, bp_id, info)
total_in = value
consumed_rows, consumed_value = [], ZERO
if source_kind == "subcontract_receipt" and consume_provided:
consumed_rows, consumed_value = _consume_components(unit, doc_id, line_no, loc, consume_provided, bp_id, on)
fee = ZERO
if service_fee is not None:
p_qty = pricing_quantity or Decimal(1)
fee = money(conv.base * Decimal(1) / Decimal(1) * Decimal(service_fee) / p_qty) if not pricing_numerator else \
cost_in_fractions(service_fee, pricing_numerator, pricing_denominator or 1, p_qty, conv.base)
total_in = consumed_value + fee
value, paid = total_in, total_in
document = docs.stage_document(unit, doc_id, "receipt", loc["warehouse_id"], loc["company_code"], bp_id=bp_id,
source_kind=source_kind, source_ref=purchase_order_id or receipt_id, done_on=on,
done_by=U.who(caller))
if existing_doc := unit.repo.get("inv_document", document_id=doc_id):
document["number"] = existing_doc["number"]
else:
document["number"] = docs.claim_number(unit, loc["warehouse_id"], "receipt", doc_id)
unit.put("inv_document", {"document_id": doc_id, "number": document["number"]}, "document_id")
measures = products.measures(info, unit_code)
docs.stage_line(unit, doc_id, line_no, line_kind="receipt", product_id=product_id, unit_code=unit_code,
unit_numerator=conv.numerator, unit_denominator=conv.denominator, quantity_expected=conv.quantity,
quantity_done=conv.quantity, quantity_base=conv.base, to_location_id=location_id,
lot_code=lot_code or NO_LOT, unit_cost=unit_cost, pricing_unit_code=pricing_unit_code,
pricing_numerator=pricing_numerator, pricing_denominator=pricing_denominator,
pricing_quantity=pricing_quantity, value_amount=None if owner else value,
note=docs.dump({"delivery_costs": [d.model_dump(mode="json") for d in delivery_costs]}) if delivery_costs else None,
**measures)
posted, allocated = [], ZERO
for i, split in enumerate(splits, start=1):
share = split["quantity_base"]
if owner:
v = ZERO
elif i == len(splits):
v = total_in - allocated
else:
v = money(total_in * share / conv.base)
allocated += v
spec = MoveSpec(doc_id, line_no, product_id, share, supplier_loc, location_id, on, unit_code, conv.quantity * share / conv.base
if len(splits) > 1 else conv.quantity, conv.numerator, conv.denominator, split["lot_code"],
UNRESTRICTED, arrives_as, OURS, owner, n=i, move_ref=move_ref if i == 1 else f"{move_ref}#{i}",
value_in=None if owner else v, purchase_value=None if owner else (
money(paid * share / conv.base) if len(splits) > 1 and i < len(splits) else (
paid - sum((p.extra.get("paid", ZERO) for p in posted), ZERO))),
zero_value=bool(owner))
pm = ledger.post_move(unit, spec)
pm.extra["paid"] = spec.purchase_value or ZERO
posted.append(pm)
total_value = sum((p.row["value_amount"] for p in posted), ZERO)
diff = sum((p.step.price_difference for p in posted if p.step), ZERO)
if not owner and (total_value or diff):
vclass = info.get("valuation_class") or "*"
if source_kind == "subcontract_receipt":
svc = total_in - consumed_value
lines = [postings.entry("stock_valuation", total_value, valuation_class=vclass),
postings.entry("stock_valuation", consumed_value, natural=postings.CREDIT),
postings.entry("subcontracting", svc, natural=postings.CREDIT)]
else:
lines = postings.receipt_lines(total_value, paid, diff, vclass, f"receipt {doc_id}")
postings.stage(unit, doc_id, f"receipt:{line_no}", loc["company_code"], on, f"goods received {document['number']}", lines)
consumed_out = [{"move_id": r["move_id"], "product_id": r["product_id"], "quantity_base": qstr(r["quantity_base"])}
for r in consumed_rows]
_consume_scan(unit, purchase_order_id, product_id, splits, doc_id, receipt_id)
out = _answer([{**p.row, "n": i} for i, p in enumerate(posted, start=1)], False, consumed_out)
out["number"] = document["number"]
out["value_amount"] = qstr(total_value) if False else f"{total_value:.2f}"
return out
return U.run(caller, body)
def _splits(unit, tracking: dict, product_id: str, base: Decimal, po, lot_code, expires_on, on: date, bp_id, info) -> list:
"""The moves one received line makes: one per serial, one per lot, else one. Lots and serials come from the scanner
(INV-21) or are named on the line."""
t = tracking["tracking"]
if t == "none":
return [{"lot_code": NO_LOT, "quantity_base": base}]
scan = _pending_scan(unit, po, product_id)
if scan:
lots = scan["line"]["lots"]
if sum((Decimal(l["quantity_base"]) for l in lots), ZERO) != base:
raise GateBroken("tracking_complete", f"the scan for {product_id} carries "
f"{qstr(sum((Decimal(l['quantity_base']) for l in lots), ZERO))} but {qstr(base)} arrived",
"scan the lots or serials again", 409)
out = []
for l in lots:
_ensure_lot(unit, product_id, l["lot_code"], tracking, on, day(l.get("expires_on")), bp_id)
out.append({"lot_code": l["lot_code"], "quantity_base": Decimal(l["quantity_base"])})
return out
if lot_code and t == "lot":
_ensure_lot(unit, product_id, lot_code, tracking, on, day(expires_on), bp_id)
return [{"lot_code": lot_code, "quantity_base": base}]
what = "serial numbers" if t == "serial" else "a lot"
raise GateBroken("tracking_complete", f"{product_id} is tracked by {t}: it is received through the goods-in scanner with {what}",
"scan the delivery with its " + what, 409)
def _consume_scan(unit, po, product_id, splits, doc_id, receipt_id) -> None:
scan = _pending_scan(unit, po, product_id)
if not scan:
return
d, ln = scan["document"], scan["line"]
ex = docs.extra_of(d)
for l in ex["lines"]:
if l is not None and l["line_no"] == ln["line_no"] and l["product_id"] == product_id:
l["used"] = True
l["receipt_doc"] = doc_id
unit.put("inv_document", {"document_id": d["document_id"], "extra": docs.dump(ex)}, "document_id")
def _consume_components(unit, doc_id: str, line_no: int, loc: dict, comps: list, bp_id, on: date) -> tuple:
"""The components held at the subcontractor leave it for good, at their value (still ours until now)."""
if not bp_id:
raise BadInput("a subcontract receipt names the subcontractor (bp_id)")
sub = L.subcontract_id(loc["company_code"], bp_id)
if unit.repo.get("inv_location", location_id=sub) is None:
raise GateBroken("components_at_subcontractor", f"nothing is held at subcontractor {bp_id}", "ship the components first", 409)
unit._locations.pop(sub, None)
rows, total = [], ZERO
for i, c in enumerate(comps, start=1):
conv = products.convert(unit, c.product_id, c.quantity, c.unit_code, document_ref=doc_id, what="a component quantity")
spec = MoveSpec(doc_id, line_no, c.product_id, conv.base, sub, L.virtual_id(loc["company_code"], "CONSUMPTION"), on,
c.unit_code, conv.quantity, conv.numerator, conv.denominator, n=100 + i)
try:
pm = ledger.post_move(unit, spec)
except ShortOf as s:
raise GateBroken("components_at_subcontractor", str(s), "ship the missing components to the subcontractor", 409) from None
total += -pm.row["value_amount"]
rows.append({**pm.row, "n": 100 + i})
return rows, total
@INV.use_case(kind="command", route="POST /returns-to-supplier", permission="inv:receipt:intake", idempotent=True, tool=False)
def return_to_supplier(caller: dict, move_ref: Ref, reverses_move_ref: Ref, product_id: Code, quantity: Quantity, unit_code: Code,
location_id: Code, returned_on: IsoDate, source_id: Optional[Ref] = None, source_kind: Code = "purchase_return",
bp_id: Optional[BigBp] = None) -> dict:
"""Purchasing posts a return to the supplier: goods leave a bin for the supplier at the cost of the receipt line they came
in on (never today's average). Keyed by move_ref: a repeat records nothing. When the stock is no longer at the location the
answer is 409 {gate: on_hand} and nothing moves."""
return_id, line_no = parse_ref(move_ref, source_id)
@U.decide
def body(unit):
doc_id = f"rts:pur:{return_id}"
existing = _move_rows(unit, doc_id, line_no)
if existing:
return _answer(existing, True)
on = day(returned_on)
if on > unit.today:
raise BadInput(f"returned_on {on.isoformat()} is in the future")
loc = _internal(unit, location_id)
orig_doc, orig_line = parse_ref(reverses_move_ref)
origin = _move_rows(unit, f"rcpt:pur:{orig_doc}", orig_line)
main = [r for r in origin if r["n"] < 100]
if not main:
raise NotFound(f"no received line {reverses_move_ref}")
conv = products.convert(unit, product_id, quantity, unit_code, document_ref=doc_id, what="the returned quantity")
if main[0]["product_id"] != product_id:
raise RuleBroken(f"{reverses_move_ref} was a receipt of {main[0]['product_id']}, not {product_id}")
total_base = sum((r["quantity_base"] for r in main), ZERO)
total_value = sum((r["value_amount"] for r in main), ZERO)
already = unit.repo.find("inv_move", reverses_move_id=main[0]["move_id"])
returned_before = sum((r["quantity_base"] for r in already), ZERO)
if conv.base + returned_before > total_base:
raise RuleBroken(f"{reverses_move_ref} received {qstr(total_base)} {conv.base_unit_code}, of which {qstr(returned_before)} "
f"already went back: {qstr(conv.base)} more is too much")
lot = main[0]["lot_code"] if len(main) == 1 else (main[0]["lot_code"] if all(r["lot_code"] == main[0]["lot_code"] for r in main) else NO_LOT)
value_out = money(total_value * conv.base / total_base)
owner = main[0]["owner_to"] or OURS
spec = MoveSpec(doc_id, line_no, product_id, conv.base, location_id, L.virtual_id(loc["company_code"], "SUPPLIER"), on,
unit_code, conv.quantity, conv.numerator, conv.denominator, lot, UNRESTRICTED, UNRESTRICTED, owner, OURS,
move_ref=move_ref, reverses_move_id=main[0]["move_id"], value_out=value_out, zero_value=bool(owner))
try:
pm = ledger.post_move(unit, spec)
except ShortOf as s:
raise GateBroken("on_hand", str(s), "the stock has moved: bring it back to the location, or return what is there", 409) from None
docs.stage_document(unit, doc_id, "return", loc["warehouse_id"], loc["company_code"], bp_id=bp_id, source_kind=source_kind,
source_ref=return_id, done_on=on, done_by=U.who(caller), number=docs.claim_number(
unit, loc["warehouse_id"], "return", doc_id) if unit.repo.get("inv_document", document_id=doc_id) is None else None)
docs.stage_line(unit, doc_id, line_no, line_kind="return", product_id=product_id, unit_code=unit_code,
unit_numerator=conv.numerator, unit_denominator=conv.denominator, quantity_done=conv.quantity,
quantity_base=conv.base, from_location_id=location_id, lot_code=lot, value_amount=pm.row["value_amount"])
v = pm.row["value_amount"]
info = unit.product(product_id)
postings.stage(unit, doc_id, f"return:{line_no}", loc["company_code"], on, f"return to supplier {return_id}",
[postings.entry("stock_received_not_billed", -v), postings.entry("stock_valuation", v, valuation_class=info.get("valuation_class") or "*")])
return _answer([{**pm.row, "n": 1}], False)
return U.run(caller, body, lost_message=lambda repo, row=None: "the stock was taken by another writer")
# ── the scanner: the ONE way a receipt is recorded is purchasing's (INV-21) ─────────────────────────────────────
@INV.use_case(kind="command", route="POST /receipts/scan", permission="inv:receipt:validate", idempotent=True)
def scan_delivery(caller: dict, purchase_order_id: Ref, delivery_note: Ref, location_id: Code, lines: list[ScanLine],
received_on: Optional[IsoDate] = None) -> dict:
"""Goods-in scans what a supplier delivered against what purchasing still expects, and submits it. This module never posts a
supplier move itself: it checks the gates (the line is on the order, the units convert exactly, the lots and serials are
complete and new, the quantity is within the tolerance), records the receipt through purchasing (once per delivery note),
and purchasing posts each line back here. Submitting the same delivery note twice records one receipt."""
@U.decide
def prepare(unit):
scan_id = f"scan:{purchase_order_id}:{delivery_note}"
done = unit.repo.get("inv_document", document_id=scan_id)
if done and done["lifecycle"] == "done":
return {"replay": docs.extra_of(done).get("answer")}
loc = _internal(unit, location_id)
if not seams.purchasing_installed():
raise Refused(409, "purchasing is not part of this installation: a receipt is recorded there, so there is nothing to scan against")
try:
exp = seams.expected(purchase_order_id)
except seams.Missing as m:
raise Refused(m.status if m.status in (403, 404) else 502, f"purchasing could not say what {purchase_order_id} still expects: {m.text}") from None
by_line = {l["line_no"]: l for l in exp.get("lines", [])}
tol = int(unit.settings.value("over_receipt_tolerance_pct", unit.today))
scanned = []
for ln in lines:
e = by_line.get(ln.line_no)
if e is None:
raise GateBroken("on_order", f"line {ln.line_no} is not on purchase order {purchase_order_id}",
"scan only what the order expects", 409)
product_id = e["product_id"]
info = products.stocked(unit, product_id)
conv = products.convert(unit, product_id, ln.quantity, ln.unit_code, document_ref=scan_id, what="the scanned quantity")
out_base = Decimal(str(e.get("quantity_base_outstanding") if e.get("quantity_base_outstanding") is not None else e["outstanding"]))
limit = out_base * (Decimal(100) + tol) / Decimal(100)
if conv.base > limit:
outstanding_in_unit = _express(info, out_base, ln.unit_code)
raise GateBroken("within_tolerance", f"{qstr(ln.quantity)} is more than the {outstanding_in_unit} expected: the tolerance is {tol}%",
"scan no more than the order expects, or ask the owner to raise the tolerance", 409)
tracking = unit.tracking(product_id)
lots = _tracking_lots(unit, tracking, product_id, conv.base, ln, loc)
scanned.append({"line_no": ln.line_no, "product_id": product_id, "quantity": str(ln.quantity), "unit_code": ln.unit_code,
"quantity_base": str(conv.base), "lots": lots})
unit.put("inv_document", {"document_id": scan_id, "kind": "receipt", "warehouse_id": loc["warehouse_id"],
"company_code": loc["company_code"], "source_kind": "scan", "source_ref": purchase_order_id,
"lifecycle": "scanned", "reason": delivery_note, "extra": docs.dump({"lines": scanned}),
"requested_by_email": (caller.get("email") or "").lower() or None,
"requested_by_sub": (caller.get("sub") or "").lower() or None}, "document_id")
return {"scan_id": scan_id, "warehouse": loc["warehouse_id"]}
prep = U.run(caller, prepare)
if "replay" in prep:
return {**(prep["replay"] or {}), "replay": True}
body = {"delivery_note": delivery_note, "received_on": received_on or U.Unit.__dict__.get("x") or None, "location_id": location_id,
"lines": [{"line_no": l.line_no, "quantity": str(l.quantity), "unit_code": l.unit_code} for l in lines]}
if body["received_on"] is None:
with U.read() as u:
body["received_on"] = u.today.isoformat()
try:
answer = seams.record_receipt(purchase_order_id, body)
except seams.Missing as m:
gate = m.gate
if isinstance(m.reason, dict) and gate:
raise Refused(m.status or 409, m.reason) from None
raise Refused(m.status if m.status in (403, 404, 409, 422) else 502,
f"purchasing did not record the receipt: {m.text}") from None
@U.decide
def finish(unit):
out = {"receipt_id": answer.get("receipt_id"), "purchase_order_id": purchase_order_id, "delivery_note": delivery_note,
"lines": answer.get("lines", []), "readiness": answer.get("readiness", []), "replay": False}
unit.put("inv_document", {"document_id": prep["scan_id"], "lifecycle": "done", "done_on": unit.today,
"done_by": U.who(caller)}, "document_id")
d = unit.repo.get("inv_document", document_id=prep["scan_id"])
ex = docs.extra_of(d)
ex["answer"] = out
unit.put("inv_document", {"document_id": prep["scan_id"], "extra": docs.dump(ex)}, "document_id")
return out
return U.run(caller, finish)
def _express(info: dict, base_qty: Decimal, unit_code: str) -> str:
from brick_inventory.domain import units as dunits
row = products.unit_row(info, unit_code)
if row is None:
return qstr(base_qty)
return qstr(Decimal(dunits.express(base_qty, row)["quantity"]))
def _tracking_lots(unit, tracking: dict, product_id: str, base: Decimal, ln, loc: dict) -> list:
t = tracking["tracking"]
if t == "none":
return []
if t == "serial":
serials = ln.serials or []
if len(serials) != base:
raise GateBroken("tracking_complete", f"{qstr(base)} {product_id} need {qstr(base)} serial numbers, {len(serials)} given",
"scan a serial number on each unit", 409)
if len(set(serials)) != len(serials):
raise GateBroken("tracking_complete", "a serial number is scanned twice", "scan each unit once", 409)
for s in serials:
where = ledger.serial_where(unit, product_id, s)
if where:
raise GateBroken("tracking_complete", f"{s} is already on hand at {where}", "check the serial number", 409)
return [{"lot_code": s, "quantity_base": "1"} for s in serials]
if not ln.lot_code:
raise GateBroken("tracking_complete", f"{product_id} is tracked by lot: scan its lot number", "scan the lot number", 409)
return [{"lot_code": ln.lot_code, "quantity_base": str(base), "expires_on": ln.expires_on}]