dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/unit.py
"""The unit of work: plan from reads taken before, write at the end.
DataK3 has no read-your-writes inside an open transaction, so a use case that writes a move and then reads on hand would
not see it. The unit therefore keeps what it has PLANNED (the quantity deltas, the next claim_no, the valuation position, the
next document number) beside what the repository read, and writes every row once, at the end:
1. the rows writers RACE for (claims, valuation seq, document numbers, reservation seq): plain INSERTs, so a key another
writer took first is a lost race (ClaimLost) before anything else of this request is written;
2. the other append-only rows (moves, postings...);
3. keyed upserts (documents, locations...).
A lost race re-runs the whole request on a fresh read (a few times): the loser then sees the winner's rows and is refused for
the real reason, or succeeds. Nothing of a lost attempt is kept, and a delivery already told to the sales order is corrected
(compensations) so the order never records what the warehouse did not ship.
"""
from collections import defaultdict
from contextlib import contextmanager
from datetime import date
from decimal import Decimal
from typing import Callable, Optional
from erp_platform import auth
from erp_platform.bricks import Refused
from sqlalchemy.exc import IntegrityError
from brick_inventory.domain.errors import BadInput, ClaimLost, GateBroken, NotFound, RuleBroken
from brick_inventory.domain.settings import Settings
from brick_inventory.domain.values import ZERO
from brick_inventory.infrastructure import business_time, seams
from brick_inventory.infrastructure.tables import Repository, insert_many_or_lose, lost
ATTEMPTS = 3
@contextmanager
def _default_repository():
from erp_platform import db # imported late: it needs BUCKET
with db.engine.begin() as conn:
yield Repository(conn)
def repository():
"""The unit of work's connection: use_cases.repository, looked up per call. erp-platform's in-process composition
(Catalogue, `erp-compose init`'s composition test, `serve` on tier 1) swaps `<package>.application.use_cases.repository`;
keeping the real one there is what lets it wire this brick (found by the small-ERP dry run, 2026-10-25: the
composition test refused inventory with "no brick_inventory.application.use_cases.repository"). Tests may still
monkeypatch this name."""
from brick_inventory.application import use_cases
return use_cases.repository()
def who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def identities(caller: dict) -> tuple:
"""Every identity the caller has (email and sign-in subject, lower-case): the same person under either is the same person."""
return tuple(dict.fromkeys(x.strip().lower() for x in (caller.get("email"), caller.get("sub")) if x and x.strip()))
def can(caller: dict, permission: str) -> bool:
return auth.has_permission(caller, permission)
def fmt(d) -> str:
"""A quantity as a person reads it: 9, 0.75, 144 (no trailing zeros)."""
s = f"{Decimal(d):f}"
return s.rstrip("0").rstrip(".") if "." in s else s
class Unit:
"""One request's reads, plans and writes."""
def __init__(self, repo: Repository, caller: dict, attempt: int = 1):
self.repo, self.caller, self.attempt = repo, caller, attempt
self.now_utc = business_time.now()
self.today: date = business_time.today()
self._settings: Optional[Settings] = None
self._locations: dict = {}
self._products: dict = {}
self._tracking: dict = {}
self._warehouses: dict = {}
self._stock: dict = {} # stock key -> committed quantity (read once per request)
self._reservations: dict = {} # (order, line) -> its committed reservation rows
self.memo: dict = {} # other committed reads of this request, by the helper that made them
self.delta: dict = defaultdict(lambda: ZERO) # stock key -> planned change
self.claim_used: dict = {} # claim key -> last claim_no planned
self.positions: dict = {} # (company, product) -> valuation Position (planned)
self.numbers: dict = {} # series -> last number planned
self.reservation_seq: dict = {}
self.reserved_delta: dict = {} # (warehouse, product) -> planned change of what is reserved
self.catch_up_done: set = set()
self.used_units: set = set() # (product, unit) already told to product
self.races: list = [] # (table, row) plain INSERTs that decide who wins
self.inserts: list = [] # (table, row) plain INSERTs of append-only rows
self.upserts: list = [] # (table, rows, key)
self.after_commit: list[Callable[[], None]] = []
self._opened: set = set()
self.compensations: list[Callable[[], None]] = [] # undo a call to a neighbour if this attempt is lost
self.notes: list = []
# ── reads (cached for the request) ──────────────────────────────────────────────────────────
@property
def settings(self) -> Settings:
if self._settings is None:
self._settings = Settings(self.repo.setting_rows())
return self._settings
def location(self, location_id: str) -> dict:
if location_id not in self._locations:
row = self.repo.get("inv_location", location_id=location_id)
if row is None:
raise RuleBroken(f"there is no location {location_id}")
self._locations[location_id] = row
return self._locations[location_id]
def warehouse(self, warehouse_id: str) -> dict:
if warehouse_id not in self._warehouses:
row = self.repo.get("inv_warehouse", warehouse_id=warehouse_id)
if row is None:
raise NotFound(f"no warehouse {warehouse_id}")
self._warehouses[warehouse_id] = row
return dict(self._warehouses[warehouse_id])
def reservation_rows(self, sales_order_id: str, line_no: int) -> list:
"""The committed reservation rows of one order line: every line of the order is read in ONE query, once per request (what
this request plans is in its races)."""
if sales_order_id not in self._reservations:
by_line: dict = {}
for r in self.repo.find("inv_reservation", sales_order_id=sales_order_id):
by_line.setdefault(r["so_line_no"], []).append(r)
self._reservations[sales_order_id] = by_line
return [dict(r) for r in self._reservations[sales_order_id].get(line_no, [])]
def product(self, product_id: str) -> dict:
"""The product as its module answers (GET /products/{id}): the unit of every quantity here."""
if product_id not in self._products:
try:
self._products[product_id] = seams.product(product_id, self.today)
except seams.Missing as m:
if m.status == 404:
raise RuleBroken(f"there is no product {product_id}") from None
raise Refused(m.status if m.status in (403, 502, 503) else 502,
f"the product catalogue could not answer for {product_id}: {m.text}") from None
return self._products[product_id]
def base_unit(self, product_id: str) -> str:
try:
return self.product(product_id).get("base_unit_code", "")
except (Refused, RuleBroken):
return ""
def tracking(self, product_id: str) -> dict:
"""The product's tracking row in force today (none | lot | serial) and its removal strategy."""
if product_id not in self._tracking:
rows = [r for r in self.repo.find("inv_stocked_product", order=("valid_from",), product_id=product_id)
if r["valid_from"] <= self.today]
r = rows[-1] if rows else None
self._tracking[product_id] = {"tracking": r["tracking"] if r else "none",
"removal_strategy": (r["removal_strategy"] if r else None) or "fifo",
"shelf_life_days": r["shelf_life_days"] if r else None}
return self._tracking[product_id]
# ── the stock as planned ────────────────────────────────────────────────────────────────────
def stock(self, key: tuple) -> Decimal:
if key not in self._stock: # committed stock does not change inside the request: read it once
self._stock[key] = self.repo.stock_one(key)
return self._stock[key] + self.delta.get(key, ZERO)
def next_claim_no(self, key: tuple) -> int:
n = self.claim_used.get(key)
n = (self.repo.last_claim_no(key) if n is None else n) + 1
self.claim_used[key] = n
return n
def next_number(self, series_id: str) -> int:
last = self.numbers.get(series_id)
n = (self.repo.last_number(series_id) if last is None else last) + 1
self.numbers[series_id] = n
return n
# ── writes (staged) ─────────────────────────────────────────────────────────────────────────
def race(self, table: str, row: dict) -> None:
self.races.append((table, {**row, "created_at": self.now_utc.replace(tzinfo=None)}))
def append(self, table: str, row: dict) -> None:
self.inserts.append((table, {**row, "created_at": self.now_utc.replace(tzinfo=None)}))
def put(self, table: str, row: dict, key, *, stored: Optional[bool] = None) -> None:
"""A keyed write, staged. `stored`: the caller KNOWS the key is stored already (True: it read the row in this request) or
not (False: a key minted here) - then the flush writes it without first reading which keys exist (erp-platform 0.9.13);
None looks them up, as before."""
if table == "inv_location":
self._locations[row["location_id"]] = row # a location staged here is usable by the moves staged after it
self.upserts.append((table, [{**row, "created_at": self.now_utc.replace(tzinfo=None)}], key, stored))
def flush_races(self) -> None:
"""Take the keys writers race for NOW (claims, numbers, valuation seq): called before anything is told to a neighbour,
so a request that lost the race has called nobody. They belong to the open transaction: rolled back with it."""
_insert_grouped(self.repo, self.races)
self.races = []
def flush(self) -> None:
_insert_grouped(self.repo, self.races)
_insert_grouped(self.repo, self.inserts)
# keyed writes: one upsert per table (one read of the stored keys, one statement per column set), a key staged twice
# written once with its fields merged in order - what writing them one after another left behind
merged: dict = {}
known: dict = {} # (table, cols) -> {key: stored?}; a key with no answer forces the lookup
for table, rows, key, stored in self.upserts:
cols = (key,) if isinstance(key, str) else tuple(key)
per = merged.setdefault((table, cols), {})
said = known.setdefault((table, cols), {})
for row in rows:
kv = tuple(row[c] for c in cols)
per[kv] = {**per[kv], **row} if kv in per else dict(row)
said[kv] = stored if said.get(kv, stored) == stored else None
for (table, cols), per in merged.items():
said = known[(table, cols)]
existing = None if any(v is None for v in said.values()) else {kv for kv, v in said.items() if v}
self.repo.upsert(table, list(per.values()), cols, existing=existing)
self.races, self.inserts, self.upserts = [], [], []
def _insert_grouped(repo: Repository, staged: list) -> None:
"""Plain INSERTs, one statement per table and column set (in the order first staged): a key another writer took loses the
unit of work, as one row at a time did."""
groups: dict = {}
for table, row in staged:
groups.setdefault((table, tuple(sorted(row))), []).append(row)
for (table, _cols), rows in groups.items():
insert_many_or_lose(repo, table, rows)
def run(caller: dict, body: Callable[[Unit], object], *, lost_message: Optional[Callable[[Repository, Optional[dict]], str]] = None,
attempts: int = ATTEMPTS):
"""Run a use case's body as ONE unit of work, committing once.
A lost race (a plain INSERT hit 23505, at the statement or at COMMIT) rolls the whole transaction back, undoes what the
attempt told a neighbour, and runs the body again on a fresh read. After the last attempt the caller is told, in words,
that another writer took what they needed."""
lost_row: Optional[dict] = None
for attempt in range(1, attempts + 1):
unit = None
try:
with repository() as repo:
unit = Unit(repo, caller, attempt)
result = body(unit)
unit.flush()
except (ClaimLost, IntegrityError) as exc:
if isinstance(exc, IntegrityError) and not lost(exc):
raise
lost_row = getattr(exc, "row", None) or lost_row
if unit is not None:
for undo in reversed(unit.compensations):
try:
undo()
except Exception: # noqa: BLE001 — a failed correction is reported below
pass
continue
for hook in unit.after_commit:
hook()
return result
reason = "another writer took what this needed at the same moment"
if lost_message is not None:
with repository() as repo:
reason = lost_message(repo, lost_row)
raise Refused(409, f"{reason}; nothing was written")
@contextmanager
def read():
"""A read-only look at the stock: a Unit with nothing staged."""
with repository() as repo:
yield Unit(repo, {})
def decide(fn):
"""A broken rule is a status with its own words (a gate: its reason and fix), through both doors."""
import functools
@functools.wraps(fn)
def wrapper(*args, **kwargs):
try:
return fn(*args, **kwargs)
except GateBroken as g:
raise Refused(g.status, g.as_dict()) from None
except NotFound as e:
raise Refused(404, str(e)) from None
except BadInput as e:
raise Refused(422, str(e)) from None
except RuleBroken as e:
raise Refused(409, str(e)) from None
return wrapper