dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/infrastructure/tables.py
"""The tables, and one repository per aggregate that loads and saves it whole.
Every table is prefixed `pur_`, has a natural primary key (never SERIAL) and, for a part, a key that starts with its root's. Money is
NUMERIC(18,2); a price or rate NUMERIC(19,4); a quantity NUMERIC(19,6). The aggregates' tables are generated from the domain dataclasses
(one column model), so a field and its column cannot drift; schema.sql is generated from the same model (scripts/gen_schema.py) and
tests/test_schema.py keeps them equal.
Concurrency: an aggregate is saved at the version it was loaded (a plain INSERT into `<root>_versions` claims it); anything that must
not exceed a total takes a claim row with a plain INSERT in the same transaction (`Claims`): two writers that read the same snapshot
both try the same key and exactly one commits.
"""
import contextlib
import contextvars
import dataclasses
import json
import typing
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from erp_platform.repository import Aggregate, Conflict, Part, upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
func,
insert,
select,
)
from sqlalchemy.exc import IntegrityError
from brick_purchasing.domain import approval, bill, order, payment, receipt, request, rows
from brick_purchasing.domain.values import EPOCH, SETTING_DEFAULTS, SETTINGS
md = MetaData()
MONEY, RATE, QTY = Numeric(18, 2), Numeric(19, 4), Numeric(19, 6)
STAMPS = ("created_at", "updated_at")
BIGINTS = {"bp_id", "ship_to_bp_id", "payee_bp_id", "awarded_bp_id", "odoo_id", "supplier_bp_id"}
RATES = {"unit_price", "unit_cost", "tax_rate", "rate_pct", "cis_rate_pct", "lower_pct", "upper_pct", "quota_pct", "percent"}
QUANTITIES = {"quantity", "quantity_base", "quantity_priced", "provided", "order_multiple", "target_quantity", "min_quantity",
"catch_quantity", "base_quantity", "max_quantity", "quantity_per"}
def _coltype(name: str, tp):
origin = typing.get_origin(tp)
if origin is typing.Union:
tp = next(a for a in typing.get_args(tp) if a is not type(None))
if tp is bool:
return Boolean
if tp is int:
return BigInteger if name in BIGINTS else Integer
if tp is Decimal:
return RATE if name in RATES else QTY if name in QUANTITIES else MONEY
if tp is date:
return Date
if tp is datetime:
return TIMESTAMP
return String
def table_of(name: str, cls, key: tuple, *, version: bool = False, extra: tuple = (), skip: tuple = ()) -> Table:
"""A table from a dataclass: its scalar fields in order, `key` as the primary key, then (a root) `version`, the stamps, and any
appended columns. List-typed fields (an aggregate's parts) and `events` are not columns."""
hints = typing.get_type_hints(cls)
cols = []
for f in dataclasses.fields(cls):
if f.name in skip or typing.get_origin(hints[f.name]) is list or hints[f.name] is list or f.name == "events":
continue
cols.append(Column(f.name, _coltype(f.name, hints[f.name]), primary_key=f.name in key))
cols += list(extra)
if version:
cols.append(Column("version", Integer))
return Table(name, md, *cols, Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
def plain(name: str, key: tuple, *cols: Column) -> Table:
"""A table of columns written by hand: master data, dated config rows."""
cs = list(cols)
for c in cs:
if c.name in key:
c.primary_key = True
return Table(name, md, *cs, Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
def C(name, t=String):
return Column(name, t)
# ── aggregates ────────────────────────────────────────────────────────────────────────────────
pur_request = table_of("pur_request", request.PurchaseRequest, ("purchase_request_id",), version=True)
pur_request_line = table_of("pur_request_line", request.RequestLine, ("purchase_request_id", "line_no"))
pur_release = table_of("pur_release", approval.Release, ("document_kind", "document_id", "revision", "level"))
pur_order = table_of("pur_order", order.PurchaseOrder, ("purchase_order_id",), version=True)
pur_order_line = table_of("pur_order_line", order.OrderLine, ("purchase_order_id", "line_no"))
pur_order_revision = table_of("pur_order_revision", order.OrderRevision, ("purchase_order_id", "revision"))
pur_order_send = table_of("pur_order_send", order.OrderSend, ("purchase_order_id", "sent_at"))
pur_gate_override = table_of("pur_gate_override", order.GateOverride, ("purchase_order_id", "revision", "line_no", "gate"))
pur_order_component = table_of("pur_order_component", order.OrderComponent, ("purchase_order_id", "line_no", "component_no"))
pur_delivery_cost = table_of("pur_delivery_cost", order.DeliveryCost, ("purchase_order_id", "line_no", "kind", "seq"))
pur_receipt = table_of("pur_receipt", receipt.GoodsReceipt, ("receipt_id",), version=True)
pur_receipt_line = table_of("pur_receipt_line", receipt.ReceiptLine, ("receipt_id", "line_no"))
pur_bill = table_of("pur_bill", bill.SupplierBill, ("bill_id",), version=True)
pur_bill_line = table_of("pur_bill_line", bill.BillLine, ("bill_id", "line_no"))
pur_variance_acceptance = table_of("pur_variance_acceptance", bill.VarianceAcceptance, ("bill_id", "line_no", "kind"))
pur_billed_claim = table_of("pur_billed_claim", bill.BilledClaim, ("purchase_order_id", "line_no", "bill_id"))
pur_payment_run = table_of("pur_payment_run", payment.PaymentRun, ("run_id",), version=True)
pur_run_item = table_of("pur_run_item", payment.RunItem, ("run_id", "payment_id", "bill_id"))
pur_payment = table_of("pur_payment", payment.SupplierPayment, ("payment_id",), version=True)
pur_allocation = table_of("pur_allocation", payment.Allocation, ("source_kind", "source_id", "bill_id", "seq"))
pur_rfq = table_of("pur_rfq", rows.Rfq, ("rfq_id",), version=True)
pur_rfq_line = table_of("pur_rfq_line", rows.RfqLine, ("rfq_id", "line_no"))
pur_rfq_supplier = table_of("pur_rfq_supplier", rows.RfqInvitation, ("rfq_id", "bp_id"))
pur_quote = table_of("pur_quote", rows.SupplierQuote, ("rfq_id", "bp_id", "revision"))
pur_agreement = table_of("pur_agreement", rows.PurchaseAgreement, ("agreement_id",), version=True)
pur_agreement_line = table_of("pur_agreement_line", rows.AgreementLine, ("agreement_id", "line_no"))
pur_schedule_line = table_of("pur_schedule_line", rows.ScheduleLine, ("agreement_id", "line_no", "revision", "delivery_date"))
pur_schedule_release = table_of("pur_schedule_release", rows.ScheduleRelease, ("agreement_id", "release_no"))
pur_agreement_claim = plain("pur_agreement_claim", ("agreement_id", "line_no", "purchase_order_id", "order_line_no"),
C("agreement_id"), C("line_no", Integer), C("purchase_order_id"), C("order_line_no", Integer),
C("quantity_base", QTY), C("value", MONEY), C("claimed_at", TIMESTAMP))
pur_down_payment = table_of("pur_down_payment", rows.DownPayment, ("down_payment_id",), version=True)
pur_dp_clearing = table_of("pur_dp_clearing", rows.DownPaymentClearing, ("down_payment_id", "bill_id", "seq"))
# ── claims: a plain INSERT of a key two racing writers both want, so exactly one commits ──────────────
pur_claim_versions = plain("pur_claim_versions", ("scope", "seq"), C("scope"), C("seq", Integer), C("claimed_by"), C("claimed_at", TIMESTAMP))
pur_number_series = plain("pur_number_series", ("series_id", "year"), C("series_id"), C("year", Integer), C("format"),
C("high_water_mark", Integer), C("version", Integer))
pur_number_claim = plain("pur_number_claim", ("number",), C("number"), C("series_id"), C("document_id"), C("claimed_at", TIMESTAMP))
# ── master data and dated config rows ─────────────────────────────────────────────────────────
pur_settings = plain("pur_settings", ("setting_key", "valid_from"), C("setting_key"), C("valid_from", Date), C("value"), C("set_by"),
C("set_at", TIMESTAMP), C("changes"))
pur_approval_limit = plain("pur_approval_limit", ("approver_email", "category", "valid_from"), C("approver_email"), C("category"),
C("valid_from", Date), C("approver_sub"), C("limit_amount", MONEY), C("currency"), C("set_by_email"))
pur_release_strategy = plain("pur_release_strategy", ("category", "min_amount", "valid_from"), C("category"), C("min_amount", MONEY),
C("valid_from", Date), C("levels", Integer), C("level_thresholds"), C("currency"), C("set_by_email"))
pur_tolerance = plain("pur_tolerance", ("tolerance_group", "key", "valid_from"), C("tolerance_group"), C("key"), C("valid_from", Date),
C("lower_amount", MONEY), C("upper_amount", MONEY), C("lower_pct", RATE), C("upper_pct", RATE), C("set_by_email"))
pur_terms = plain("pur_terms", ("terms_id",), C("terms_id"), C("name"), C("days", Integer), C("end_of_month", Boolean),
C("discount_pct", RATE), C("discount_days", Integer))
pur_account_map = plain("pur_account_map", ("purpose", "category", "valid_from"), C("purpose"), C("category"), C("valid_from", Date),
C("account_id"), C("set_by_email"))
pur_supplier = plain("pur_supplier", ("bp_id", "valid_from"), C("bp_id", BigInteger), C("valid_from", Date), C("terms_id"), C("currency"),
C("payment_method"), C("remittance_email"), C("reclaims_vat", Boolean), C("tolerance_group"),
C("cis_subcontractor", Boolean), C("ers", Boolean), C("legal_name"), C("country"), C("set_by_email"))
pur_supplier_bank = table_of("pur_supplier_bank", payment.SupplierBankRow, ("bp_id", "valid_from"))
pur_supplier_price = plain("pur_supplier_price", ("bp_id", "product_id", "unit_code", "min_quantity", "valid_from"),
C("bp_id", BigInteger), C("product_id"), C("unit_code"), C("min_quantity", QTY), C("valid_from", Date),
C("unit_price", RATE), C("pricing_unit_code"), C("pricing_quantity", Integer), C("pricing_numerator", Integer),
C("pricing_denominator", Integer), C("order_multiple", QTY), C("currency"), C("lead_days", Integer),
C("supplier_product_code"), C("supplier_product_name"), C("valid_to", Date), C("consignment", Boolean),
C("set_by_email"))
pur_source = plain("pur_source", ("product_id", "bp_id", "valid_from"), C("product_id"), C("bp_id", BigInteger), C("valid_from", Date),
C("fixed", Boolean), C("blocked", Boolean), C("valid_to", Date), C("agreement_id"), C("required", Boolean),
C("set_by_email"))
pur_quota = plain("pur_quota", ("product_id", "valid_from", "bp_id"), C("product_id"), C("valid_from", Date), C("bp_id", BigInteger),
C("quota_pct", RATE), C("base_quantity", QTY), C("max_quantity", QTY), C("set_by_email"))
pur_evaluation = plain("pur_evaluation", ("bp_id", "period_id", "seq"), C("bp_id", BigInteger), C("period_id"), C("seq", Integer),
C("price_score", Integer), C("delivery_score", Integer), C("quantity_score", Integer), C("quality_score", Integer),
C("service_score", Integer), C("overall", RATE), C("weights"), C("frozen_by_email"), C("frozen_on", Date), C("reason"))
pur_evaluation_weight = plain("pur_evaluation_weight", ("criterion", "valid_from"), C("criterion"), C("valid_from", Date),
C("weight", Integer), C("set_by_email"))
pur_service_score = plain("pur_service_score", ("bp_id", "period_id"), C("bp_id", BigInteger), C("period_id"), C("score", Integer),
C("set_by_email"))
pur_self_billing = plain("pur_self_billing", ("bp_id", "valid_from"), C("bp_id", BigInteger), C("valid_from", Date), C("valid_to", Date),
C("document_key"), C("supplier_vat_number"), C("status"), C("entered_by_email"), C("entered_by_sub"),
C("approved_by_email"), C("approved_by_sub"), C("end_reason"))
pur_consignment_withdrawal = plain("pur_consignment_withdrawal", ("withdrawal_id",), C("withdrawal_id"), C("bp_id", BigInteger),
C("product_id"), C("quantity", QTY), C("unit_code"), C("withdrawn_on", Date), C("unit_price", RATE),
C("pricing_unit_code"), C("pricing_quantity", Integer), C("amount", MONEY), C("currency"),
C("status"), C("settled_bill_id"), C("reverses_withdrawal_id"), C("move_ref"))
pur_cis_deduction = plain("pur_cis_deduction", ("payment_id", "bill_id"), C("payment_id"), C("bill_id"), C("labour_paid", MONEY),
C("materials_paid", MONEY), C("rate_pct", RATE), C("deduction", MONEY), C("tax_month"), C("verification_number"),
C("status"), C("deduction_ref"))
pur_ai_finding = plain("pur_ai_finding", ("subject_ref", "step", "run_no", "finding_no"), C("subject_ref"), C("step"), C("run_no", Integer),
C("finding_no", Integer), C("model"), C("inputs_digest"), C("finding"), C("evidence"), C("decision"),
C("decided_by_email"), C("decided_at", TIMESTAMP))
pur_task = plain("pur_task", ("task_id",), C("task_id"), C("capability"), C("subject_ref"), C("plan"), C("reason"), C("status"))
pur_handover = plain("pur_handover", ("handover_id",), C("handover_id"), C("kind"), C("subject_ref"), C("status"), C("reason"),
C("attempts", Integer), C("payload"), C("last_at", TIMESTAMP))
pur_object = plain("pur_object", ("object_key",), C("object_key"), C("content_type"), C("body"), C("sha256"))
# the Odoo id map (shared.yaml conventions.odoo_migration.id_map) and the run that wrote it
pur_migration_map = plain("pur_migration_map", ("odoo_db", "odoo_model", "odoo_id"), C("odoo_db"), C("odoo_model"), C("odoo_id", BigInteger),
C("natural_key"), C("run_id"), C("migrated_at", TIMESTAMP))
pur_migration_run = plain("pur_migration_run", ("run_id",), C("run_id"), C("odoo_db"), C("cutover_date", Date), C("status"), C("doors"),
C("checksums"), C("loaded_at", TIMESTAMP), C("reconciled_at", TIMESTAMP), C("signed_by_email"), C("signed_by_sub"),
C("signed_at", TIMESTAMP), C("discard_reason"), C("version", Integer))
pur_migration_check = plain("pur_migration_check", ("run_id", "check_no"), C("run_id"), C("check_no", Integer), C("name"), C("odoo"),
C("ours"), C("equal", Boolean), C("must"), C("explanation"), C("explained_by_email"))
# ── building and saving aggregates ────────────────────────────────────────────────────────────
def _cols(table: Table) -> tuple:
return tuple(c.name for c in table.columns if c.name not in STAMPS and c.name != "version")
def _make(cls, row: dict):
names = {f.name for f in dataclasses.fields(cls)}
return cls(**{k: v for k, v in row.items() if k in names and v is not None})
def _pick(obj, table: Table) -> dict:
return {c: getattr(obj, c) for c in _cols(table)}
def _sorted(rows_: list, key: tuple) -> list:
return sorted(rows_, key=lambda r: tuple((0, r[k]) if isinstance(r[k], (int, Decimal)) else (1, str(r[k])) for k in key))
class Spec:
"""One aggregate's shape: a root dataclass and table, its list-typed parts (attribute, table, dataclass), and whether it is versioned.
Built at import, so the `<root>_versions` claim tables exist in `md` before any test or migration creates the schema."""
def __init__(self, cls, root: Table, parts: tuple = (), versioned: bool = True):
self.cls, self.root, self.parts, self.versioned = cls, root, parts, versioned
self.rootkey = tuple(c.name for c in root.primary_key.columns)
key = lambda t: tuple(c.name for c in t.primary_key.columns) # noqa: E731
self.agg = Aggregate(Part(root, key(root)), *[Part(t, key(t)) for _, t, _ in parts], version="version" if versioned else None)
class AggRepo:
"""Load and save one aggregate whole: a root dataclass and its list-typed parts, each a table whose key starts with the root's."""
def __init__(self, conn, spec: Spec):
self.conn, self.spec = conn, spec
self.cls, self.root, self.rootkey, self.parts, self.versioned, self.agg = spec.cls, spec.root, spec.rootkey, spec.parts, spec.versioned, spec.agg
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, *key) -> Optional[tuple]:
rows_ = self.agg.load(self.conn, dict(zip(self.rootkey, key)))
if rows_ is None:
return None
self._snapshots[tuple(key)] = rows_.snapshot
return self._build(rows_)
def get_many(self, keys) -> dict:
"""{key: (aggregate, version)} for a set of root keys (a scalar for a one-column key): one query per table per 100 roots, in one round trip."""
keys = list(dict.fromkeys(k if isinstance(k, tuple) else (k,) for k in keys))
out = {}
for key, rows_ in self.agg.load_many(self.conn, [dict(zip(self.rootkey, k)) for k in keys]).items():
if rows_ is None:
continue
self._snapshots[tuple(key)] = rows_.snapshot
out[key[0] if len(key) == 1 else tuple(key)] = self._build(rows_)
return out
def _build(self, rows_) -> tuple:
obj = _make(self.cls, rows_[self.root.name][0])
for attr, t, child in self.parts:
setattr(obj, attr, [_make(child, r) for r in _sorted(rows_[t.name], tuple(c.name for c in t.primary_key.columns))])
return obj, self.agg.version_of(rows_)
def save(self, obj, expected_version: int = 0) -> int:
snapshot = self._snapshots.pop(tuple(getattr(obj, k) for k in self.rootkey), None) # spent: after this save the rows are no longer what it read
data = {self.root.name: [_pick(obj, self.root)]}
for attr, t, _ in self.parts:
data[t.name] = [_pick(x, t) for x in getattr(obj, attr)]
return self.agg.save(self.conn, data, expected_version=expected_version if self.versioned else None, snapshot=snapshot)
def exists(self, *key) -> bool:
cond = [self.root.c[k] == v for k, v in zip(self.rootkey, key)]
return self.conn.execute(select(self.root.c[self.rootkey[0]]).where(*cond)).first() is not None
def heads(self, *where, order=None, limit: int = 50, offset: int = 0) -> tuple:
total = self.conn.execute(select(func.count()).select_from(self.root).where(*where)).scalar() or 0
stmt = select(self.root).where(*where).order_by(*(order if order is not None else [self.root.c[self.rootkey[0]]])).limit(limit).offset(offset)
return [_make(self.cls, dict(r._mapping)) for r in self.conn.execute(stmt)], int(total)
def all_heads(self, *where) -> list:
return [_make(self.cls, dict(r._mapping)) for r in self.conn.execute(select(self.root).where(*where))]
@dataclasses.dataclass
class _Series:
series_id: str
year: int
format: str = ""
high_water_mark: int = 0
def claim(self) -> int:
self.high_water_mark += 1
return self.high_water_mark
def raise_to(self, mark: int) -> None:
self.high_water_mark = max(self.high_water_mark, mark)
NUMBER_SERIES = Spec(_Series, pur_number_series)
REQUEST = Spec(request.PurchaseRequest, pur_request, (("lines", pur_request_line, request.RequestLine),))
ORDER = Spec(order.PurchaseOrder, pur_order, (
("lines", pur_order_line, order.OrderLine), ("revisions", pur_order_revision, order.OrderRevision), ("sends", pur_order_send, order.OrderSend),
("overrides", pur_gate_override, order.GateOverride), ("components", pur_order_component, order.OrderComponent),
("delivery_costs", pur_delivery_cost, order.DeliveryCost)))
RECEIPT = Spec(receipt.GoodsReceipt, pur_receipt, (("lines", pur_receipt_line, receipt.ReceiptLine),))
BILL = Spec(bill.SupplierBill, pur_bill, (("lines", pur_bill_line, bill.BillLine), ("acceptances", pur_variance_acceptance, bill.VarianceAcceptance)))
RUN = Spec(payment.PaymentRun, pur_payment_run, (("items", pur_run_item, payment.RunItem),))
PAYMENT = Spec(payment.SupplierPayment, pur_payment)
RFQ = Spec(rows.Rfq, pur_rfq, (("lines", pur_rfq_line, rows.RfqLine), ("invitations", pur_rfq_supplier, rows.RfqInvitation), ("quotes", pur_quote, rows.SupplierQuote)))
AGREEMENT = Spec(rows.PurchaseAgreement, pur_agreement, (("lines", pur_agreement_line, rows.AgreementLine), ("schedule", pur_schedule_line, rows.ScheduleLine),
("releases_sent", pur_schedule_release, rows.ScheduleRelease)))
DOWN_PAYMENT = Spec(rows.DownPayment, pur_down_payment, (("clearings", pur_dp_clearing, rows.DownPaymentClearing),))
class ReleaseMixin:
"""Release rows (pur_release) belong to a request, order or agreement but are keyed by document, so they are read and written
beside the aggregate (append-only: a level approved is never edited)."""
kind = ""
def _releases(self, doc_id: str) -> list:
t = pur_release
return [_make(approval.Release, dict(r._mapping)) for r in self.conn.execute(
select(t).where(t.c.document_kind == self.kind, t.c.document_id == doc_id).order_by(t.c.revision, t.c.level))]
def _save_releases(self, doc_id: str, releases: list) -> None:
upsert(self.conn, pur_release, [_pick(r, pur_release) for r in releases], ("document_kind", "document_id", "revision", "level"))
# A request that opens several transactions in a row (approve a bill: approve, post, hand to assets, answer) reads the same reference rows
# (settings, account map, supplier profile, banks, tolerances) in each. `with request_memo():` makes them read once; a write through
# Dated.put / settings.put drops the memo. Reference rows only: aggregates are never memoised.
REQUEST_MEMO: contextvars.ContextVar = contextvars.ContextVar("pur_request_memo", default=None)
@contextlib.contextmanager
def request_memo():
token = REQUEST_MEMO.set({})
try:
yield
finally:
REQUEST_MEMO.reset(token)
SMALL_TABLES = frozenset({"pur_account_map", "pur_tolerance", "pur_terms"})
class RequestRepository(ReleaseMixin):
kind = "request"
def __init__(self, conn):
self.conn = conn
self.repo = AggRepo(conn, REQUEST)
def get(self, rid):
got = self.repo.get(rid)
if got:
got[0].releases = self._releases(rid)
return got
def save(self, obj, expected_version=0):
v = self.repo.save(obj, expected_version)
self._save_releases(obj.purchase_request_id, obj.releases)
return v
def page(self, *, ids=None, status=None, limit=50, offset=0):
t = pur_request
where = []
if ids is not None:
from sqlalchemy import or_
where.append(or_(func.lower(t.c.requested_by_email).in_(list(ids) or ["\x00"]), func.lower(t.c.requested_by_sub).in_(list(ids) or ["\x00"])))
if status:
where.append(t.c.status == status)
return self.repo.heads(*where, order=[t.c.purchase_request_id.desc()], limit=limit, offset=offset)
def by_ref(self, request_ref: str):
r = self.conn.execute(select(pur_request.c.purchase_request_id).where(pur_request.c.request_ref == request_ref)).first()
return self.get(r[0]) if r else None
class OrderRepository(ReleaseMixin):
kind = "order"
def __init__(self, conn):
self.conn = conn
self.repo = AggRepo(conn, ORDER)
def get(self, pid):
memo = REQUEST_MEMO.get()
if memo is not None and ("order", pid) in memo: # inside request_memo() an order is read once (the approving request never writes one)
return memo[("order", pid)]
got = self.repo.get(pid)
if got:
got[0].releases = self._releases(pid)
if memo is not None:
memo[("order", pid)] = got
return got
def save(self, obj, expected_version=0):
if REQUEST_MEMO.get() is not None:
REQUEST_MEMO.get().pop(("order", obj.purchase_order_id), None)
v = self.repo.save(obj, expected_version)
self._save_releases(obj.purchase_order_id, obj.releases)
return v
def exists(self, pid):
return self.repo.exists(pid)
def page(self, *, bp_id=None, status=None, limit=50, offset=0):
t = pur_order
where = [c for c in (t.c.bp_id == int(bp_id) if bp_id is not None else None, t.c.status == status if status else None) if c is not None]
return self.repo.heads(*where, order=[t.c.purchase_order_id.desc()], limit=limit, offset=offset)
def open_lines(self, product_id=None):
"""(order head, line) of every line of a confirmed order (what is still to arrive is derived from these and the receipts)."""
q = (select(pur_order, pur_order_line).select_from(pur_order.join(pur_order_line, pur_order.c.purchase_order_id == pur_order_line.c.purchase_order_id))
.where(pur_order.c.status == "confirmed"))
if product_id:
q = q.where(pur_order_line.c.product_id == product_id)
out = []
for r in self.conn.execute(q):
m = r._mapping
out.append((_make(order.PurchaseOrder, {c.name: m[c] for c in pur_order.c}), _make(order.OrderLine, {c.name: m[c] for c in pur_order_line.c})))
return out
def confirmed_quantity_since(self, product_id: str, bp_id: int, since: date) -> Decimal:
"""Confirmed order quantity (base unit) of a product from a supplier since a date: the quota's allocated quantity, derived."""
q = (select(func.coalesce(func.sum(pur_order_line.c.quantity_base), 0))
.select_from(pur_order.join(pur_order_line, pur_order.c.purchase_order_id == pur_order_line.c.purchase_order_id))
.where(pur_order.c.bp_id == int(bp_id), pur_order_line.c.product_id == product_id, pur_order.c.confirmed_on >= since,
pur_order.c.status.in_(("confirmed", "closed"))))
return Decimal(str(self.conn.execute(q).scalar() or 0))
class AgreementRepository(ReleaseMixin):
"""A contract or scheduling agreement; its approval levels (pur_release, kind agreement) are read and written beside it."""
kind = "agreement"
def __init__(self, conn):
self.conn = conn
self.repo = AggRepo(conn, AGREEMENT)
def get(self, aid):
got = self.repo.get(aid)
if got:
got[0].releases = self._releases(aid)
return got
def save(self, obj, expected_version=0):
v = self.repo.save(obj, expected_version)
self._save_releases(obj.agreement_id, obj.releases)
return v
def heads(self, **kw):
return self.repo.heads(**kw)
class ReceiptRepository:
def __init__(self, conn):
self.conn = conn
self.repo = AggRepo(conn, RECEIPT)
def get(self, rid):
return self.repo.get(rid)
def save(self, obj, expected_version=0):
return self.repo.save(obj, expected_version)
def of_order(self, purchase_order_id: str) -> list:
"""Every receipt (and return) of an order, with its lines."""
ids = [r[0] for r in self.conn.execute(select(pur_receipt.c.receipt_id).where(pur_receipt.c.purchase_order_id == purchase_order_id))]
got = self.repo.get_many(ids)
return [got[i][0] for i in sorted(got)]
def lines_of_order(self, purchase_order_id: str) -> list:
"""[{receipt_id, kind, order_line_no, quantity_base, quantity, received_on}] of every receipt line of the order."""
memo = REQUEST_MEMO.get()
if memo is not None and ("receipt_lines", purchase_order_id) in memo:
return memo[("receipt_lines", purchase_order_id)]
q = (select(pur_receipt.c.receipt_id, pur_receipt.c.kind, pur_receipt.c.received_on, pur_receipt.c.corrects_receipt_id,
pur_receipt_line.c.order_line_no, pur_receipt_line.c.quantity_base, pur_receipt_line.c.quantity, pur_receipt_line.c.line_no)
.select_from(pur_receipt.join(pur_receipt_line, pur_receipt.c.receipt_id == pur_receipt_line.c.receipt_id))
.where(pur_receipt.c.purchase_order_id == purchase_order_id))
out = [dict(r._mapping) for r in self.conn.execute(q)]
if memo is not None:
memo[("receipt_lines", purchase_order_id)] = out
return out
def lines_of_orders_for_product(self, product_id: str, warehouse_id: Optional[str] = None) -> list:
return []
def pending_handover(self, limit: int = 100) -> list:
q = select(pur_receipt_line.c.receipt_id).where(pur_receipt_line.c.handed_over_at.is_(None)).distinct().limit(limit)
return [r[0] for r in self.conn.execute(q)]
def page(self, *, purchase_order_id=None, limit=50, offset=0):
t = pur_receipt
where = [t.c.purchase_order_id == purchase_order_id] if purchase_order_id else []
return self.repo.heads(*where, order=[t.c.receipt_id.desc()], limit=limit, offset=offset)
class Claims:
"""Claim rows: a plain INSERT of the next number in a scope, in the SAME transaction as the write it protects. Two writers that read
the same snapshot both pick the same seq; exactly one commits and the other's whole transaction fails 23505 -> Conflict (retry)."""
def __init__(self, conn):
self.conn = conn
def next_seq(self, scope: str) -> int:
t = pur_claim_versions
return int(self.conn.execute(select(func.coalesce(func.max(t.c.seq), 0)).where(t.c.scope == scope)).scalar() or 0) + 1
def claim(self, scope: str, seq: int, by: str = "") -> int:
try:
self.conn.execute(insert(pur_claim_versions).values(scope=scope, seq=int(seq), claimed_by=by, claimed_at=_now(),
created_at=_now(), updated_at=_now()))
except IntegrityError:
raise Conflict("pur_claim_versions", {"scope": scope, "seq": seq}) from None
return int(seq)
def take(self, scope: str, by: str = "") -> int:
return self.claim(scope, self.next_seq(scope), by)
def claim_many(self, keys: list, by: str = "") -> None:
"""Several (scope, seq) claims in ONE statement (a run claims each bill it takes): any one taken -> Conflict, nothing claimed."""
if not keys:
return
now = _now()
try:
self.conn.execute(insert(pur_claim_versions), [{"scope": sc, "seq": int(sq), "claimed_by": by, "claimed_at": now, "created_at": now,
"updated_at": now} for sc, sq in keys])
except IntegrityError:
raise Conflict("pur_claim_versions", {"scope": keys[0][0], "seq": keys[0][1]}) from None
def _now() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None)
class BillRepository:
def __init__(self, conn):
self.conn = conn
self.repo = AggRepo(conn, BILL)
def get(self, bid):
got = self.repo.get(bid)
if got:
got[0].claims = self.claims_of_bill(bid)
return got
def save(self, obj, expected_version=0):
v = self.repo.save(obj, expected_version)
self.put_claims(obj.claims)
return v
def claims_of_bill(self, bid: str) -> list:
t = pur_billed_claim
return [_make(bill.BilledClaim, dict(r._mapping)) for r in self.conn.execute(select(t).where(t.c.bill_id == bid))]
def put_claims(self, claims_: list) -> None:
t = pur_billed_claim
have = {(r[0], r[1], r[2]) for r in self.conn.execute(select(t.c.purchase_order_id, t.c.line_no, t.c.bill_id))} if claims_ else set()
new = [c for c in claims_ if (c.purchase_order_id, c.line_no, c.bill_id) not in have]
for c in new:
self.conn.execute(insert(t).values(**_pick(c, t), created_at=_now(), updated_at=_now())) # a plain INSERT: a collision is a Conflict
def claimed_by_line(self, purchase_order_id: str, kind: str = "bill", exclude_bill: str = "") -> dict:
"""{order line_no: Fraction base quantity claimed by OTHER approved bills (or credits)}."""
from fractions import Fraction
memo, mk = REQUEST_MEMO.get(), ("claimed", purchase_order_id, kind, exclude_bill)
if memo is not None and mk in memo:
return dict(memo[mk])
c, b = pur_billed_claim, pur_bill
q = (select(c.c.line_no, c.c.quantity_base).select_from(c.join(b, b.c.bill_id == c.c.bill_id))
.where(c.c.purchase_order_id == purchase_order_id, b.c.kind == kind, c.c.bill_id != exclude_bill))
out = {}
for ln, qb in self.conn.execute(q):
out[ln] = out.get(ln, Fraction(0)) + Fraction(Decimal(str(qb)))
if memo is not None:
memo[mk] = dict(out)
return out
def of_order(self, purchase_order_id: str) -> list:
t = pur_bill
return [_make(bill.SupplierBill, dict(r._mapping)) for r in self.conn.execute(select(t).where(t.c.purchase_order_id == purchase_order_id))]
def heads(self, **f) -> tuple:
t = pur_bill
where = []
for k in ("bp_id", "status", "kind"):
if f.get(k) is not None:
where.append(t.c[k] == f[k])
return self.repo.heads(*where, order=[t.c.invoice_date.desc(), t.c.bill_id], limit=f.get("limit", 50), offset=f.get("offset", 0))
def approved(self) -> list:
t = pur_bill
return self.repo.all_heads(t.c.status.in_(("approved", "disputed")), t.c.bill_no != "")
def lines_of(self, bill_ids: list) -> list:
t = pur_bill_line
return [_make(bill.BillLine, dict(r._mapping)) for r in self.conn.execute(select(t).where(t.c.bill_id.in_(bill_ids)))] if bill_ids else []
def by_source(self, source_kind: str, source_id: str):
t = pur_bill
r = self.conn.execute(select(t.c.bill_id).where(t.c.source_kind == source_kind, t.c.source_id == source_id)).first()
return self.get(r[0]) if r else None
class RunRepository:
def __init__(self, conn):
self.conn = conn
self.repo = AggRepo(conn, RUN)
def get(self, rid):
return self.repo.get(rid)
def save(self, obj, expected_version=0):
return self.repo.save(obj, expected_version)
def page(self, limit=50, offset=0):
return self.repo.heads(order=[pur_payment_run.c.run_id.desc()], limit=limit, offset=offset)
def open_items(self) -> list:
"""Items of runs that are not finished (draft, approved, exported): the bills they hold are not proposed again."""
r, i = pur_payment_run, pur_run_item
q = select(i.c.bill_id, i.c.run_id).select_from(i.join(r, r.c.run_id == i.c.run_id)).where(r.c.status.in_(("draft", "approved", "exported")))
return [(a, b) for a, b in self.conn.execute(q)]
def closed_inclusions(self, bill_ids) -> dict:
"""{bill_id: how many FINISHED runs (done, cancelled) held it}: the run claim's sequence. It moves only when a run finishes, so two
proposals racing each other compute the same key for a bill whatever either has committed, and exactly one takes it."""
if not bill_ids:
return {}
r, i = pur_payment_run, pur_run_item
q = (select(i.c.bill_id, func.count()).select_from(i.join(r, r.c.run_id == i.c.run_id))
.where(i.c.bill_id.in_(list(bill_ids)), r.c.status.notin_(("draft", "approved", "exported"))).group_by(i.c.bill_id))
return {a: int(n) for a, n in self.conn.execute(q)}
class PaymentRepository:
def __init__(self, conn):
self.conn = conn
self.repo = AggRepo(conn, PAYMENT)
def get(self, pid):
return self.repo.get(pid)
def save(self, obj, expected_version=0):
return self.repo.save(obj, expected_version)
def get_many(self, ids) -> dict:
return self.repo.get_many(ids)
def of_run(self, run_id: str) -> list:
return self.repo.all_heads(pur_payment.c.run_id == run_id)
def find(self, bank_account_id=None, status=None, paid_from=None, paid_to=None) -> list:
t = pur_payment
where = []
if bank_account_id:
where.append(t.c.bank_account_id == bank_account_id)
if status:
where.append(t.c.status == status)
if paid_from:
where.append(t.c.paid_on >= paid_from)
if paid_to:
where.append(t.c.paid_on <= paid_to)
return self.repo.all_heads(*where)
class AllocationRepository:
"""Append-only: a payment, a credit or a down payment settling a bill. A correction is a reversing row (negative amount)."""
def __init__(self, conn):
self.conn = conn
def put(self, a: payment.Allocation) -> None:
t = pur_allocation
self.conn.execute(insert(t).values(**_pick(a, t), created_at=_now(), updated_at=_now())) # a plain INSERT: a collision is a Conflict
def of_bill(self, bill_id: str, as_of: Optional[date] = None) -> list:
t = pur_allocation
q = select(t).where(t.c.bill_id == bill_id)
if as_of:
q = q.where(t.c.allocated_on <= as_of)
return [_make(payment.Allocation, dict(r._mapping)) for r in self.conn.execute(q)]
def of_source(self, source_kind: str, source_id: str) -> list:
t = pur_allocation
return [_make(payment.Allocation, dict(r._mapping)) for r in self.conn.execute(
select(t).where(t.c.source_kind == source_kind, t.c.source_id == source_id))]
def sums(self, bill_ids: list, as_of: Optional[date] = None) -> dict:
"""{bill_id: Decimal allocated} in one query: outstanding = total - this."""
t = pur_allocation
if not bill_ids:
return {}
q = select(t.c.bill_id, func.sum(t.c.amount)).where(t.c.bill_id.in_(bill_ids)).group_by(t.c.bill_id)
if as_of:
q = q.where(t.c.allocated_on <= as_of)
return {b: Decimal(str(s or 0)) for b, s in self.conn.execute(q)}
def credit_used(self, credit_ids: list) -> dict:
t = pur_allocation
if not credit_ids:
return {}
q = select(t.c.source_id, func.sum(t.c.amount)).where(t.c.source_kind == "credit", t.c.source_id.in_(credit_ids)).group_by(t.c.source_id)
return {b: Decimal(str(s or 0)) for b, s in self.conn.execute(q)}
class Dated:
"""A table of dated rows: [key..., valid_from]. A correction is a new row, never an edit; the row in force on a date is the latest
valid_from on or before it."""
def __init__(self, conn, table: Table):
self.conn, self.t = conn, table
cache: Optional[dict] = None # a request that reads one key many times (a payment run's bills) sets {} to read each key once
def _memo(self) -> Optional[dict]:
return self.cache if self.cache is not None else REQUEST_MEMO.get()
def all(self, **where) -> list:
ck = (self.t.name, tuple(sorted(where.items())))
memo = self._memo()
if memo is not None and ck in memo:
return memo[ck]
if memo is not None and where and self.t.name in SMALL_TABLES: # a small reference table: read whole once, filter here
whole = self.all()
return [r for r in whole if all(r[k] == v for k, v in where.items())]
q = select(self.t)
for k, v in where.items():
q = q.where(self.t.c[k] == v)
out = [{c: r._mapping[c] for c in _cols(self.t)} for r in self.conn.execute(q)]
if memo is not None:
memo[ck] = out
return out
def put(self, row: dict) -> None:
for m in (self.cache, REQUEST_MEMO.get()):
if m is not None:
m.clear()
keep = {k: v for k, v in row.items() if k in self.t.c}
upsert(self.conn, self.t, [keep], tuple(c.name for c in self.t.primary_key.columns))
def insert(self, row: dict) -> None:
"""A plain INSERT: a second writer of the same key fails (a Conflict) instead of overwriting."""
try:
self.conn.execute(insert(self.t).values(**{k: v for k, v in row.items() if k in self.t.c}, created_at=_now(), updated_at=_now()))
except IntegrityError:
raise Conflict(self.t.name, {k: row.get(k) for k in (c.name for c in self.t.primary_key.columns)}) from None
def get(self, *key) -> Optional[dict]:
names = [c.name for c in self.t.primary_key.columns]
r = self.conn.execute(select(self.t).where(*[self.t.c[n] == v for n, v in zip(names, key)])).first()
return None if r is None else {c: r._mapping[c] for c in _cols(self.t)}
class SettingsRepository:
"""pur_settings: one dated row per change [setting_key, valid_from]. Seeded defaults start 1970-01-01; every later row starts AFTER
the day it is set (the use case checks)."""
def __init__(self, conn):
self.conn = conn
self._by_key: dict = {} # key -> its rows as this transaction read them (a put drops it): a request reads each setting once
def rows(self, key: Optional[str] = None) -> list:
memo = REQUEST_MEMO.get()
if key and key in self._by_key:
return self._by_key[key]
if key and memo is not None:
if ("settings", key) not in memo: # the whole (small) settings table in one read, then each key from it
by: dict = {}
for r in self.rows():
by.setdefault(r["setting_key"], []).append(r)
memo.update({("settings", k): v for k, v in by.items()})
memo.setdefault(("settings", key), [])
return memo[("settings", key)]
q = select(pur_settings).order_by(pur_settings.c.setting_key, pur_settings.c.valid_from)
if key:
q = q.where(pur_settings.c.setting_key == key)
out = [{**{k: r._mapping[k] for k in ("setting_key", "valid_from", "set_by", "set_at", "changes")}, "value": json.loads(r._mapping["value"])}
for r in self.conn.execute(q)]
if key:
self._by_key[key] = out
if memo is not None:
memo[("settings", key)] = out
return out
def value_on(self, key: str, day: date):
got = [r for r in self.rows(key) if r["valid_from"] <= day]
return got[-1]["value"] if got else SETTING_DEFAULTS.get(key)
def put(self, key: str, valid_from: date, value, set_by: str, set_at: datetime, changes: str = "") -> None:
self._by_key.pop(key, None)
if REQUEST_MEMO.get() is not None:
REQUEST_MEMO.get().pop(("settings", key), None)
upsert(self.conn, pur_settings, [{"setting_key": key, "valid_from": valid_from, "value": json.dumps(value), "set_by": set_by,
"set_at": set_at, "changes": changes}], ("setting_key", "valid_from"))
def seed(self) -> None:
have = {r[0] for r in self.conn.execute(select(pur_settings.c.setting_key))}
for s in SETTINGS:
if s.key not in have:
self.put(s.key, date.fromisoformat(EPOCH), s.default, "default", datetime(1970, 1, 1), s.changes)
class NumberSeriesRepository:
"""order_no and bill_no etc.: a high-water mark per (series, year) plus a claim row keyed on the RENDERED number, so a number is never
reused (a cancelled confirmed order keeps its number) and a migrated number below the mark is never filled."""
def __init__(self, conn):
self.conn = conn
self.series = AggRepo(conn, NUMBER_SERIES)
def get(self, series_id: str, year: int) -> tuple:
got = self.series.get(series_id, year)
return got if got else (_Series(series_id, year), 0)
def save(self, s, expected_version: int) -> int:
return self.series.save(s, expected_version)
def claim_number(self, number: str, series_id: str, document_id: str) -> bool:
"""True when the number is (now or already) this document's; False when another document holds it."""
t = pur_number_claim
row = self.conn.execute(select(t.c.document_id).where(t.c.number == number)).first()
if row is not None:
return row[0] == document_id
self.conn.execute(insert(t).values(number=number, series_id=series_id, document_id=document_id, claimed_at=_now(), created_at=_now(),
updated_at=_now()))
return True
def marks(self) -> list:
t = pur_number_series
return [(r[0], r[1], int(r[2] or 0)) for r in self.conn.execute(select(t.c.series_id, t.c.year, t.c.high_water_mark))]
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.requests = RequestRepository(conn)
self.orders = OrderRepository(conn)
self.receipts = ReceiptRepository(conn)
self.bills = BillRepository(conn)
self.runs = RunRepository(conn)
self.payments = PaymentRepository(conn)
self.allocations = AllocationRepository(conn)
self.claims = Claims(conn)
self.numbers = NumberSeriesRepository(conn)
self.settings = SettingsRepository(conn)
self.rfqs = AggRepo(conn, RFQ)
self.agreements = AgreementRepository(conn)
self.down_payments = AggRepo(conn, DOWN_PAYMENT)
for name, t in (("limits", pur_approval_limit), ("strategies", pur_release_strategy), ("tolerances", pur_tolerance), ("terms", pur_terms),
("account_map", pur_account_map), ("suppliers", pur_supplier), ("banks", pur_supplier_bank),
("supplier_prices", pur_supplier_price), ("sources", pur_source), ("quotas", pur_quota),
("evaluations", pur_evaluation), ("weights", pur_evaluation_weight), ("service_scores", pur_service_score),
("self_billing", pur_self_billing), ("withdrawals", pur_consignment_withdrawal),
("cis", pur_cis_deduction), ("findings", pur_ai_finding), ("tasks", pur_task), ("handovers", pur_handover),
("objects", pur_object), ("migration_map", pur_migration_map), ("migration_runs", pur_migration_run),
("migration_checks", pur_migration_check), ("agreement_claims", pur_agreement_claim)):
setattr(self, name, Dated(conn, t))
def seed(self) -> None:
"""The rows a new bucket starts with: the settings, and the standard payment terms."""
self.settings.seed()
for tid, name, days, eom in (("NET30", "30 days", 30, False), ("NET14", "14 days", 14, False), ("NET60", "60 days", 60, False),
("EOM30", "end of month following", 30, True)):
if self.terms.get(tid) is None:
self.terms.put({"terms_id": tid, "name": name, "days": days, "end_of_month": eom})