dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/transfers.py
"""Moving stock: between bins, between sites through transit, stock types where it stands, and correcting a mistake (INV-09,
INV-15, INV-36, INV-53, INV-55). A move between places of one company changes no value. A transfer to another company is
refused: it is a sale and a purchase."""
import uuid
from datetime import timedelta
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import docs, ledger, products
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, ReceiveLineIn, Ref, TransferLineIn
from brick_inventory.domain import location as L
from brick_inventory.domain.errors import BadInput, RuleBroken
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import NO_LOT, OURS, STOCK_TYPES, UNRESTRICTED, ZERO
from brick_inventory.infrastructure import seams
def types_for(src: dict, dst: dict, line_from: Optional[str], line_to: Optional[str]) -> tuple:
"""(type moved, type arriving): a hold area (quality or blocked) hands over unrestricted stock to a normal bin (a RELEASE),
a normal bin hands stock to a hold area as that area's type (a BLOCK); between two normal bins the type is kept."""
src_default = src.get("stock_status") or UNRESTRICTED
dst_default = dst.get("stock_status") or UNRESTRICTED
frm = line_from or src_default
if line_to:
to = line_to
elif dst_default != UNRESTRICTED:
to = dst_default
elif src_default != UNRESTRICTED and frm == src_default:
to = UNRESTRICTED
else:
to = frm
if frm not in STOCK_TYPES or to not in STOCK_TYPES:
raise BadInput(f"a stock type is one of {', '.join(STOCK_TYPES)}")
return frm, to
def _check_places(unit, src: dict, dst: dict, *, between_sites: bool = False) -> None:
for loc in (src, dst):
if loc["kind"] not in ("internal", "consignment_customer", "subcontractor", "transit"):
raise RuleBroken(f"{loc['location_id']} is not a place stock is kept in")
if src["company_code"] != dst["company_code"]:
other = dst if dst["company_code"] != src["company_code"] else src
raise RuleBroken(f"{dst['warehouse_id'] or dst['location_id']} belongs to company {dst['company_code']}: an inter-company move is a sale and a purchase"
if other is dst else f"{src['location_id']} belongs to company {src['company_code']}: an inter-company move is a sale and a purchase")
def _line_conv(unit, doc_id: str, ln: TransferLineIn):
conv = products.convert(unit, ln.product_id, ln.quantity, ln.unit_code, document_ref=doc_id, what="the quantity")
return conv
@INV.use_case(kind="command", route="POST /transfers", permission="inv:transfer:write", idempotent=True)
def plan_transfer(caller: dict, from_location_id: Code, to_location_id: Code, lines: list[TransferLineIn],
request_id: Optional[Ref] = None) -> dict:
"""Plan a move of stock from one place to another of the same company: between bins of one warehouse (then
move_within_site), or to another warehouse (ship_to_site, then receive_at_site). A move between a normal bin and a quality or
blocked place is a block or a RELEASE of that stock, with no value change. A move to a place of another company is refused.
Keyed by request_id: planning it again returns the same draft."""
@U.decide
def body(unit):
src, dst = unit.location(from_location_id), unit.location(to_location_id)
_check_places(unit, src, dst)
if src["location_id"] == dst["location_id"] and not any(l.from_stock_type or l.to_stock_type for l in lines):
raise RuleBroken("a transfer goes from one place to another")
rid = request_id or uuid.uuid4().hex[:12]
wh = src["warehouse_id"] or dst["warehouse_id"]
doc_id = f"int:{wh}:{rid}"
existing = unit.repo.get("inv_document", document_id=doc_id)
if existing:
return {"document_id": doc_id, "lifecycle": existing["lifecycle"], "replay": True}
docs.stage_document(unit, doc_id, "transfer", wh, src["company_code"], lifecycle="draft", source_kind="transfer",
source_ref=from_location_id)
out = []
for i, ln in enumerate(lines, start=1):
conv = _line_conv(unit, doc_id, ln)
frm, to = types_for(src, dst, ln.from_stock_type, ln.to_stock_type)
L.accepts_type(dst, to)
tr = unit.tracking(ln.product_id)["tracking"]
if tr != "none" and not ln.lot_code:
raise RuleBroken(f"{ln.product_id} is tracked by {tr}: name the {'serial number' if tr == 'serial' else 'lot'}")
docs.stage_line(unit, doc_id, i, line_kind="transfer", product_id=ln.product_id, unit_code=ln.unit_code,
unit_numerator=conv.numerator, unit_denominator=conv.denominator, quantity_expected=conv.quantity,
quantity_base=conv.base, from_location_id=from_location_id, to_location_id=to_location_id,
lot_code=ln.lot_code or NO_LOT, stock_type=frm, owner=to)
out.append({"line_no": i, "product_id": ln.product_id, "quantity_base": qstr(conv.base), "from_stock_type": frm, "to_stock_type": to})
return {"document_id": doc_id, "lifecycle": "draft", "from_location_id": from_location_id, "to_location_id": to_location_id,
"lines": out, "replay": False}
return U.run(caller, body)
def _transfer(unit, document_id: str, *states) -> tuple:
d = unit.repo.get("inv_document", document_id=document_id)
if d is None or d["kind"] != "transfer":
raise Refused(404, f"no transfer {document_id}")
if states and d["lifecycle"] not in states:
raise RuleBroken(f"{document_id} is {d['lifecycle']}: this step needs it {' or '.join(states)}")
lines = [l for l in unit.repo.find("inv_document_line", order=("line_no",), document_id=document_id) if l["line_kind"] in ("transfer", "transit")]
return d, lines
def _move(unit, doc_id, line, src, dst, on, n=1, **kw):
spec = MoveSpec(doc_id, line["line_no"], line["product_id"], kw.pop("quantity_base", line["quantity_base"]), src, dst, on, line["unit_code"],
kw.pop("quantity", line["quantity_expected"]), line["unit_numerator"], line["unit_denominator"], line["lot_code"] or NO_LOT,
kw.pop("stock_type_from", line["stock_type"] or UNRESTRICTED), kw.pop("stock_type_to", line["owner"] or UNRESTRICTED), n=n, **kw)
return ledger.post_move(unit, spec)
@INV.use_case(kind="command", route="POST /transfers/{document_id}/move", permission="inv:transfer:write", idempotent=True)
def move_within_site(caller: dict, document_id: Ref) -> dict:
"""Do a planned move between two bins of one warehouse: one move bin to bin per line, no value change, numbered
{warehouse}/INT/00001. The source must hold it (a lot or serial per tracked unit); the last unit goes to exactly one of two
movers."""
@U.decide
def body(unit):
d, lines = _transfer(unit, document_id)
if d["lifecycle"] == "done":
return {"document_id": document_id, "number": d["number"], "lifecycle": "done", "replay": True}
if d["lifecycle"] != "draft":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only a planned move is done here")
src, dst = unit.location(lines[0]["from_location_id"]), unit.location(lines[0]["to_location_id"])
if src["warehouse_id"] != dst["warehouse_id"]:
raise RuleBroken(f"{src['location_id']} and {dst['location_id']} are in different warehouses: ship it to the other site")
moves = [_move(unit, document_id, l, l["from_location_id"], l["to_location_id"], unit.today) for l in lines]
number = docs.claim_number(unit, d["warehouse_id"], "transfer", document_id)
docs.stage_document(unit, document_id, "transfer", d["warehouse_id"], d["company_code"], lifecycle="done", number=number,
done_on=unit.today, done_by=U.who(caller), source_kind="transfer", source_ref=d["source_ref"])
return {"document_id": document_id, "number": number, "lifecycle": "done", "replay": False,
"moves": [{"move_id": m.row["move_id"], "quantity_base": qstr(m.row["quantity_base"]),
"stock_type_from": m.row["stock_type_from"], "stock_type_to": m.row["stock_type_to"]} for m in moves]}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")
def planned_days(unit, product_id: str, supplying: str, receiving: str) -> Optional[int]:
"""The planned days in transit from product's site data for the receiving site (supplied from the shipping one)."""
if not seams.product_installed():
return None
try:
rows = seams.product_sites(receiving, unit.today)
except seams.Missing:
return None
for r in rows:
if r.get("product_id") == product_id and r.get("supplying_site_id") in (supplying, None, ""):
return int(r.get("planned_delivery_days") or 0)
return None
@INV.use_case(kind="command", route="POST /transfers/{document_id}/ship", permission="inv:transfer:write", idempotent=True)
def ship_to_site(caller: dict, document_id: Ref) -> dict:
"""Ship a planned transfer to another warehouse: the stock moves bin to TRANSIT (the company's transit place), counted in
transit and available nowhere, still valued. The expected arrival date is the ship date plus the days product plans for the
receiving site (null when product plans none)."""
@U.decide
def body(unit):
d, lines = _transfer(unit, document_id)
if d["lifecycle"] == "in_transit" or d["shipped_on"]:
return {"document_id": document_id, "lifecycle": d["lifecycle"], "expected_arrival_on": d["expected_arrival_on"].isoformat()
if d["expected_arrival_on"] else None, "replay": True}
if d["lifecycle"] != "draft":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only a planned transfer ships")
src, dst = unit.location(lines[0]["from_location_id"]), unit.location(lines[0]["to_location_id"])
if src["warehouse_id"] == dst["warehouse_id"]:
raise RuleBroken(f"{src['location_id']} and {dst['location_id']} are in one warehouse: move it within the site")
transit = L.virtual_id(src["company_code"], "TRANSIT")
days = [planned_days(unit, l["product_id"], src["warehouse_id"], dst["warehouse_id"]) for l in lines]
known = [x for x in days if x is not None]
arrival = unit.today + timedelta(days=max(known)) if known else None
for l in lines:
m = _move(unit, document_id, l, l["from_location_id"], transit, unit.today, n=1, stock_type_to=l["owner"] or UNRESTRICTED)
docs.stage_line(unit, document_id, l["line_no"], line_kind="transit", quantity_done=ZERO)
_ = m
docs.stage_document(unit, document_id, "transfer", d["warehouse_id"], d["company_code"], lifecycle="in_transit", shipped_on=unit.today,
expected_arrival_on=arrival, source_kind="transfer", source_ref=d["source_ref"])
return {"document_id": document_id, "lifecycle": "in_transit", "shipped_on": unit.today.isoformat(),
"expected_arrival_on": arrival.isoformat() if arrival else None, "replay": False}
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 /transfers/{document_id}/receive", permission="inv:transfer:write")
def receive_at_site(caller: dict, document_id: Ref, lines: Optional[list[ReceiveLineIn]] = None) -> dict:
"""Receive what arrived at the other warehouse: TRANSIT to the bin. No more may be received than was shipped; what is short
stays in transit on the transfer, with the gate transit_cleared unmet until a person with a second person writes it off.
`lines` is [{line_no, quantity, unit_code}]; leave it out to receive everything shipped."""
@U.decide
def body(unit):
d, tl = _transfer(unit, document_id, "in_transit")
transit = L.virtual_id(d["company_code"], "TRANSIT")
want = {x.line_no: x for x in (lines or [])}
out, short = [], []
for l in tl:
shipped = l["quantity_base"]
already = l["quantity_done"] or ZERO
if want:
x = want.get(l["line_no"])
if x is None:
continue
conv = products.convert(unit, l["product_id"], x.quantity, x.unit_code, document_ref=document_id, what="the received quantity")
got = conv.base
else:
got = shipped - already
if got + already > shipped:
raise RuleBroken(f"line {l['line_no']}: {qstr(got + already)} received but only {qstr(shipped)} was shipped")
if got <= 0:
continue
spec = MoveSpec(document_id, l["line_no"], l["product_id"], got, transit, l["to_location_id"], unit.today, l["unit_code"], got,
1, 1, l["lot_code"] or NO_LOT, l["owner"] or UNRESTRICTED, l["owner"] or UNRESTRICTED, n=2 + int(already > 0))
ledger.post_move(unit, spec)
docs.stage_line(unit, document_id, l["line_no"], quantity_done=already + got)
out.append({"line_no": l["line_no"], "received_base": qstr(got)})
if already + got < shipped:
short.append({"line_no": l["line_no"], "product_id": l["product_id"], "short_base": qstr(shipped - already - got)})
number = docs.claim_number(unit, d["warehouse_id"], "transfer", document_id)
extra = docs.extra_of(d)
extra["shortfall"] = short
docs.stage_document(unit, document_id, "transfer", d["warehouse_id"], d["company_code"], lifecycle="done", number=number,
done_on=unit.today, done_by=U.who(caller), extra=docs.dump(extra), source_kind="transfer", source_ref=d["source_ref"])
return {"document_id": document_id, "number": number, "lifecycle": "done", "received": out, "shortfall": short,
"readiness": [{"gate": "transit_cleared", "met": not short, "reason": "" if not short else "stock is short and still in transit",
"fix": "" if not short else "write the shortfall off with a second person"}]}
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 /transfers/{document_id}/cancel", permission="inv:transfer:write", idempotent=True)
def cancel_transfer(caller: dict, document_id: Ref) -> dict:
"""Cancel a planned move (nothing moved yet); it never had a number, so no number is wasted."""
@U.decide
def body(unit):
d, _ = _transfer(unit, document_id)
if d["lifecycle"] == "cancelled":
return {"document_id": document_id, "lifecycle": "cancelled", "replay": True}
if d["lifecycle"] != "draft":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only a planned move is cancelled")
docs.stage_document(unit, document_id, "transfer", d["warehouse_id"], d["company_code"], lifecycle="cancelled", source_kind="transfer", source_ref=d["source_ref"])
return {"document_id": document_id, "lifecycle": "cancelled", "replay": False}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /transfers/{document_id}")
def get_transfer(caller: dict, document_id: Ref) -> dict:
"""A transfer: its lines, state, number, expected arrival and what is short in transit."""
need_read(caller)
with U.read() as unit:
d, lines = _transfer(unit, document_id)
extra = docs.extra_of(d)
return {"document_id": document_id, "number": d["number"], "lifecycle": d["lifecycle"], "shipped_on": d["shipped_on"].isoformat() if d["shipped_on"] else None,
"expected_arrival_on": d["expected_arrival_on"].isoformat() if d["expected_arrival_on"] else None,
"shortfall": extra.get("shortfall", []),
"lines": [{"line_no": l["line_no"], "product_id": l["product_id"], "quantity_base": qstr(l["quantity_base"]),
"received_base": qstr(l["quantity_done"]), "from_location_id": l["from_location_id"],
"to_location_id": l["to_location_id"], "lot_code": l["lot_code"] or None} for l in lines]}
@INV.use_case(kind="query", route="GET /transfers")
def list_transfers(caller: dict, lifecycle: Optional[Code] = None, overdue: bool = False) -> dict:
"""Transfers, newest first; `overdue` lists those in transit past their expected arrival."""
need_read(caller)
with U.read() as unit:
rows = unit.repo.find("inv_document", order=("-document_id",), limit=200, kind="transfer", **({"lifecycle": lifecycle} if lifecycle else {}))
if overdue:
rows = [r for r in rows if r["lifecycle"] == "in_transit" and r["expected_arrival_on"] and r["expected_arrival_on"] < unit.today]
return {"transfers": [{"document_id": r["document_id"], "number": r["number"], "lifecycle": r["lifecycle"],
"expected_arrival_on": r["expected_arrival_on"].isoformat() if r["expected_arrival_on"] else None} for r in rows]}
@INV.use_case(kind="command", route="POST /stock/change-type", permission="inv:stock_type:write")
def change_stock_type(caller: dict, location_id: Code, product_id: Code, quantity: str, unit_code: Code, from_type: Code, to_type: Code,
reason: Reason, lot_code: Optional[Code] = None) -> dict:
"""Put stock into quality inspection or blocked status where it stands, or release it, with a reason: a move between types at
the same place, no value change. The place must accept the new type (a QC area takes quality only). Available drops by what
is held and rises by what is released; held stock stays on hand and valued."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to change a stock type")
if from_type not in STOCK_TYPES or to_type not in STOCK_TYPES or from_type == to_type:
raise BadInput(f"from_type and to_type are two different types of {', '.join(STOCK_TYPES)}")
loc = unit.location(location_id)
if loc["kind"] != "internal":
raise RuleBroken(f"{location_id} is not a bin")
L.accepts_type(loc, to_type)
conv = products.convert(unit, product_id, quantity, unit_code, document_ref="type_change", what="the quantity")
rid = uuid.uuid4().hex[:12]
doc_id = f"tch:{loc['warehouse_id']}:{rid}"
docs.stage_document(unit, doc_id, "type_change", loc["warehouse_id"], loc["company_code"], lifecycle="done", reason=reason,
done_on=unit.today, done_by=U.who(caller), source_kind="type_change", source_ref=location_id,
number=docs.claim_number(unit, loc["warehouse_id"], "type_change", doc_id))
spec = MoveSpec(doc_id, 1, product_id, conv.base, location_id, location_id, unit.today, unit_code, conv.quantity, conv.numerator,
conv.denominator, lot_code or NO_LOT, from_type, to_type, reason=reason)
pm = ledger.post_move(unit, spec)
return {"document_id": doc_id, "move_id": pm.row["move_id"], "location_id": location_id, "from_type": from_type, "to_type": to_type,
"quantity_base": qstr(conv.base), "by": U.who(caller), "reason": reason}
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 /moves/{move_id}/reverse", permission="inv:move:reverse")
def reverse_move(caller: dict, move_id: Ref, reason: Reason) -> dict:
"""Correct a move recorded by mistake WITHOUT editing it: a new move the other way names the original, which is marked
reversed. Reversing twice is refused. A delivery or a receipt is never reversed this way: a delivery is undone by a return,
a receipt by a return to the supplier."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to reverse a move")
m = unit.repo.get("inv_move", move_id=move_id)
if m is None:
raise Refused(404, f"no move {move_id}")
doc = unit.repo.get("inv_document", document_id=m["document_id"])
if doc is None or doc["kind"] not in ("transfer", "putaway", "type_change") or m["reverses_move_id"]:
raise RuleBroken(f"{move_id} is part of a {doc['kind'] if doc else 'document'}: a delivery is undone by a return, a receipt by a "
"return to the supplier; only a transfer recorded by mistake is reversed here")
if unit.repo.find("inv_move", reverses_move_id=move_id):
raise RuleBroken(f"{move_id} was already reversed")
rid = f"rev:{move_id}"
spec = MoveSpec(rid, 1, m["product_id"], m["quantity_base"], m["to_location_id"], m["from_location_id"], unit.today, m["unit_code"],
m["quantity"], m["unit_numerator"] or 1, m["unit_denominator"] or 1, m["lot_code"] or NO_LOT, m["stock_type_to"],
m["stock_type_from"], m["owner_to"] or OURS, m["owner_from"] or OURS, reason=reason, reverses_move_id=move_id,
value_out=None)
wh = unit.location(m["to_location_id"])["warehouse_id"] or doc["warehouse_id"]
pm = ledger.post_move(unit, spec)
docs.stage_document(unit, rid, "transfer", wh, m["company_code"], lifecycle="done", reason=reason, done_on=unit.today,
done_by=U.who(caller), source_kind="reversal", source_ref=move_id,
number=docs.claim_number(unit, wh, "transfer", rid))
return {"reversal_move_id": pm.row["move_id"], "reverses_move_id": move_id, "reason": reason}
return U.run(caller, body, lost_message=lambda repo, row=None: "the stock has moved on since: it cannot be reversed")