dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/infrastructure/tables.py
"""The tables, and the repositories that load and save them.
Every table is prefixed `mfg_` and has a natural primary key (never SERIAL). Two aggregates load and save whole through
erp_platform's Aggregate - a BOM (versions and lines under it) and a production order - and save only what changed, so a
statement count does not grow with the number of lines. Everything else is a FACT written once with a plain INSERT (an
order movement, a claim, an outbox row, a settlement): the only edits are the outbox row's status and the movement's answer
from inventory, both written by the dispatcher after the local commit. schema.sql and migrations/ hold the same column
model; tests/test_schema.py keeps them equal.
"""
import json
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from zoneinfo import ZoneInfo
from erp_platform import work
from erp_platform.repository import Aggregate, Part, upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
and_,
bindparam,
func,
insert,
select,
update,
)
from brick_manufacturing.domain.bom import Bom, BomLine, BomVersion
from brick_manufacturing.domain.production_order import ProductionOrder
md = MetaData()
_TS = lambda: (Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP)) # noqa: E731
def _t(name, *cols):
return Table(name, md, *cols, *_TS())
mfg_bom = _t("mfg_bom", Column("bom_id", String, primary_key=True), Column("product_id", String), Column("usage", String),
Column("alternative", String), Column("company_code", String), Column("row_version", Integer))
mfg_bom_version = _t("mfg_bom_version", Column("bom_id", String, primary_key=True), Column("version", Integer, primary_key=True),
Column("valid_from", Date), Column("valid_to", Date), Column("output_quantity", Numeric(19, 4)),
Column("output_unit_code", String), Column("output_quantity_base", Numeric(19, 4)),
Column("output_base_unit_code", String), Column("planned_yield_pct", Numeric(9, 4)), Column("status", String),
Column("released_by", String), Column("released_sub", String))
mfg_bom_line = _t("mfg_bom_line", Column("bom_id", String, primary_key=True), Column("version", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True), Column("kind", String), Column("product_id", String),
Column("quantity", Numeric(19, 4)), Column("unit_code", String), Column("quantity_base", Numeric(19, 4)),
Column("base_unit_code", String), Column("replaces_product_id", String), Column("modifier_id", String),
Column("nrv_unit_cost", Numeric(19, 4)))
mfg_production_order = _t(
"mfg_production_order", Column("production_order_id", String, primary_key=True), Column("company_code", String),
Column("product_id", String), Column("warehouse_id", String), Column("planned_quantity", Numeric(19, 4)),
Column("unit_code", String), Column("due_on", Date), Column("bom_id", String), Column("bom_version", Integer),
Column("frozen_bom_json", String), Column("status", String), Column("created_by", String), Column("released_on", Date),
Column("released_by", String), Column("shortage_warning", String), Column("completed_on", Date), Column("closed_on", Date),
Column("cancel_reason", String), Column("row_version", Integer), Column("routing_version", Integer),
Column("frozen_routing_json", String), Column("planned_order_ref", String), Column("stale_note", String))
mfg_order_movement = _t(
"mfg_order_movement", Column("move_ref", String, primary_key=True), Column("production_order_id", String),
Column("kind", String), Column("line_no", Integer), Column("seq", Integer), Column("product_id", String),
Column("lot_code", String), Column("quantity", Numeric(19, 4)), Column("unit_code", String),
Column("quantity_base", Numeric(19, 4)), Column("base_unit_code", String), Column("location_id", String),
Column("over_issue", Boolean), Column("unit_cost", Numeric(19, 4)), Column("inventory_move_id", String),
Column("value", Numeric(18, 2)), Column("status", String), Column("reject_reason", String), Column("recorded_by", String),
Column("recorded_sub", String), Column("business_date", Date))
mfg_lot_genealogy = _t(
"mfg_lot_genealogy", Column("output_product_id", String, primary_key=True), Column("output_lot_code", String, primary_key=True),
Column("input_product_id", String, primary_key=True), Column("input_lot_code", String, primary_key=True),
Column("production_order_id", String, primary_key=True), Column("input_quantity", Numeric(19, 4)),
Column("output_quantity", Numeric(19, 4)))
mfg_settlement = _t(
"mfg_settlement", Column("production_order_id", String, primary_key=True), Column("settlement_no", Integer, primary_key=True),
Column("wip_debit", Numeric(18, 2)), Column("wip_credit", Numeric(18, 2)), Column("standard_material", Numeric(18, 2)),
Column("abnormal_scrap", Numeric(18, 2)), Column("material_usage_variance", Numeric(18, 2)),
Column("activity_variance", Numeric(18, 2)), Column("journal_id", BigInteger), Column("valuation_method", String),
Column("settled_by", String), Column("settled_sub", String), Column("approved_by", String), Column("approved_sub", String),
Column("business_date", Date))
mfg_variance_approval = _t(
"mfg_variance_approval", Column("production_order_id", String, primary_key=True),
Column("approval_no", Integer, primary_key=True), Column("variance", Numeric(18, 2)), Column("approved_by", String),
Column("approved_sub", String), Column("approved_on", Date))
mfg_outbox = _t(
"mfg_outbox", Column("target", String, primary_key=True), Column("ref", String, primary_key=True),
Column("production_order_id", String), Column("payload", String), Column("status", String), Column("attempts", Integer),
Column("last_error", String), Column("sent_at", TIMESTAMP))
mfg_claim = Table("mfg_claim", md, Column("claim_scope", String, primary_key=True), Column("claim_key", String, primary_key=True),
Column("n", Integer, primary_key=True), Column("ref", String), Column("claimed_by", String),
Column("claimed_at", TIMESTAMP))
mfg_number_series = Table("mfg_number_series", md, Column("company_code", String, primary_key=True),
Column("year", Integer, primary_key=True), Column("n", Integer, primary_key=True),
Column("production_order_id", String), Column("claimed_at", TIMESTAMP))
mfg_settings = _t("mfg_settings", Column("setting_key", String, primary_key=True), Column("valid_from", Date, primary_key=True),
Column("value", String), Column("set_by", String))
# ── slice B ─────────────────────────────────────────────────────────────────────────────────────────────────
mfg_work_centre = _t("mfg_work_centre", Column("work_centre_id", String, primary_key=True), Column("company_code", String),
Column("warehouse_id", String), Column("cost_centre", String), Column("batch_capacity", Numeric(19, 4)),
Column("capacity_unit_code", String), Column("hours_per_day", Numeric(19, 4)), Column("status", String))
mfg_work_centre_rate = _t("mfg_work_centre_rate", Column("work_centre_id", String, primary_key=True),
Column("rate_kind", String, primary_key=True), Column("valid_from", Date, primary_key=True),
Column("rate_amount", Numeric(19, 4)), Column("currency", String), Column("set_by", String))
mfg_routing = _t("mfg_routing", Column("product_id", String, primary_key=True), Column("routing_version", Integer, primary_key=True),
Column("valid_from", Date), Column("operations_json", String), Column("written_by", String))
mfg_confirmation = _t(
"mfg_confirmation", Column("production_order_id", String, primary_key=True), Column("op_no", Integer, primary_key=True),
Column("confirmation_no", Integer, primary_key=True), Column("work_centre_id", String), Column("machine_minutes", Numeric(19, 4)),
Column("labour_minutes", Numeric(19, 4)), Column("is_final", Boolean), Column("cost", Numeric(18, 2)), Column("journal_id", String),
Column("reverses_confirmation_no", Integer), Column("recorded_by", String), Column("recorded_sub", String),
Column("business_date", Date), Column("status", String), Column("outbox_ref", String))
mfg_scrap_reason = _t("mfg_scrap_reason", Column("reason_code", String, primary_key=True), Column("description", String),
Column("abnormal", Boolean))
mfg_scrap = _t(
"mfg_scrap", Column("production_order_id", String, primary_key=True), Column("scrap_no", Integer, primary_key=True),
Column("product_id", String), Column("quantity", Numeric(19, 4)), Column("unit_code", String),
Column("quantity_base", Numeric(19, 4)), Column("reason_code", String), Column("standard_value", Numeric(18, 2)),
Column("status", String), Column("recorded_by", String), Column("recorded_sub", String), Column("approved_by", String),
Column("approved_sub", String), Column("business_date", Date))
mfg_mrp_run = _t("mfg_mrp_run", Column("run_no", Integer, primary_key=True), Column("warehouse_id", String),
Column("from_date", Date), Column("horizon_weeks", Integer), Column("planned_count", Integer),
Column("stale", Boolean), Column("stale_cause", String), Column("started_by", String))
mfg_planned_order = _t(
"mfg_planned_order", Column("warehouse_id", String, primary_key=True), Column("product_id", String, primary_key=True),
Column("need_date", Date, primary_key=True), Column("run_no", Integer, primary_key=True), Column("quantity", Numeric(19, 4)),
Column("unit_code", String), Column("planned_order_ref", String), Column("status", String), Column("cause_json", String),
Column("production_order_id", String), Column("firmed_by", String), Column("firmed_sub", String))
mfg_dependent_requirement = _t(
"mfg_dependent_requirement", Column("warehouse_id", String, primary_key=True), Column("product_id", String, primary_key=True),
Column("need_date", Date, primary_key=True), Column("source_ref", String, primary_key=True), Column("quantity", Numeric(19, 4)),
Column("unit_code", String), Column("source_kind", String), Column("run_no", Integer))
mfg_forecast = _t("mfg_forecast", Column("warehouse_id", String, primary_key=True), Column("product_id", String, primary_key=True),
Column("week_start", Date, primary_key=True), Column("quantity", Numeric(19, 4)), Column("unit_code", String),
Column("entered_by", String))
mfg_cost_rollup = _t(
"mfg_cost_rollup", Column("rollup_id", String, primary_key=True), Column("company_code", String), Column("effective_on", Date),
Column("status", String), Column("ran_by", String), Column("ran_sub", String), Column("approved_by", String),
Column("approved_sub", String), Column("released_on", Date), Column("stale_cause", String), Column("used_json", String))
mfg_cost_rollup_line = _t(
"mfg_cost_rollup_line", Column("rollup_id", String, primary_key=True), Column("product_id", String, primary_key=True),
Column("bom_id", String), Column("bom_version", Integer), Column("base_unit_code", String), Column("material", Numeric(19, 4)),
Column("conversion", Numeric(19, 4)), Column("by_product_credit", Numeric(19, 4)), Column("total", Numeric(19, 4)))
mfg_ai_finding = _t(
"mfg_ai_finding", Column("production_order_id", String, primary_key=True), Column("finding_no", Integer, primary_key=True),
Column("step_key", String), Column("model", String), Column("inputs_digest", String), Column("finding_json", String),
Column("decision", String), Column("requested_by", String), Column("requested_sub", String), Column("decided_by", String),
Column("decided_sub", String), Column("decided_at", TIMESTAMP))
mfg_migration_run = _t(
"mfg_migration_run", Column("run_id", String, primary_key=True), Column("odoo_db", String), Column("company_code", String),
Column("cutover_date", Date), Column("status", String), Column("doors_json", String), Column("checks_json", String),
Column("summary_json", String), Column("loaded_by", String), Column("signed_by", String), Column("signed_sub", String),
Column("signed_on", Date), Column("discard_reason", String))
mfg_migration_map = _t(
"mfg_migration_map", Column("odoo_db", String, primary_key=True), Column("odoo_model", String, primary_key=True),
Column("odoo_id", String, primary_key=True), Column("our_key", String), Column("run_id", String))
BOM = Aggregate(Part(mfg_bom, ("bom_id",)), Part(mfg_bom_version, ("bom_id", "version")),
Part(mfg_bom_line, ("bom_id", "version", "line_no")), version="row_version")
ORDER = Aggregate(Part(mfg_production_order, ("production_order_id",)), version="row_version")
# The controls and defaults (module.yaml config_rows). Every one is a CONTROL (mfg:settings:controls): the planner and the
# operator they steer never hold it. `kind` decides how a value is read back.
SETTING_DEFS = {
"over_issue_tolerance_pct": ("5", "decimal"), "release_with_shortage": ("false", "bool"),
"output_plausibility_pct": ("110", "decimal"), "scrap_approval_threshold": ("50.00", "decimal"),
"variance_approval_threshold": ("250.00", "decimal"), "business_timezone": ("Europe/London", "tz"),
"valuation_method": ("standard", "method"), "currency": ("GBP", "currency"),
"ai_explain_variance_mode": ("suggest", "aimode"), "mrp_horizon_weeks": ("4", "weeks"),
}
SCRAP_REASONS = {"BURNT": "a roast that went too far", "UNDER_ROASTED": "roasted too little to sell", "SPILLED": "lost in handling"}
def _utc(at: Optional[datetime] = None) -> datetime:
return (at or work.now()).astimezone(timezone.utc).replace(tzinfo=None)
def _fields(cls, row: dict, drop=()):
names = cls.__dataclass_fields__
return cls(**{k: v for k, v in row.items() if k in names and k not in drop})
def _stamped(rows: list, at: datetime) -> list:
return [{**r, "created_at": at, "updated_at": at} for r in rows]
class BomRepository:
def __init__(self, conn):
self.conn = conn
def get(self, bom_id: str) -> Optional[tuple]:
"""(Bom, version, loaded) or None. Hand `loaded` back to save() so only the diff is written."""
loaded = BOM.load(self.conn, {"bom_id": bom_id})
if loaded is None:
return None
r = loaded["mfg_bom"][0]
bom = Bom(r["bom_id"], r["product_id"], r["usage"], r["alternative"] or "", r["company_code"] or "",
[_fields(BomVersion, x, ("bom_id",)) for x in loaded["mfg_bom_version"]],
[_fields(BomLine, x, ("bom_id",)) for x in loaded["mfg_bom_line"]])
return bom, BOM.version_of(loaded), loaded
def save(self, bom: Bom, expected_version: int, loaded=None) -> int:
rows = {"mfg_bom": [{"bom_id": bom.bom_id, "product_id": bom.product_id, "usage": bom.usage,
"alternative": bom.alternative, "company_code": bom.company_code}],
"mfg_bom_version": [{"bom_id": bom.bom_id, **v.__dict__} for v in bom.versions],
"mfg_bom_line": [{"bom_id": bom.bom_id, **x.__dict__} for x in bom.lines]}
return BOM.save(self.conn, rows, expected_version=expected_version,
snapshot=loaded if loaded is not None else BOM.nothing_stored({"bom_id": bom.bom_id}))
def ids(self) -> list:
return sorted(r[0] for r in self.conn.execute(select(mfg_bom.c.bom_id)))
def edges(self, exclude: str) -> dict:
"""product -> {components} over every OTHER production BOM, in ONE statement however many BOMs there are."""
stmt = (select(mfg_bom.c.product_id, mfg_bom_line.c.product_id)
.join(mfg_bom_line, mfg_bom_line.c.bom_id == mfg_bom.c.bom_id)
.where(and_(mfg_bom.c.usage == "production", mfg_bom_line.c.kind == "component", mfg_bom.c.bom_id != exclude)))
out: dict = {}
for parent, child in self.conn.execute(stmt):
out.setdefault(parent, set()).add(child)
return out
class OrderRepository:
def __init__(self, conn):
self.conn = conn
def get(self, order_id: str) -> Optional[tuple]:
loaded = ORDER.load(self.conn, {"production_order_id": order_id})
if loaded is None:
return None
return _fields(ProductionOrder, loaded["mfg_production_order"][0]), ORDER.version_of(loaded), loaded
def save(self, order: ProductionOrder, expected_version: int, loaded=None) -> int:
row = {c.name: getattr(order, c.name) for c in mfg_production_order.columns if hasattr(order, c.name)}
return ORDER.save(self.conn, {"mfg_production_order": [row]}, expected_version=expected_version,
snapshot=loaded if loaded is not None else ORDER.nothing_stored(
{"production_order_id": order.production_order_id}))
def ids(self, status: Optional[str] = None) -> list:
stmt = select(mfg_production_order.c.production_order_id).order_by(mfg_production_order.c.production_order_id)
if status:
stmt = stmt.where(mfg_production_order.c.status == status)
return [r[0] for r in self.conn.execute(stmt)]
def open_for(self, warehouse_id: str, statuses: tuple) -> list:
"""Orders of the warehouse in these statuses: [{production_order_id, product_id, planned_quantity, due_on, status, frozen_bom_json}]."""
t = mfg_production_order
return _rows(self.conn, select(t.c.production_order_id, t.c.product_id, t.c.planned_quantity, t.c.due_on, t.c.status,
t.c.frozen_bom_json).where(and_(t.c.warehouse_id == warehouse_id, t.c.status.in_(statuses)))
.order_by(t.c.due_on, t.c.production_order_id))
def unreleased_due_from(self, product_id: str, day: date) -> list:
"""Created (unreleased) orders of the product that fall due on or after `day`."""
t = mfg_production_order
return [r[0] for r in self.conn.execute(select(t.c.production_order_id).where(and_(
t.c.product_id == product_id, t.c.status == "created", t.c.due_on >= day)))]
def naming_bom(self, bom_id: str, version: int, after: date) -> list:
"""Created orders that name this BOM version and fall due after `after` (the obsolete gate)."""
t = mfg_production_order
stmt = select(t.c.production_order_id).where(and_(t.c.bom_id == bom_id, t.c.bom_version == version,
t.c.status == "created", t.c.due_on > after))
return [r[0] for r in self.conn.execute(stmt)]
class MovementRepository:
def __init__(self, conn):
self.conn = conn
def for_order(self, order_id: str) -> list:
stmt = select(mfg_order_movement).where(mfg_order_movement.c.production_order_id == order_id).order_by(
mfg_order_movement.c.move_ref)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def add(self, rows: list, at: datetime) -> None:
if rows:
self.conn.execute(insert(mfg_order_movement), _stamped(rows, _utc(at)))
def answer(self, updates: list) -> None:
"""What inventory said about each movement (its move id and value, or why it refused): one statement."""
if updates:
self.conn.execute(update(mfg_order_movement).where(mfg_order_movement.c.move_ref == bindparam("b_ref")).values(
inventory_move_id=bindparam("b_move"), value=bindparam("b_value"), status=bindparam("b_status"),
reject_reason=bindparam("b_reason"), updated_at=_utc()),
[{"b_ref": u["move_ref"], "b_move": u.get("inventory_move_id"), "b_value": u.get("value"), "b_status": u["status"],
"b_reason": u.get("reject_reason", "")} for u in updates])
class ClaimRepository:
"""The one guard key per bounded thing (audit habit): every stock or money operation takes the next slot of ITS key with a
plain INSERT in ITS transaction, n = the committed claims already there + 1. Two writers who read the same state claim
the same (scope, key, n) and the loser is a 409 - never a second movement with the same reference."""
def __init__(self, conn):
self.conn = conn
def counts(self, scope: str, keys: list) -> dict:
stmt = (select(mfg_claim.c.claim_key, func.count()).where(and_(mfg_claim.c.claim_scope == scope,
mfg_claim.c.claim_key.in_(keys))).group_by(mfg_claim.c.claim_key))
got = {k: n for k, n in self.conn.execute(stmt)}
return {k: got.get(k, 0) for k in keys}
def take(self, rows: list, by: str, at: datetime) -> None:
if rows:
self.conn.execute(insert(mfg_claim), [{"claim_scope": s, "claim_key": k, "n": n, "ref": ref, "claimed_by": by,
"claimed_at": _utc(at)} for s, k, n, ref in rows])
class OutboxRepository:
def __init__(self, conn):
self.conn = conn
def add(self, rows: list, at: datetime) -> None:
if rows:
self.conn.execute(insert(mfg_outbox), _stamped([{"status": "pending", "attempts": 0, "last_error": "", **r}
for r in rows], _utc(at)))
def pending(self, order_id: Optional[str] = None) -> list:
stmt = select(mfg_outbox).where(mfg_outbox.c.status == "pending").order_by(mfg_outbox.c.production_order_id,
mfg_outbox.c.ref)
if order_id:
stmt = stmt.where(mfg_outbox.c.production_order_id == order_id)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def for_order(self, order_id: str) -> list:
stmt = select(mfg_outbox).where(mfg_outbox.c.production_order_id == order_id).order_by(mfg_outbox.c.ref)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def mark(self, updates: list) -> None:
"""updates: [{target, ref, status, attempts, last_error, sent}] - one statement. A row that is sent or rejected is final:
the WHERE refuses to move it back to pending."""
if updates:
self.conn.execute(update(mfg_outbox).where(and_(
mfg_outbox.c.target == bindparam("b_target"), mfg_outbox.c.ref == bindparam("b_ref"),
mfg_outbox.c.status == "pending")).values(
status=bindparam("b_status"), attempts=bindparam("b_attempts"), last_error=bindparam("b_error"),
sent_at=bindparam("b_sent"), updated_at=_utc()),
[{"b_target": u["target"], "b_ref": u["ref"], "b_status": u["status"], "b_attempts": u["attempts"],
"b_error": u["last_error"], "b_sent": _utc() if u["status"] == "sent" else None} for u in updates])
class FactRepository:
"""Genealogy, settlements, approvals, order numbers: written once, read by the order they belong to."""
def __init__(self, conn):
self.conn = conn
def genealogy(self, rows: list) -> None:
upsert(self.conn, mfg_lot_genealogy, [{**r, "created_at": _utc(), "updated_at": _utc()} for r in rows],
("output_product_id", "output_lot_code", "input_product_id", "input_lot_code", "production_order_id"))
def lots_of(self, order_id: str) -> list:
t = mfg_lot_genealogy
return [dict(r._mapping) for r in self.conn.execute(select(t).where(t.c.production_order_id == order_id))]
def settlements(self, order_id: str) -> list:
t = mfg_settlement
return [dict(r._mapping) for r in self.conn.execute(select(t).where(t.c.production_order_id == order_id).order_by(t.c.settlement_no))]
def add_settlement(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_settlement), _stamped([row], _utc(at)))
def approvals(self, order_id: str) -> list:
t = mfg_variance_approval
return [dict(r._mapping) for r in self.conn.execute(select(t).where(t.c.production_order_id == order_id).order_by(t.c.approval_no))]
def add_approval(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_variance_approval), _stamped([row], _utc(at)))
def next_order_no(self, company_code: str, year: int) -> int:
t = mfg_number_series
return (self.conn.execute(select(func.count()).where(and_(t.c.company_code == company_code, t.c.year == year))).scalar() or 0) + 1
def take_order_no(self, company_code: str, year: int, n: int, order_id: str, at: datetime) -> None:
self.conn.execute(insert(mfg_number_series).values(company_code=company_code, year=year, n=n,
production_order_id=order_id, claimed_at=_utc(at)))
class SettingsRepository:
"""Dated controls. A value is in force from its valid_from; the latest row not after today wins; no row means the default."""
def __init__(self, conn):
self.conn = conn
def rows(self) -> list:
stmt = select(mfg_settings).order_by(mfg_settings.c.setting_key, mfg_settings.c.valid_from)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def all(self, today: Optional[date] = None) -> dict:
today = today or work.now().date()
out = {k: v[0] for k, v in SETTING_DEFS.items()}
for r in self.rows():
if r["valid_from"] <= today:
out[r["setting_key"]] = r["value"]
return out
def timezone(self) -> str:
"""THE port for business_timezone: 'today' is the date in this zone. The row in force counts the zone BEFORE it, so a row
dated tomorrow cannot take effect by the UTC date."""
name = SETTING_DEFS["business_timezone"][0]
now = work.now()
for r in self.rows():
if r["setting_key"] == "business_timezone" and r["valid_from"] <= now.astimezone(ZoneInfo(name)).date():
name = r["value"]
return name
def set(self, key: str, value: str, valid_from: date, by: str, at: datetime) -> None:
"""A plain INSERT: the same key from the same day twice is a 409, never a quiet overwrite."""
self.conn.execute(insert(mfg_settings).values(setting_key=key, valid_from=valid_from, value=value, set_by=by,
created_at=_utc(at), updated_at=_utc(at)))
def _rows(conn, stmt) -> list:
return [dict(r._mapping) for r in conn.execute(stmt)]
class MasterRepository:
"""Work centres, their dated rates and the routings (written once per version: a routing is never edited)."""
def __init__(self, conn):
self.conn = conn
def work_centre(self, wc_id: str) -> Optional[dict]:
got = _rows(self.conn, select(mfg_work_centre).where(mfg_work_centre.c.work_centre_id == wc_id))
return got[0] if got else None
def work_centres(self) -> list:
return _rows(self.conn, select(mfg_work_centre).order_by(mfg_work_centre.c.work_centre_id))
def add_work_centre(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_work_centre), _stamped([row], _utc(at)))
def rates(self, wc_ids: Optional[list] = None) -> list:
t = mfg_work_centre_rate
stmt = select(t).order_by(t.c.work_centre_id, t.c.rate_kind, t.c.valid_from)
if wc_ids is not None:
stmt = stmt.where(t.c.work_centre_id.in_(wc_ids))
return _rows(self.conn, stmt)
def add_rates(self, rows: list, at: datetime) -> None:
if rows:
self.conn.execute(insert(mfg_work_centre_rate), _stamped(rows, _utc(at)))
def routings(self, product_id: str) -> list:
t = mfg_routing
return _rows(self.conn, select(t).where(t.c.product_id == product_id).order_by(t.c.routing_version))
def routed_products(self) -> list:
return sorted(r[0] for r in self.conn.execute(select(mfg_routing.c.product_id).distinct()))
def add_routing(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_routing), _stamped([row], _utc(at)))
def routings_using(self, wc_id: str) -> list:
t = mfg_routing
return sorted({r["product_id"] for r in _rows(self.conn, select(t)) if f'"{wc_id}"' in (r["operations_json"] or "")})
class ShopFloorRepository:
"""Confirmations and scrap on an order, and the scrap reasons (the defaults, then the rows over them)."""
def __init__(self, conn):
self.conn = conn
def confirmations(self, order_id: str) -> list:
t = mfg_confirmation
return _rows(self.conn, select(t).where(t.c.production_order_id == order_id).order_by(t.c.op_no, t.c.confirmation_no))
def add_confirmation(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_confirmation), _stamped([row], _utc(at)))
def activity_cost(self, order_id: str) -> Decimal:
"""What the order's confirmations cost, net of mirrors; a confirmation gl refused (void) costs nothing."""
return sum((c["cost"] or Decimal("0") for c in self.confirmations(order_id) if c["status"] != "void"), Decimal("0"))
def void_by_ref(self, refs: list) -> None:
"""gl refused the journal of these confirmations: they are void and count for nothing (the row stays, as a movement does)."""
if refs:
t = mfg_confirmation
self.conn.execute(update(t).where(t.c.outbox_ref.in_(refs)).values(status="void", updated_at=_utc()))
def void_scrap(self, order_id: str, scrap_no: int) -> None:
t = mfg_scrap
self.conn.execute(update(t).where(and_(t.c.production_order_id == order_id, t.c.scrap_no == scrap_no,
t.c.status.in_(("pending_approval", "approved")))).values(status="void", updated_at=_utc()))
def scraps(self, order_id: str) -> list:
t = mfg_scrap
return _rows(self.conn, select(t).where(t.c.production_order_id == order_id).order_by(t.c.scrap_no))
def add_scrap(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_scrap), _stamped([row], _utc(at)))
def decide_scrap(self, order_id: str, scrap_no: int, status: str, by: str, sub: str) -> None:
"""pending_approval -> approved (the WHERE refuses any other move: a decided scrap is final)."""
t = mfg_scrap
self.conn.execute(update(t).where(and_(t.c.production_order_id == order_id, t.c.scrap_no == scrap_no,
t.c.status == "pending_approval")).values(
status=status, approved_by=by, approved_sub=sub, updated_at=_utc()))
def reasons(self) -> dict:
"""{reason_code: abnormal}: the defaults, overridden by rows."""
out = {k: True for k in SCRAP_REASONS}
for r in _rows(self.conn, select(mfg_scrap_reason)):
out[r["reason_code"]] = bool(r["abnormal"])
return out
def set_reason(self, code: str, description: str, abnormal: bool, at: datetime) -> None:
upsert(self.conn, mfg_scrap_reason, [{"reason_code": code, "description": description, "abnormal": abnormal,
"created_at": _utc(at), "updated_at": _utc(at)}], ("reason_code",))
class PlanRepository:
"""MRP-lite: forecasts, runs, planned orders and the dependent requirements each run publishes."""
def __init__(self, conn):
self.conn = conn
def forecasts(self, warehouse_id: str, start: date, end: date) -> list:
t = mfg_forecast
return _rows(self.conn, select(t).where(and_(t.c.warehouse_id == warehouse_id, t.c.week_start >= start,
t.c.week_start <= end)).order_by(t.c.product_id, t.c.week_start))
def set_forecasts(self, rows: list, at: datetime) -> None:
upsert(self.conn, mfg_forecast, [{**r, "created_at": _utc(at), "updated_at": _utc(at)} for r in rows],
("warehouse_id", "product_id", "week_start"))
def last_run(self) -> Optional[dict]:
got = _rows(self.conn, select(mfg_mrp_run).order_by(mfg_mrp_run.c.run_no.desc()).limit(1))
return got[0] if got else None
def add_run(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_mrp_run), _stamped([row], _utc(at)))
def mark_stale(self, cause: str) -> None:
"""The latest run is out of date (a flag; a run is a pure function of its inputs, so a re-run is the fix): one statement."""
t = mfg_mrp_run
self.conn.execute(update(t).where(and_(t.c.run_no == select(func.max(t.c.run_no)).scalar_subquery(), t.c.stale == False)).values( # noqa: E712
stale=True, stale_cause=cause[:500], updated_at=_utc()))
def planned(self, warehouse_id: Optional[str] = None, statuses: Optional[tuple] = None, product_id: Optional[str] = None) -> list:
t = mfg_planned_order
stmt = select(t).order_by(t.c.need_date, t.c.product_id, t.c.run_no)
if warehouse_id:
stmt = stmt.where(t.c.warehouse_id == warehouse_id)
if statuses:
stmt = stmt.where(t.c.status.in_(statuses))
if product_id:
stmt = stmt.where(t.c.product_id == product_id)
return _rows(self.conn, stmt)
def planned_by_ref(self, ref: str) -> Optional[dict]:
got = _rows(self.conn, select(mfg_planned_order).where(mfg_planned_order.c.planned_order_ref == ref))
return got[0] if got else None
def add_planned(self, rows: list, at: datetime) -> None:
if rows:
self.conn.execute(insert(mfg_planned_order), _stamped(rows, _utc(at)))
def supersede(self, warehouse_id: str) -> int:
"""Every PROPOSED order of the warehouse is replaced by the new run; firmed and converted ones are kept."""
t = mfg_planned_order
return self.conn.execute(update(t).where(and_(t.c.warehouse_id == warehouse_id, t.c.status == "proposed")).values(
status="superseded", updated_at=_utc())).rowcount
def move_planned(self, key: dict, frm: str, to: str, **more) -> None:
"""proposed -> firmed -> converted: forward only (the WHERE refuses any other move)."""
t = mfg_planned_order
self.conn.execute(update(t).where(and_(t.c.warehouse_id == key["warehouse_id"], t.c.product_id == key["product_id"],
t.c.need_date == key["need_date"], t.c.run_no == key["run_no"],
t.c.status == frm)).values(status=to, updated_at=_utc(), **more))
def add_requirements(self, rows: list, at: datetime) -> None:
if rows:
self.conn.execute(insert(mfg_dependent_requirement), _stamped(rows, _utc(at)))
def requirements(self, product_id: str, warehouse_id: str, start: date, end: date, statuses: tuple) -> list:
"""The dependent requirements of planned orders still live (their planned order is in `statuses`) in [start, end]."""
r, p = mfg_dependent_requirement, mfg_planned_order
stmt = (select(r.c.need_date, r.c.quantity, r.c.unit_code, r.c.source_ref)
.join(p, and_(p.c.planned_order_ref == r.c.source_ref, p.c.warehouse_id == r.c.warehouse_id))
.where(and_(r.c.product_id == product_id, r.c.warehouse_id == warehouse_id, r.c.need_date >= start,
r.c.need_date <= end, p.c.status.in_(statuses))).order_by(r.c.need_date, r.c.source_ref))
return _rows(self.conn, stmt)
class CostRepository:
"""Cost roll-ups: a proposal (computed) until a second person releases it; stale if an input changed first; never edited after."""
def __init__(self, conn):
self.conn = conn
def get(self, rollup_id: str) -> Optional[dict]:
got = _rows(self.conn, select(mfg_cost_rollup).where(mfg_cost_rollup.c.rollup_id == rollup_id))
return got[0] if got else None
def ids(self) -> list:
return sorted(r[0] for r in self.conn.execute(select(mfg_cost_rollup.c.rollup_id)))
def lines(self, rollup_id: str) -> list:
t = mfg_cost_rollup_line
return _rows(self.conn, select(t).where(t.c.rollup_id == rollup_id).order_by(t.c.product_id))
def add(self, row: dict, lines: list, at: datetime) -> None:
self.conn.execute(insert(mfg_cost_rollup), _stamped([row], _utc(at)))
if lines:
self.conn.execute(insert(mfg_cost_rollup_line), _stamped(lines, _utc(at)))
def open_ones(self) -> list:
return _rows(self.conn, select(mfg_cost_rollup).where(mfg_cost_rollup.c.status == "computed"))
def mark_stale(self, ids: list, cause: str) -> None:
"""computed -> stale only: a released roll-up never changes."""
if ids:
t = mfg_cost_rollup
self.conn.execute(update(t).where(and_(t.c.rollup_id.in_(ids), t.c.status == "computed")).values(
status="stale", stale_cause=cause[:500], updated_at=_utc()))
def release(self, rollup_id: str, by: str, sub: str, day: date) -> None:
t = mfg_cost_rollup
self.conn.execute(update(t).where(and_(t.c.rollup_id == rollup_id, t.c.status == "computed")).values(
status="released", approved_by=by, approved_sub=sub, released_on=day, updated_at=_utc()))
class AiRepository:
"""The audit of the explain_variance step: what was asked, what the model said, what the accountant decided."""
def __init__(self, conn):
self.conn = conn
def findings(self, order_id: str) -> list:
t = mfg_ai_finding
return _rows(self.conn, select(t).where(t.c.production_order_id == order_id).order_by(t.c.finding_no))
def add(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_ai_finding), _stamped([row], _utc(at)))
def decide(self, order_id: str, no: int, decision: str, by: str, sub: str, at: datetime) -> None:
"""pending -> accepted | dismissed once (the WHERE refuses to change a decision)."""
t = mfg_ai_finding
self.conn.execute(update(t).where(and_(t.c.production_order_id == order_id, t.c.finding_no == no,
t.c.decision == "pending")).values(
decision=decision, decided_by=by, decided_sub=sub, decided_at=_utc(at), updated_at=_utc()))
def recent_settlements(self, product_id: str, before: str, limit: int = 20) -> list:
"""The same product's last closed orders with what each one's settlement found: ONE statement."""
o, st = mfg_production_order, mfg_settlement
stmt = (select(o.c.production_order_id, o.c.planned_quantity, st.c.material_usage_variance, st.c.activity_variance, st.c.abnormal_scrap)
.join(st, st.c.production_order_id == o.c.production_order_id)
.where(and_(o.c.product_id == product_id, o.c.production_order_id != before, o.c.status == "closed", st.c.settlement_no == 1))
.order_by(o.c.closed_on.desc(), o.c.production_order_id.desc()).limit(limit))
return _rows(self.conn, stmt)
class TraceRepository:
"""Genealogy edges one level at a time: a trace is a walk, one statement per level whatever the number of lots."""
def __init__(self, conn):
self.conn = conn
def outputs_of(self, lots: list) -> list:
t = mfg_lot_genealogy
if not lots:
return []
keys = [f"{p}\x1f{c}" for p, c in lots]
stmt = select(t).where((t.c.input_product_id + "\x1f" + t.c.input_lot_code).in_(keys))
return _rows(self.conn, stmt)
def inputs_of(self, lots: list) -> list:
t = mfg_lot_genealogy
if not lots:
return []
keys = [f"{p}\x1f{c}" for p, c in lots]
stmt = select(t).where((t.c.output_product_id + "\x1f" + t.c.output_lot_code).in_(keys))
return _rows(self.conn, stmt)
class MigrationRepository:
def __init__(self, conn):
self.conn = conn
def run(self, run_id: str) -> Optional[dict]:
got = _rows(self.conn, select(mfg_migration_run).where(mfg_migration_run.c.run_id == run_id))
return got[0] if got else None
def runs(self) -> list:
return _rows(self.conn, select(mfg_migration_run).order_by(mfg_migration_run.c.run_id))
def add_run(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(mfg_migration_run), _stamped([row], _utc(at)))
def update_run(self, run_id: str, **values) -> None:
self.conn.execute(update(mfg_migration_run).where(mfg_migration_run.c.run_id == run_id).values(updated_at=_utc(), **values))
def mapped(self, odoo_db: str, model: str) -> dict:
t = mfg_migration_map
return {r["odoo_id"]: r["our_key"] for r in _rows(self.conn, select(t).where(and_(t.c.odoo_db == odoo_db, t.c.odoo_model == model)))}
def migrated_outputs(self, odoo_db: str) -> list:
"""The output movements of orders this migration carried: [{product_id, business_date, quantity_base}] - one statement."""
t = mfg_order_movement
stmt = select(t.c.product_id, t.c.business_date, t.c.quantity_base).where(and_(
t.c.kind == "output", t.c.production_order_id.like(f"odoo:{odoo_db}:%")))
return _rows(self.conn, stmt)
def migrated_genealogy(self, odoo_db: str) -> list:
t = mfg_lot_genealogy
return _rows(self.conn, select(t).where(t.c.production_order_id.like(f"odoo:{odoo_db}:%")))
def add_map(self, rows: list, at: datetime) -> None:
if rows:
self.conn.execute(insert(mfg_migration_map), _stamped(rows, _utc(at)))
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.boms, self.orders = BomRepository(conn), OrderRepository(conn)
self.movements, self.claims, self.outbox = MovementRepository(conn), ClaimRepository(conn), OutboxRepository(conn)
self.facts, self.settings = FactRepository(conn), SettingsRepository(conn)
self.masters, self.floor, self.plan = MasterRepository(conn), ShopFloorRepository(conn), PlanRepository(conn)
self.costs, self.ai, self.trace, self.migration = CostRepository(conn), AiRepository(conn), TraceRepository(conn), MigrationRepository(conn)
def seed(self) -> None:
"""Nothing to seed: a setting with no row is its default."""
def dump(obj) -> str:
return json.dumps(obj, sort_keys=True, default=str)