dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/infrastructure/tables.py
"""The tables and one repository over an open connection.
Every table is prefixed `inv_`, has a natural primary key (never SERIAL), quantities are NUMERIC(19,4) in the product's
base unit, money NUMERIC(18,2), a cost per unit a RATE NUMERIC(19,4). The ledger tables (inv_move, inv_claim,
inv_valuation, inv_number_series, inv_reservation) are append-only: they are written with a plain INSERT, and a key
another writer already took is the signal that it won the race (23505 is not an upsert). schema.sql is written from this
column model (scripts/gen_schema.py); tests/test_schema.py keeps the two equal.
DataK3 has no read-your-writes inside an open transaction, so no use case reads back what it wrote: it plans from reads
taken before (application/unit.py) and writes at the end.
"""
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from erp_platform.repository import upsert as platform_upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
and_,
func,
insert,
select,
)
from sqlalchemy.exc import IntegrityError
from brick_inventory.domain.errors import ClaimLost
from brick_inventory.domain.values import OURS, ZERO
md = MetaData()
QTY, MONEY, RATE = Numeric(19, 4), Numeric(18, 2), Numeric(19, 4)
S = String
def _t(name, *cols):
return Table(name, md, *cols, Column("created_at", TIMESTAMP))
def pk(name, type_=S):
return Column(name, type_, primary_key=True)
def c(name, type_=S):
return Column(name, type_)
inv_warehouse = _t("inv_warehouse", pk("warehouse_id"), c("name"), c("company_code"), c("address"), c("status"))
inv_location = _t("inv_location", pk("location_id"), c("warehouse_id"), c("company_code"), c("parent_location_id"), c("kind"),
c("name"), c("stock_status"), c("allowed_stock_types"), c("pick_sequence", Integer), c("max_quantity_base", QTY),
c("max_weight", QTY), c("max_volume", QTY), c("barcode"), c("bp_id", BigInteger), c("status"))
inv_stocked_product = _t("inv_stocked_product", pk("product_id"), pk("valid_from", Date), c("tracking"), c("removal_strategy"),
c("shelf_life_days", Integer), c("set_by"))
inv_lot = _t("inv_lot", pk("product_id"), pk("lot_code"), c("kind"), c("expires_on", Date), c("received_on", Date),
c("blocked", Boolean), c("supplier_bp_id", BigInteger), c("lot_unit_code"), c("lot_numerator", BigInteger),
c("lot_denominator", BigInteger), c("catch_unit_code"))
inv_document = _t("inv_document", pk("document_id"), c("kind"), c("warehouse_id"), c("company_code"), c("bp_id", BigInteger),
c("source_kind"), c("source_ref"), c("purpose"), c("cost_centre"), c("rma_ref"), c("expected_arrival_on", Date),
c("shipped_on", Date), c("pick_batch_id"), c("number"), c("lifecycle"), c("done_on", Date), c("done_by"),
c("requested_by_email"), c("requested_by_sub"), c("approved_by_email"), c("approved_by_sub"),
c("back_order_of"), c("reason"), c("reversed", Boolean), c("extra"), c("updated_at", TIMESTAMP))
inv_document_line = _t("inv_document_line", pk("document_id"), pk("line_no", Integer), c("line_kind"), c("parent_line_no", Integer),
c("product_id"), c("unit_code"), c("unit_numerator", BigInteger), c("unit_denominator", BigInteger),
c("quantity_expected", QTY), c("quantity_done", QTY), c("quantity_base", QTY), c("source_line_ref"),
c("from_location_id"), c("to_location_id"), c("lot_code"), c("stock_type"), c("owner"),
c("unit_cost", RATE), c("pricing_unit_code"), c("pricing_numerator", BigInteger),
c("pricing_denominator", BigInteger), c("pricing_quantity", QTY), c("gross_weight", QTY),
c("net_weight", QTY), c("weight_unit_code"), c("volume", QTY), c("volume_unit_code"),
c("catch_quantity_expected", QTY), c("value_amount", MONEY), c("note"))
inv_move = _t("inv_move", pk("move_id"), c("document_id"), c("line_no", Integer), c("product_id"), c("lot_code"),
c("stock_type_from"), c("stock_type_to"), c("owner_from"), c("owner_to"), c("quantity_base", QTY), c("unit_code"),
c("quantity", QTY), c("unit_numerator", BigInteger), c("unit_denominator", BigInteger), c("from_location_id"),
c("to_location_id"), c("company_code"), c("moved_on", Date), c("moved_at", TIMESTAMP), c("value_amount", MONEY),
c("reverses_move_id"), c("move_ref"), c("by_email"), c("reason"), c("catch_quantity", QTY),
c("catch_unit_code"), c("conversion_source"))
inv_claim = _t("inv_claim", pk("product_id"), pk("location_id"), pk("lot_code"), pk("stock_type"), pk("owner"),
pk("claim_no", Integer), c("move_id"), c("reservation_ref"), c("quantity_base", QTY), c("on_hand_after", QTY))
inv_reservation = _t("inv_reservation", pk("sales_order_id"), pk("so_line_no", Integer), pk("seq", Integer), c("kind"),
c("warehouse_id"), c("product_id"), c("quantity_base", QTY), c("revision", Integer),
c("schedule_version", Integer), c("confirmed_on", Date), c("priority", Integer), c("reason"),
c("by_email"), c("reserved_at", TIMESTAMP))
inv_count = _t("inv_count", pk("count_id"), c("warehouse_id"), c("location_id"), c("scope"), c("product_ids"), c("source"),
c("lifecycle"), c("started_by"), c("started_by_sub"), c("submitted_by"), c("submitted_by_sub"),
c("decided_by"), c("decided_sub"), c("reason"), c("value_amount", MONEY), c("applied_on", Date),
c("updated_at", TIMESTAMP))
inv_count_line = _t("inv_count_line", pk("count_id"), pk("location_id"), pk("product_id"), pk("lot_code"), pk("stock_type"),
c("counted_base", QTY), c("counted_at", TIMESTAMP), c("counted_by"), c("counted_sub"),
c("expected_base", QTY), c("difference_base", QTY), c("value_amount", MONEY), c("suggestion"))
inv_count_entry = _t("inv_count_entry", pk("count_id"), pk("location_id"), pk("product_id"), pk("lot_code"), pk("stock_type"),
pk("entry_no", Integer), c("quantity", QTY), c("unit_code"), c("unit_numerator", BigInteger),
c("unit_denominator", BigInteger), c("quantity_base", QTY), c("catch_quantity", QTY),
c("counted_by"), c("counted_sub"), c("counted_at", TIMESTAMP))
inv_suggestion = _t("inv_suggestion", pk("count_id"), pk("location_id"), pk("product_id"), pk("lot_code"), pk("step"),
c("text"), c("decided"), c("decided_by"), c("decided_at", TIMESTAMP))
inv_valuation = _t("inv_valuation", pk("company_code"), pk("product_id"), pk("seq", Integer), c("kind"), c("method"),
c("move_id"), c("quantity_base_change", QTY), c("value_change", MONEY), c("quantity_after", QTY),
c("value_after", MONEY), c("average_cost_after", RATE), c("price_difference", MONEY), c("moved_on", Date),
c("catch_quantity_change", QTY), c("catch_quantity_after", QTY))
inv_posting = _t("inv_posting", pk("document_id"), pk("event"), c("company_code"), c("journal_id", BigInteger), c("status"),
c("reason"), c("attempts", Integer), c("posted_at", TIMESTAMP), c("payload"))
inv_account_map = _t("inv_account_map", pk("company_code"), pk("purpose"), pk("valuation_class"), pk("valid_from", Date),
c("account_id", BigInteger), c("set_by"))
inv_number_series = _t("inv_number_series", pk("series_id"), pk("number", Integer), c("document_id"), c("claimed_at", TIMESTAMP))
inv_migration_map = _t("inv_migration_map", pk("odoo_db"), pk("odoo_model"), pk("odoo_id", BigInteger), c("our_key"), c("run_id"))
inv_migration_run = _t("inv_migration_run", pk("run_id"), c("odoo_db"), c("cutover", Date), c("status"), c("signed_by"),
c("signed_at", TIMESTAMP), c("summary"))
inv_odoo_history = _t("inv_odoo_history", pk("odoo_db"), pk("odoo_move_line_id", BigInteger), c("reference"), c("product_id"),
c("lot_code"), c("quantity_base", QTY), c("from_location_id"), c("to_location_id"),
c("done_at", TIMESTAMP), c("origin"))
inv_expected_return = _t("inv_expected_return", pk("rma_ref"), pk("line_no", Integer), c("sales_order_id"), c("bp_id", BigInteger),
c("warehouse_id"), c("product_id"), c("unit_code"), c("quantity_base", QTY),
c("original_delivery_ref"), c("serials"), c("inspect", Boolean), c("received_base", QTY), c("status"))
inv_inspection = _t("inv_inspection", pk("document_id"), pk("line_no", Integer), pk("decision_no", Integer), c("decision"),
c("quantity_base", QTY), c("reason"), c("inspected_by_email"), c("inspected_by_sub"),
c("reported_to_order", Boolean))
inv_abc_class = _t("inv_abc_class", pk("warehouse_id"), pk("product_id"), pk("valid_from", Date), c("abc_class"),
c("consumption_value", MONEY), c("cumulative_share", RATE), c("source"), c("reason"), c("set_by"))
inv_putaway_rule = _t("inv_putaway_rule", pk("warehouse_id"), pk("seq", Integer), pk("valid_from", Date), c("applies_to"),
c("target"), c("strategy"), c("target_location_id"))
inv_pick_batch = _t("inv_pick_batch", pk("pick_batch_id"), c("warehouse_id"), c("delivery_document_ids"), c("status"),
c("created_by"))
inv_package_type = _t("inv_package_type", pk("package_type_code"), c("tare_weight", QTY), c("weight_unit_code"), c("length", QTY),
c("width", QTY), c("height", QTY), c("dimension_unit_code"), c("max_weight", QTY))
inv_package = _t("inv_package", pk("sscc"), c("document_id"), c("package_type_code"), c("contents"), c("gross_weight", QTY),
c("weight_unit_code"), c("packed_by"), c("packed_at", TIMESTAMP))
inv_replenishment_proposal = _t("inv_replenishment_proposal", pk("warehouse_id"), pk("product_id"), pk("business_date", Date),
c("site_row_valid_from", Date), c("on_hand_base", QTY), c("reserved_base", QTY),
c("expected_base", QTY), c("projected_base", QTY), c("reorder_point", QTY), c("max_stock", QTY),
c("proposed_base", QTY), c("rounded", QTY), c("unit_code"), c("procurement"),
c("supplying_site_id"), c("lifecycle"), c("request_ref"), c("purchase_request_id"),
c("transfer_document_id"), c("decided_by"), c("reason"))
inv_standard_price = _t("inv_standard_price", pk("company_code"), pk("product_id"), pk("valid_from", Date),
c("standard_price", RATE), c("set_by"), c("lifecycle"), c("source_ref"), c("set_sub")) # v6: mfg:<rollup_id>, the releaser's sub
# v6 (INV-59): an output lot made from an input lot, the step production adds to the trace. Append-only.
inv_lot_link = _t("inv_lot_link", pk("product_id"), pk("lot_code"), pk("input_product_id"), pk("input_lot_code"), pk("move_ref"),
c("input_quantity", QTY), c("source_kind"), c("source_id"))
inv_settings = _t("inv_settings", pk("key"), pk("valid_from", Date), c("value"), c("set_by"))
ALL_TABLES = {t.name: t for t in md.sorted_tables}
# the rows the doors never edit: written once, with a plain INSERT
APPEND_ONLY = {"inv_move", "inv_claim", "inv_valuation", "inv_number_series", "inv_reservation"}
def _naive(dt: datetime) -> datetime:
return dt.replace(tzinfo=None) if dt.tzinfo else dt
class Repository:
"""Reads and writes over one open connection. Reads never include what the same transaction wrote."""
def __init__(self, conn):
self.conn = conn
# ── generic ───────────────────────────────────────────────────────────────────────────────
def table(self, name):
return ALL_TABLES[name]
def get(self, table: str, **key) -> Optional[dict]:
t = ALL_TABLES[table]
row = self.conn.execute(select(t).where(and_(*[t.c[k] == v for k, v in key.items()]))).mappings().first()
return dict(row) if row else None
def find(self, table: str, *, order: Optional[tuple] = None, limit: Optional[int] = None, **where) -> list:
t = ALL_TABLES[table]
q = select(t)
if where:
q = q.where(and_(*[t.c[k].in_(list(v)) if isinstance(v, (list, set, tuple)) else t.c[k] == v
for k, v in where.items()]))
if order:
q = q.order_by(*[t.c[o.lstrip("-")].desc() if o.startswith("-") else t.c[o] for o in order])
if limit:
q = q.limit(limit)
return [dict(r) for r in self.conn.execute(q).mappings()]
def insert(self, table: str, row: dict) -> None:
"""A plain INSERT: a key another writer already took is a lost race (ClaimLost), never an overwrite."""
self.conn.execute(insert(ALL_TABLES[table]).values(**row))
def upsert(self, table: str, rows: list, key, existing=None) -> None:
"""`existing`: the keys known to be stored (None: look them up first)."""
platform_upsert(self.conn, ALL_TABLES[table], rows, key, existing=existing)
# ── settings (dated, never back-dated) ─────────────────────────────────────────────────────
def setting_rows(self) -> list:
return self.find("inv_settings", order=("key", "valid_from"))
# ── the stock, derived from the moves ──────────────────────────────────────────────────────
def stock(self, *, product_id: Optional[str] = None, as_of: Optional[datetime] = None, locations: Optional[set] = None,
lot_code: Optional[str] = None, product_ids: Optional[set] = None) -> dict:
"""{(product, location, lot, stock_type, owner): quantity} = moves in less moves out, recorded by `as_of`."""
m = inv_move
out = {}
for side, sign in (("to", 1), ("from", -1)):
loc, typ, own = m.c[f"{side}_location_id"], m.c[f"stock_type_{side}"], m.c[f"owner_{side}"]
q = (select(m.c.product_id, loc, m.c.lot_code, typ, own, func.sum(m.c.quantity_base))
.group_by(m.c.product_id, loc, m.c.lot_code, typ, own))
conds = []
if product_id is not None:
conds.append(m.c.product_id == product_id)
if product_ids is not None:
conds.append(m.c.product_id.in_(list(product_ids)))
if as_of is not None:
conds.append(m.c.moved_at <= _naive(as_of))
if locations is not None:
conds.append(loc.in_(list(locations)))
if lot_code is not None:
conds.append(m.c.lot_code == lot_code)
if conds:
q = q.where(and_(*conds))
for p, l, lot, t, o, total in self.conn.execute(q):
k = (p, l, lot or "", t or "unrestricted", o or OURS)
out[k] = out.get(k, ZERO) + sign * Decimal(str(total or 0))
return {k: v.quantize(Decimal("0.0001")) for k, v in out.items() if v != 0}
def stock_one(self, key: tuple) -> Decimal:
got = self.stock(product_id=key[0], locations={key[1]}, lot_code=key[2])
return got.get(key, ZERO)
def moves_of(self, *, document_id: Optional[str] = None, product_id: Optional[str] = None,
lot_code: Optional[str] = None, order=("moved_at", "move_id")) -> list:
where = {k: v for k, v in (("document_id", document_id), ("product_id", product_id), ("lot_code", lot_code)) if v}
return self.find("inv_move", order=order, **where)
# ── claims, valuation, numbers: the keys writers race for ──────────────────────────────────
def last_claim_no(self, key: tuple) -> int:
t = inv_claim
v = self.conn.execute(select(func.max(t.c.claim_no)).where(and_(
t.c.product_id == key[0], t.c.location_id == key[1], t.c.lot_code == key[2], t.c.stock_type == key[3],
t.c.owner == key[4]))).scalar()
return int(v or 0)
def last_valuation(self, company_code: str, product_id: str) -> Optional[dict]:
t = inv_valuation
row = self.conn.execute(select(t).where(and_(t.c.company_code == company_code, t.c.product_id == product_id))
.order_by(t.c.seq.desc()).limit(1)).mappings().first()
return dict(row) if row else None
def valuation_rows(self, company_code: str, product_id: str) -> list:
return self.find("inv_valuation", order=("seq",), company_code=company_code, product_id=product_id)
def last_number(self, series_id: str) -> int:
t = inv_number_series
return int(self.conn.execute(select(func.max(t.c.number)).where(t.c.series_id == series_id)).scalar() or 0)
def last_reservation_seq(self, sales_order_id: str, so_line_no: int) -> int:
t = inv_reservation
return int(self.conn.execute(select(func.max(t.c.seq)).where(and_(
t.c.sales_order_id == sales_order_id, t.c.so_line_no == so_line_no))).scalar() or 0)
def reserved(self, warehouse_id: str, product_id: str, *, on: Optional[date] = None) -> Decimal:
"""Reserved at a warehouse = the sum of its reservation rows (a release is negative; a delivery consumes)."""
t = inv_reservation
v = self.conn.execute(select(func.sum(t.c.quantity_base)).where(and_(
t.c.warehouse_id == warehouse_id, t.c.product_id == product_id))).scalar()
return Decimal(str(v or 0)).quantize(Decimal("0.0001"))
def locations_in(self, warehouse_id: str) -> list:
return self.find("inv_location", order=("location_id",), warehouse_id=warehouse_id)
def delete_where(self, table: str, **where) -> None:
t = ALL_TABLES[table]
self.conn.execute(t.delete().where(and_(*[t.c[k].in_(list(v)) if isinstance(v, (list, set, tuple)) else t.c[k] == v
for k, v in where.items()])))
def lost(exc: BaseException) -> bool:
"""Is this the engine saying another writer took the key (23505, or SQLite's UNIQUE failure)?"""
from erp_platform import retry
text = str(exc).lower()
return retry.sqlstate(exc) == "23505" or "unique constraint failed" in text or "duplicate key" in text
def insert_many_or_lose(repo: Repository, table: str, rows: list) -> None:
"""Plain INSERTs of rows naming the same columns, in one statement: a key another writer already took loses them all (ClaimLost),
as the first of them failing one by one would have lost the whole unit of work."""
if len(rows) == 1:
return insert_or_lose(repo, table, rows[0])
try:
repo.conn.execute(insert(ALL_TABLES[table]).values(rows))
except IntegrityError as exc:
if lost(exc):
raise ClaimLost(table, rows[0]) from None
raise
def insert_or_lose(repo: Repository, table: str, row: dict) -> None:
try:
repo.insert(table, row)
except IntegrityError as exc:
if lost(exc):
raise ClaimLost(table, row) from None
raise