dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/ledger.py
"""The move engine: append a move, claim what it takes, value it, and plan the journal that follows.
Every quantity that moves is an append-only move from one place to another, in the product's BASE unit. On hand is the sum of
moves, derived on read; this module never stores or edits a quantity. A move out of a stock-holding place claims its bin AND
its warehouse with a plain INSERT, so two writers cannot both take the last unit; a move that changes a company's quantity of
a product appends a valuation row, claimed the same way.
"""
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_inventory.application.unit import Unit, fmt
from brick_inventory.domain import location as loc_rules
from brick_inventory.domain import stock as stock_rules
from brick_inventory.domain import valuation as val
from brick_inventory.domain.errors import BadInput, RuleBroken, ShortOf
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import BLOCKED, HOLDING, UNRESTRICTED, VALUED, ZERO
@dataclass
class Posted:
"""What appending one move did: the row, the company effect, and the valuation step (None: no company quantity changed)."""
row: dict
company: str
direction: int = 0 # +1 into the company's stock, -1 out, 0 inside it
step: Optional[val.Step] = None
method: str = "average"
extra: dict = field(default_factory=dict)
def _ours(loc: dict, owner: str) -> bool:
return loc["kind"] in VALUED and not owner
def root_of(loc: dict) -> str:
return loc.get("warehouse_id") or f"{loc['company_code']}:*"
def key_of(spec_product: str, loc_id: str, lot: str, stock_type: str, owner: str) -> tuple:
return stock_rules.key(spec_product, loc_id, lot, stock_type, owner)
def claim(unit: Unit, key: tuple, move_id: str, quantity: Decimal, on_hand_after: Decimal, reservation_ref: str = "") -> None:
"""The row two writers race for: claim_no is the last claim_no on the key + 1, a plain INSERT, so the loser gets 23505."""
unit.race("inv_claim", {"product_id": key[0], "location_id": key[1], "lot_code": key[2], "stock_type": key[3],
"owner": key[4], "claim_no": unit.next_claim_no(key), "move_id": move_id,
"reservation_ref": reservation_ref, "quantity_base": quantity, "on_hand_after": on_hand_after})
def claim_root(unit: Unit, product_id: str, warehouse_root: str, ref: str, quantity: Decimal, reservation: bool = False) -> None:
key = (product_id, warehouse_root, "", "", "")
claim(unit, key, "" if reservation else ref, quantity, ZERO, ref if reservation else "")
def _last_valuation(unit: Unit, company: str, product_id: str) -> Optional[dict]:
"""The committed last valuation row, read once per request (what this request appends is in unit.positions)."""
k = ("inv_valuation", company, product_id)
if k not in unit.memo:
unit.memo[k] = unit.repo.last_valuation(company, product_id)
return dict(unit.memo[k]) if unit.memo[k] else None
def position(unit: Unit, company: str, product_id: str) -> val.Position:
k = (company, product_id)
if k not in unit.positions:
row = _last_valuation(unit, company, product_id)
unit.positions[k] = (val.Position(row["quantity_after"] or ZERO, row["value_after"] or ZERO,
row["average_cost_after"] or ZERO, row["seq"]) if row else val.Position())
return unit.positions[k]
def standard_price_on(unit: Unit, company: str, product_id: str, on: date) -> Optional[dict]:
rows = [r for r in unit.repo.find("inv_standard_price", order=("valid_from",), company_code=company, product_id=product_id)
if r["valid_from"] <= on]
return rows[-1] if rows else None
def append_valuation(unit: Unit, company: str, product_id: str, step: val.Step, *, kind: str, method: str, move_id: str,
quantity_change: Decimal, moved_on: date) -> None:
pos = step.after
unit.positions[(company, product_id)] = pos
unit.race("inv_valuation", {"company_code": company, "product_id": product_id, "seq": pos.seq, "kind": kind,
"method": method, "move_id": move_id, "quantity_base_change": quantity_change,
"value_change": step.value_change, "quantity_after": pos.quantity, "value_after": pos.value,
"average_cost_after": pos.average, "price_difference": step.price_difference,
"moved_on": moved_on})
def catch_up(unit: Unit, company: str, product_id: str, on: date) -> None:
"""Bring a product's valuation up to the company's method and standard price in force on `on`.
A switch to standard cost, or a standard price that has taken effect, revalues the stock on hand once (a revaluation row,
keyed by company, product and valid_from, with its journal); a switch back to average keeps the value. A company on
standard with no price for a product that has stock is refused: a value cannot be made up."""
k = (company, product_id, on)
if k in unit.catch_up_done:
return
unit.catch_up_done.add(k)
method = unit.settings.method(company, on)
prev = position(unit, company, product_id)
last = _last_valuation(unit, company, product_id)
last_method = (last or {}).get("method") or "average"
if method == "standard":
std = standard_price_on(unit, company, product_id, on)
if std is None:
if prev.quantity != 0 or last is not None:
raise RuleBroken(f"{company} values stock at standard cost and {product_id} has no standard price: "
"the accountant sets one (a standard price starts tomorrow at the earliest)")
return
price = std["standard_price"]
if last_method != "standard" or Decimal(str(last["average_cost_after"] or 0)) != price:
changed_method = last_method != "standard"
step = val.switch_to_standard(prev, price) if changed_method else val.revalue(prev, price)
since = max(std["valid_from"], unit.settings.method_since(company, on) or std["valid_from"])
append_valuation(unit, company, product_id, step, kind="method_switch" if changed_method else "revaluation",
method="standard", move_id="", quantity_change=ZERO, moved_on=since)
from brick_inventory.application import postings
postings.revaluation(unit, company, product_id, since, step, prev)
unit.put("inv_standard_price", {"company_code": company, "product_id": product_id, "valid_from": std["valid_from"],
"lifecycle": "effective"}, ("company_code", "product_id", "valid_from"))
elif last_method == "standard":
append_valuation(unit, company, product_id, val.switch_to_average(prev), kind="method_switch", method="average",
move_id="", quantity_change=ZERO, moved_on=on)
def post_move(unit: Unit, spec: MoveSpec) -> Posted:
"""Append one move. Plans from reads taken before; stages claims, the move and its valuation row for the unit's flush."""
src, dst = unit.location(spec.from_location), unit.location(spec.to_location)
if (src["location_id"], spec.stock_type_from, spec.owner_from) == (dst["location_id"], spec.stock_type_to, spec.owner_to):
raise RuleBroken("a move goes from one place to another: these are the same place and stock type")
if src["company_code"] != dst["company_code"]:
other = dst if dst["company_code"] != src["company_code"] else src
raise RuleBroken(f"{other['location_id']} belongs to company {other['company_code']}, not {src['company_code']}: "
"an inter-company move is a sale and a purchase")
for loc in (src, dst):
if loc["kind"] in HOLDING or loc["kind"] == "view":
loc_rules.usable_for_moves(loc)
if (loc_rules.holds_stock(dst) and spec.lot_code and spec.stock_type_to == UNRESTRICTED and not spec.releasing
and (unit.repo.get("inv_lot", product_id=spec.product_id, lot_code=spec.lot_code) or {}).get("blocked")):
spec.stock_type_to = BLOCKED # a blocked lot holds its later stock too: it lands blocked (INV-41)
loc_rules.accepts_type(dst, spec.stock_type_to)
company = src["company_code"]
qty = spec.quantity_base
if qty <= 0:
raise BadInput("a move is of a quantity above zero: a mistake is a reversing move")
_check_tracking(unit, spec, src, dst)
k_from = key_of(spec.product_id, src["location_id"], spec.lot_code, spec.stock_type_from, spec.owner_from)
k_to = key_of(spec.product_id, dst["location_id"], spec.lot_code, spec.stock_type_to, spec.owner_to)
if loc_rules.holds_stock(src) and not spec.skip_claim:
on_hand = unit.stock(k_from)
allowed = spec.allow_negative or (src["kind"] == "internal" and unit.settings.negative_allowed(src["warehouse_id"], unit.today))
if on_hand < qty and not allowed:
lot = f" lot {spec.lot_code}" if spec.lot_code else ""
raise ShortOf(f"{src['location_id']} holds {fmt(max(on_hand, ZERO))} {unit.base_unit(spec.product_id)} of "
f"{spec.product_id}{lot}, {fmt(qty)} asked".replace(" ", " "), src["location_id"], on_hand)
claim(unit, k_from, spec.move_id, qty, on_hand - qty)
claim_root(unit, spec.product_id, root_of(src), spec.move_id, qty)
unit.delta[k_from] -= qty
unit.delta[k_to] += qty
direction = (1 if _ours(dst, spec.owner_to) else 0) - (1 if _ours(src, spec.owner_from) else 0)
step, method = None, "average"
value = ZERO
if direction and not spec.zero_value:
catch_up(unit, company, spec.product_id, spec.moved_on)
method = unit.settings.method(company, spec.moved_on)
pos = position(unit, company, spec.product_id)
change = qty * direction
if method == "standard":
std = standard_price_on(unit, company, spec.product_id, spec.moved_on)
if std is None:
raise RuleBroken(f"{company} values stock at standard cost and {spec.product_id} has no standard price")
step = val.standard_step(pos, change, std["standard_price"], purchase_value=spec.purchase_value,
reversal_value=spec.value_in if spec.reverses_move_id else None)
else:
step = val.average_step(pos, change, value_in=spec.value_in, value_out=spec.value_out)
value = step.value_change
append_valuation(unit, company, spec.product_id, step, kind="move", method=method, move_id=spec.move_id,
quantity_change=change, moved_on=spec.moved_on)
row = {"move_id": spec.move_id, "document_id": spec.document_id, "line_no": spec.line_no, "product_id": spec.product_id,
"lot_code": spec.lot_code, "stock_type_from": spec.stock_type_from, "stock_type_to": spec.stock_type_to,
"owner_from": spec.owner_from, "owner_to": spec.owner_to, "quantity_base": qty, "unit_code": spec.unit_code,
"quantity": spec.quantity if spec.quantity is not None else qty, "unit_numerator": spec.unit_numerator,
"unit_denominator": spec.unit_denominator, "from_location_id": src["location_id"],
"to_location_id": dst["location_id"], "company_code": company, "moved_on": spec.moved_on,
"moved_at": unit.now_utc.replace(tzinfo=None), "value_amount": value, "reverses_move_id": spec.reverses_move_id,
"move_ref": spec.move_ref, "by_email": (unit.caller.get("email") or unit.caller.get("sub") or ""),
"reason": spec.reason, "conversion_source": "product"}
unit.append("inv_move", row)
return Posted(row, company, direction, step, method)
def _check_tracking(unit: Unit, spec: MoveSpec, src: dict, dst: dict) -> None:
t = unit.tracking(spec.product_id)["tracking"]
if t == "none" or (not loc_rules.holds_stock(src) and not loc_rules.holds_stock(dst)):
return
if not spec.lot_code:
what = "a serial number" if t == "serial" else "a lot"
raise RuleBroken(f"{spec.product_id} is tracked by {t}: every move names {what}")
if t == "serial":
if spec.quantity_base != 1:
raise RuleBroken(f"serial {spec.lot_code} of {spec.product_id} is one unit: a serial is on hand with quantity 1")
if not loc_rules.holds_stock(src) and loc_rules.holds_stock(dst):
where = serial_where(unit, spec.product_id, spec.lot_code)
if where:
raise RuleBroken(f"{spec.lot_code} is already on hand at {where}")
if loc_rules.holds_stock(dst):
# every move INTO a holding place claims the serial, so two receivers who both saw it nowhere collide (claim_no = last + 1)
claim(unit, (spec.product_id, "serial", spec.lot_code, "", ""), spec.move_id, spec.quantity_base, spec.quantity_base)
def serial_where(unit: Unit, product_id: str, serial: str) -> str:
"""Where a serial is on hand now (any stock-holding place), or ''. A serial is on hand in at most one place."""
rows = unit.repo.stock(product_id=product_id, lot_code=serial)
for k, v in rows.items():
total = v + unit.delta.get(k, ZERO)
loc = unit.location(k[1]) if _known(unit, k[1]) else None
if total > 0 and loc and loc["kind"] in HOLDING:
return k[1]
for k, v in unit.delta.items():
if k[0] == product_id and k[2] == serial and v > 0 and k not in rows and _known(unit, k[1]):
if unit.location(k[1])["kind"] in HOLDING:
return k[1]
return ""
def _known(unit: Unit, location_id: str) -> bool:
try:
unit.location(location_id)
return True
except RuleBroken:
return False
def is_goods_in(unit: Unit, loc: dict) -> bool:
"""Is this the warehouse's goods-in place while receipts are two-step? Its stock is on hand, never available (INV-50)."""
return (loc.get("kind") == "internal" and loc["location_id"] == f"{loc.get('warehouse_id')}/IN"
and unit.settings.value("receipt_steps", unit.today) == "two")
def stock_rows(unit: Unit, product_id: str, *, warehouse_id: Optional[str] = None, as_of=None,
locations: Optional[set] = None) -> list:
"""The derived stock rows of a product (optionally of one warehouse, or as of a past moment), with their lot's expiry."""
k = ("stock", product_id, as_of, frozenset(locations) if locations is not None else None)
if k not in unit.memo: # committed stock and lots: read once per request (the plan is in unit.delta)
unit.memo[k] = unit.repo.stock(product_id=product_id, as_of=as_of, locations=locations)
if ("inv_lot", product_id) not in unit.memo:
unit.memo[("inv_lot", product_id)] = {r["lot_code"]: r for r in unit.repo.find("inv_lot", product_id=product_id)}
raw, lots = dict(unit.memo[k]), unit.memo[("inv_lot", product_id)]
rows = []
for (p, loc_id, lot, stype, owner), qty in sorted(raw.items()):
if as_of is None:
qty += unit.delta.get((p, loc_id, lot, stype, owner), ZERO)
try:
loc = unit.location(loc_id)
except RuleBroken:
continue
if warehouse_id and loc.get("warehouse_id") != warehouse_id:
continue
rows.append(stock_rules.Row(p, loc_id, lot, stype, owner, qty, loc["kind"], loc.get("warehouse_id") or "",
(lots.get(lot) or {}).get("expires_on"), is_goods_in(unit, loc)))
return rows