dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/infrastructure/tables.py
"""The tables, and one repository per aggregate that loads and saves it whole.
Every table is prefixed `ar_`, has a natural primary key (never SERIAL) and, for a part, a key that starts with its root's
(the repository finds stale rows under a root by that prefix). Money is NUMERIC(18,2); a rate (a price per unit, an FX
rate, a percentage) is NUMERIC(19,4) / NUMERIC(9,4). Dated rows (settings, billing profiles, the account map) are only ever
INSERTED: a row is never edited, a change is a new row from a later day. schema.sql is written from this column model by
scripts/gen_schema.py; tests/test_schema.py keeps the two equal.
A document number is claimed with a plain INSERT into ar_number_claims (series, counter): the primary key is the uniqueness
guard, the loser of a race gets 23505 and the use case retries (platform_constraints in the spec).
"""
import dataclasses
import copy
import json
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from erp_platform.repository import Aggregate, Part, upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
Text,
and_,
delete,
func,
insert,
or_,
select,
)
from brick_ar.domain import customer as customer_mod
from brick_ar.domain.credit_note import CreditNote
from brick_ar.domain.invoice import Finding, Invoice, Line, Send, TaxRow
from brick_ar.domain.migration_run import MigrationCheck, MigrationRun
from brick_ar.domain.payment_request import PaymentRequest
from brick_ar.domain.provision import ProvisionRun
from brick_ar.domain.schedule import Schedule
from brick_ar.domain.values import SETTINGS, Settings
from brick_ar.domain.writeoff import Writeoff
md = MetaData()
MONEY, RATE, PCT = Numeric(18, 2), Numeric(19, 4), Numeric(9, 4)
def _t(name, *cols):
return Table(name, md, *cols, Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
def _c(name, typ=String, **kw):
return Column(name, typ, **kw)
ar_invoices = _t("ar_invoices",
_c("invoice_id", primary_key=True), _c("bp_id", BigInteger), _c("lifecycle"), _c("kind"), _c("source_kind"),
_c("source_id"), _c("invoice_no"), _c("series_id"), _c("currency"), _c("fx_rate", RATE), _c("issue_date", Date),
_c("tax_point_date", Date), _c("due_date", Date), _c("terms_id"), _c("customer_po_ref"), _c("bill_to_json", Text),
_c("subtotal", MONEY), _c("tax_total", MONEY), _c("total", MONEY), _c("total_functional", MONEY),
_c("export_batch_id"), _c("migration_run_id"), _c("company_code"), _c("trading_partner"),
_c("settlement_discount_pct", PCT), _c("settlement_discount_until", Date), _c("einvoice_message_id"), _c("sold_to_bp_id", BigInteger), _c("ship_to_bp_id", BigInteger),
_c("bill_to_bp_id", BigInteger), _c("created_by"), _c("reviewed_by"), _c("reviewed_at", TIMESTAMP),
_c("void_reason"), _c("voided_on", Date), _c("version", Integer))
ar_invoice_lines = _t("ar_invoice_lines",
_c("invoice_id", primary_key=True), _c("line_no", Integer, primary_key=True), _c("product_id"), _c("description", Text),
_c("unit_code"), _c("quantity", RATE), _c("unit_price", RATE), _c("discount_pct", PCT), _c("tax_category"),
_c("tax_code"), _c("net_amount", MONEY), _c("tax_amount", MONEY), _c("sales_order_id"), _c("so_line_no", Integer),
_c("service_from", Date), _c("service_to", Date), _c("schedule_id"), _c("item_id"), _c("deducts_invoice_id"),
_c("odoo_line_id"), _c("usage", Boolean), _c("pinned_from"), _c("interest_on_invoice_id"), _c("interest_days", Integer),
_c("interest_rate_pct", PCT), _c("cost_centre"), _c("pricing_quantity", RATE), _c("pricing_unit_code"),
_c("quantity_priced", RATE), _c("net_given", Boolean))
ar_invoice_tax = _t("ar_invoice_tax",
_c("invoice_id", primary_key=True), _c("tax_code", primary_key=True), _c("rate_pct", PCT), _c("net_amount", MONEY),
_c("tax_amount", MONEY), _c("rounding_method"))
ar_sends = _t("ar_sends",
_c("invoice_id", primary_key=True), _c("sent_at", TIMESTAMP, primary_key=True), _c("channel"), _c("address"),
_c("format"), _c("result"), _c("message_id"), _c("statuses_json", Text))
ar_ai_finding = _t("ar_ai_finding",
_c("invoice_id", primary_key=True), _c("step", primary_key=True), _c("run_no", Integer, primary_key=True), _c("model"),
_c("inputs_digest"), _c("finding", Text), _c("evidence", Text), _c("decision"), _c("reason", Text), _c("decided_by"),
_c("decided_at", TIMESTAMP))
INVOICE = Aggregate(Part(ar_invoices, ("invoice_id",)), Part(ar_invoice_lines, ("invoice_id", "line_no")),
Part(ar_invoice_tax, ("invoice_id", "tax_code")), Part(ar_sends, ("invoice_id", "sent_at")),
Part(ar_ai_finding, ("invoice_id", "step", "run_no")), version="version")
ar_credit_notes = _t("ar_credit_notes",
_c("credit_id", primary_key=True), _c("bp_id", BigInteger), _c("invoice_id"), _c("credit_no"), _c("series_id"),
_c("credit_date", Date), _c("currency"), _c("reason", Text), _c("status"), _c("subtotal", MONEY), _c("tax_total", MONEY),
_c("total", MONEY), _c("source_kind"), _c("drafted_by"), _c("drafted_by_sub"), _c("approved_by"), _c("approved_at", TIMESTAMP),
_c("export_batch_id"), _c("migration_run_id"), _c("payment_id"), _c("company_code"), _c("version", Integer))
ar_credit_note_lines = _t("ar_credit_note_lines",
_c("credit_id", primary_key=True), _c("line_no", Integer, primary_key=True), _c("product_id"), _c("description", Text),
_c("unit_code"), _c("quantity", RATE), _c("unit_price", RATE), _c("discount_pct", PCT), _c("tax_category"),
_c("tax_code"), _c("net_amount", MONEY), _c("tax_amount", MONEY), _c("sales_order_id"), _c("so_line_no", Integer),
_c("odoo_line_id"), _c("pinned_from"), _c("cost_centre"))
ar_credit_note_tax = _t("ar_credit_note_tax",
_c("credit_id", primary_key=True), _c("tax_code", primary_key=True), _c("rate_pct", PCT), _c("net_amount", MONEY),
_c("tax_amount", MONEY), _c("rounding_method"))
CREDIT_NOTE = Aggregate(Part(ar_credit_notes, ("credit_id",)), Part(ar_credit_note_lines, ("credit_id", "line_no")),
Part(ar_credit_note_tax, ("credit_id", "tax_code")), version="version")
ar_writeoffs = _t("ar_writeoffs",
_c("writeoff_id", primary_key=True), _c("invoice_id"), _c("amount", MONEY), _c("reason", Text), _c("status"),
_c("proposed_by"), _c("proposed_by_sub"), _c("writeoff_date", Date), _c("approved_by"), _c("approved_at", TIMESTAMP),
_c("rejected_reason", Text), _c("version", Integer))
WRITEOFF = Aggregate(Part(ar_writeoffs, ("writeoff_id",)), version="version")
ar_schedule = _t("ar_schedule",
_c("schedule_id", primary_key=True), _c("bp_id", BigInteger), _c("source_kind"), _c("source_id"), _c("line_id"),
_c("product_id"), _c("description", Text), _c("billing_kind"), _c("billing_plan"), _c("starts_on", Date),
_c("periods", Integer), _c("quantity", RATE), _c("unit_price", RATE), _c("discount_pct", PCT), _c("unit_code"),
_c("tax_category"), _c("tax_code"), _c("currency"), _c("in_advance", Boolean), _c("billing_day"),
_c("prorate_first", Boolean), _c("status"), _c("stopped_at", Date), _c("stop_reason", Text), _c("created_by"),
_c("version", Integer))
SCHEDULE = Aggregate(Part(ar_schedule, ("schedule_id",)), version="version")
ar_customer = _t("ar_customer",
_c("bp_id", BigInteger, primary_key=True), _c("valid_from", Date, primary_key=True), _c("bill_to_name"),
_c("bill_to_address", Text), _c("invoice_email"), _c("terms_id"), _c("currency"), _c("intercompany_company_code"),
_c("einvoice_channel"), _c("peppol_participant_id"), _c("default_cost_centre"),
_c("set_by"), _c("set_at", TIMESTAMP))
ar_terms = _t("ar_terms",
_c("terms_id", primary_key=True), _c("days", Integer), _c("basis"), _c("description", Text),
_c("early_discount_pct", PCT), _c("active", Boolean), _c("early_discount_days", Integer))
ar_settings = _t("ar_settings",
_c("setting_key", primary_key=True), _c("valid_from", Date, primary_key=True), _c("value", Text), _c("set_by"),
_c("set_at", TIMESTAMP))
ar_account_map = _t("ar_account_map",
_c("purpose", primary_key=True), _c("tax_code", primary_key=True), _c("valid_from", Date, primary_key=True),
_c("account_id", BigInteger), _c("set_by"))
ar_number_series = _t("ar_number_series", _c("series_id", primary_key=True), _c("high_water_mark", Integer))
ar_number_claims = _t("ar_number_claims",
_c("series_id", primary_key=True), _c("seq", Integer, primary_key=True), _c("document_id"), _c("claimed_at", TIMESTAMP))
ar_invoice_no_claims = _t("ar_invoice_no_claims", _c("invoice_no", primary_key=True), _c("document_id"), _c("claimed_at", TIMESTAMP))
ar_credit_claim = _t("ar_credit_claim",
_c("invoice_id", primary_key=True), _c("n", Integer, primary_key=True), _c("credit_id"), _c("claimed_at", TIMESTAMP))
ar_payment_request = _t("ar_payment_request",
_c("request_id", primary_key=True), _c("bp_id", BigInteger), _c("company_code"), _c("lifecycle"), _c("kind"), _c("sales_order_id"),
_c("currency"), _c("tax_category"), _c("net_amount", MONEY), _c("tax_amount_display", MONEY), _c("total", MONEY),
_c("request_no"), _c("issue_date", Date), _c("due_date", Date), _c("deposit_invoice_ids", Text), _c("cancel_reason", Text),
_c("created_by"), _c("version", Integer))
PAYMENT_REQUEST = Aggregate(Part(ar_payment_request, ("request_id",)), version="version")
# One row per receipt that settled (part of) a request, keyed on the payment: the same payment settles once (AR-36).
ar_request_settlement = _t("ar_request_settlement",
_c("payment_id", primary_key=True), _c("request_id"), _c("n", Integer), _c("received_on", Date), _c("amount", MONEY),
_c("deposit_invoice_id"), _c("settled_at", TIMESTAMP))
# A deposit's deductions: slot n = deductions so far + 1 keyed (deposit, n), so two finals racing for one remainder collide (AR-15).
ar_deposit_claim = _t("ar_deposit_claim",
_c("deposit_invoice_id", primary_key=True), _c("n", Integer, primary_key=True), _c("final_invoice_id"), _c("net", MONEY),
_c("claimed_at", TIMESTAMP))
# A settlement-discount credit note, keyed on the receipt it was taken on: a second call for the payment answers the same one.
ar_settlement_discount = _t("ar_settlement_discount",
_c("payment_id", primary_key=True), _c("invoice_id"), _c("credit_id"), _c("claimed_at", TIMESTAMP))
ar_interest_rate = _t("ar_interest_rate",
_c("valid_from", Date, primary_key=True), _c("reference_rate_pct", PCT), _c("margin_pct", PCT), _c("source"), _c("set_by"))
# The intercompany hand-over: a document handed to the partner company's purchasing, once (AR-41).
ar_ic_handover = _t("ar_ic_handover",
_c("document_id", primary_key=True), _c("bill_id"), _c("partner_company"), _c("handed_at", TIMESTAMP))
# AR-46: a line's service revenue released from deferred revenue, one row per (invoice, line, period) - the claim that makes a second
# run add nothing - and (kind credit, period_id "cn:<credit_id>") what a credit note took back from the unreleased part first.
ar_deferral_release = _t("ar_deferral_release",
_c("invoice_id", primary_key=True), _c("line_no", Integer, primary_key=True), _c("period_id", primary_key=True), _c("kind"),
_c("amount", MONEY), _c("journal_date", Date), _c("released_at", TIMESTAMP))
# AR-45: the doubtful-debt provision of a month, and the dated rate per aging bucket (bucket "specific" = an invoice in dispute).
ar_provision_run = _t("ar_provision_run",
_c("run_id", primary_key=True), _c("company_code"), _c("period_id"), _c("lifecycle"), _c("as_at", Date), _c("lines_json", Text),
_c("total", MONEY), _c("previous_total", MONEY), _c("movement", MONEY), _c("proposed_by"), _c("proposed_by_sub"), _c("approved_by"),
_c("approved_at", TIMESTAMP), _c("rejected_reason", Text), _c("journal_id", BigInteger), _c("version", Integer))
PROVISION_RUN = Aggregate(Part(ar_provision_run, ("run_id",)), version="version")
ar_provision_rate = _t("ar_provision_rate",
_c("bucket", primary_key=True), _c("valid_from", Date, primary_key=True), _c("rate_pct", PCT), _c("set_by"))
# AR-43 / AR-44: what was handed to the access point (a message per document) and each status it reported, once per (message, status).
ar_einvoice_message = _t("ar_einvoice_message",
_c("message_id", primary_key=True), _c("document_id"), _c("document_kind"), _c("provider"), _c("sent_at", TIMESTAMP))
ar_einvoice_status = _t("ar_einvoice_status",
_c("message_id", primary_key=True), _c("status", primary_key=True), _c("status_at", TIMESTAMP), _c("reason", Text))
ar_export_batch = _t("ar_export_batch",
_c("batch_id", primary_key=True), _c("period"), _c("format"), _c("document_count", Integer), _c("file_key"),
_c("content", Text), _c("created_by"))
ar_postings = _t("ar_postings",
_c("document_id", primary_key=True), _c("event", primary_key=True), _c("journal_id", BigInteger),
_c("journal_date", Date), _c("posted_by"))
ar_migration_run = _t("ar_migration_run",
_c("run_id", primary_key=True), _c("source"), _c("source_db"), _c("cutover_date", Date), _c("doors_json", Text),
_c("status"), _c("loaded_at", TIMESTAMP), _c("checksums_json", Text), _c("signed_by"), _c("signed_at", TIMESTAMP),
_c("counts_json", Text), _c("version", Integer))
ar_migration_check = _t("ar_migration_check",
_c("run_id", primary_key=True), _c("check_key", primary_key=True), _c("slice", primary_key=True), _c("odoo_value", Text),
_c("ours_value", Text), _c("difference", Text), _c("equal", Boolean), _c("must"), _c("explanation", Text),
_c("explained_by"))
MIGRATION_RUN = Aggregate(Part(ar_migration_run, ("run_id",)), Part(ar_migration_check, ("run_id", "check_key", "slice")),
version="version")
ar_migration_map = _t("ar_migration_map",
_c("odoo_db", primary_key=True), _c("odoo_model", primary_key=True), _c("odoo_id", primary_key=True), _c("natural_key"),
_c("run_id"))
# What a release after the first ADDED (scripts/gen_schema.py writes 00001_init without these and a migration with them):
# {release: {table: "*" (the whole table) or [columns]}}. Additive only.
ADDED = {2: {"ar_writeoffs": ["proposed_by_sub"], "ar_credit_notes": ["drafted_by_sub"],
"ar_invoice_no_claims": "*", "ar_credit_claim": "*"},
3: {"ar_invoices": ["company_code", "trading_partner", "settlement_discount_pct", "settlement_discount_until"],
"ar_invoice_lines": ["pinned_from", "interest_on_invoice_id", "interest_days", "interest_rate_pct"],
"ar_credit_notes": ["payment_id", "company_code"], "ar_credit_note_lines": ["pinned_from"],
"ar_customer": ["intercompany_company_code"], "ar_terms": ["early_discount_days"], "ar_payment_request": "*",
"ar_request_settlement": "*", "ar_deposit_claim": "*", "ar_settlement_discount": "*", "ar_interest_rate": "*",
"ar_ic_handover": "*"},
4: {"ar_invoices": ["einvoice_message_id"], "ar_invoice_lines": ["cost_centre"], "ar_credit_note_lines": ["cost_centre"],
"ar_customer": ["einvoice_channel", "peppol_participant_id", "default_cost_centre"], "ar_sends": ["message_id", "statuses_json"],
"ar_deferral_release": "*", "ar_provision_run": "*", "ar_provision_rate": "*", "ar_einvoice_message": "*",
"ar_einvoice_status": "*"},
5: {"ar_invoices": ["sold_to_bp_id", "ship_to_bp_id", "bill_to_bp_id"], "ar_invoice_lines": ["pricing_quantity", "pricing_unit_code"]},
6: {"ar_invoice_lines": ["quantity_priced", "net_given"]}}
NON_STORED = {"events", "lines", "tax", "sends", "findings", "checks"}
def _names(table: Table) -> tuple:
return tuple(c.name for c in table.c if c.name not in ("created_at", "updated_at", "version"))
def _pick(obj, table: Table) -> dict:
return {n: getattr(obj, n, None) for n in _names(table)}
def _make(cls, row: dict, table: Table):
"""The object from a stored row; a NULL column keeps the field's default."""
fields = {f.name for f in dataclasses.fields(cls)}
return cls(**{n: row[n] for n in _names(table) if n in fields and row.get(n) is not None})
def _dec(v) -> Optional[Decimal]:
return None if v is None else Decimal(str(v))
def _utc(v: Optional[datetime]) -> Optional[datetime]:
"""Stored timestamps are naive UTC."""
if v is None:
return None
return v.astimezone(timezone.utc).replace(tzinfo=None) if v.tzinfo else v
def _stamp() -> dict:
now = datetime.now(timezone.utc).replace(tzinfo=None)
return {"created_at": now, "updated_at": now}
def _fix_money(obj, names: tuple) -> None:
for n in names:
v = getattr(obj, n, None)
if v is not None:
setattr(obj, n, Decimal(str(v)))
MONEY_FIELDS = ("subtotal", "tax_total", "total", "total_functional", "fx_rate", "settlement_discount_pct")
LINE_DEC = ("quantity", "unit_price", "discount_pct", "net_amount", "tax_amount", "interest_rate_pct", "pricing_quantity", "quantity_priced")
TAX_DEC = ("rate_pct", "net_amount", "tax_amount")
class InvoiceRepository:
"""Loads and saves an invoice whole (lines, tax, sends, findings). Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, invoice_id) -> Optional[tuple]:
"""(Invoice, version) or None. Read once per request: a second get answers a fresh aggregate built from the same rows,
until a save of it. An invoice not stored yet is saved without re-reading its keys (nothing_stored; the version claim
still refuses a racer)."""
loaded = self.__dict__.setdefault("_loaded", {})
if invoice_id not in loaded:
loaded[invoice_id] = INVOICE.load(self.conn, {"invoice_id": invoice_id})
rows = loaded[invoice_id]
if rows is None:
self._snapshots.setdefault(invoice_id, INVOICE.nothing_stored({"invoice_id": invoice_id}))
return None
self._snapshots[invoice_id] = rows.snapshot
return self._build(rows)
def get_many(self, ids) -> dict:
"""{id: (aggregate, version)} for a set of ids: one query per table per 100 roots (a loop of get() is one round trip per root)."""
out = {}
loaded = self.__dict__.setdefault("_loaded", {})
for (rid,), rows in INVOICE.load_many(self.conn, [{"invoice_id": i} for i in dict.fromkeys(ids)]).items():
loaded[rid] = rows # get() answers from these rows for the rest of the request
if rows is None:
continue
self._snapshots[rid] = rows.snapshot
out[rid] = self._build(rows)
return out
def prefetch(self, ids) -> None:
"""Read these invoices in one go (one query per table per 100): get() then answers them without a read each."""
loaded = self.__dict__.setdefault("_loaded", {})
want = [i for i in dict.fromkeys(ids) if i not in loaded]
if want:
self.get_many(want)
def _build(self, rows) -> tuple:
inv = _make(Invoice, rows["ar_invoices"][0], ar_invoices)
_fix_money(inv, MONEY_FIELDS)
inv.lines = []
for r in sorted(rows["ar_invoice_lines"], key=lambda r: r["line_no"]):
ln = _make(Line, r, ar_invoice_lines)
_fix_money(ln, LINE_DEC)
ln.usage, ln.net_given = bool(r.get("usage")), bool(r.get("net_given"))
inv.lines.append(ln)
inv.tax = []
for r in sorted(rows["ar_invoice_tax"], key=lambda r: r["tax_code"]):
t = _make(TaxRow, r, ar_invoice_tax)
_fix_money(t, TAX_DEC)
inv.tax.append(t)
inv.sends = [_make(Send, {**r, "sent_at": _utc(r["sent_at"])}, ar_sends) for r in sorted(rows["ar_sends"], key=lambda r: _utc(r["sent_at"]))]
inv.findings = [_make(Finding, r, ar_ai_finding) for r in sorted(rows["ar_ai_finding"], key=lambda r: (r["step"], r["run_no"]))]
return inv, INVOICE.version_of(rows)
def save(self, inv: Invoice, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
snapshot = self._snapshots.pop(inv.invoice_id, None) # spent: after this save the rows are no longer what it read
self.__dict__.get("_loaded", {}).pop(inv.invoice_id, None)
lines = [{**_pick(ln, ar_invoice_lines), "invoice_id": inv.invoice_id} for ln in inv.lines]
return INVOICE.save(self.conn, {
"ar_invoices": [_pick(inv, ar_invoices)], "ar_invoice_lines": lines,
"ar_invoice_tax": [{**_pick(t, ar_invoice_tax), "invoice_id": inv.invoice_id} for t in inv.tax],
"ar_sends": [{**_pick(s, ar_sends), "invoice_id": inv.invoice_id, "sent_at": _utc(s.sent_at)} for s in inv.sends],
"ar_ai_finding": [{**_pick(f, ar_ai_finding), "invoice_id": inv.invoice_id, "decided_at": _utc(f.decided_at)}
for f in inv.findings]}, expected_version=expected_version, snapshot=snapshot)
def exists(self, invoice_id) -> bool:
return self.conn.execute(select(ar_invoices.c.invoice_id).where(ar_invoices.c.invoice_id == invoice_id)).first() is not None
def head(self, invoice_id) -> Optional[dict]:
r = self.conn.execute(select(ar_invoices).where(ar_invoices.c.invoice_id == invoice_id)).first()
return dict(r._mapping) if r else None
def intercompany_ids(self, company_code: str, trading_partner: str) -> list:
q = select(ar_invoices.c.invoice_id).where(and_(ar_invoices.c.company_code == company_code, ar_invoices.c.trading_partner == trading_partner,
ar_invoices.c.lifecycle == "issued")).order_by(ar_invoices.c.invoice_id)
return [r[0] for r in self.conn.execute(q)]
def page(self, *, lifecycle=None, bp_id=None, currency=None, source_kind=None, issued_from=None, issued_to=None,
limit=50, offset=0) -> tuple:
"""(rows, total): the list of invoices, newest first. Heads only: no lines."""
cond = []
for col, v in ((ar_invoices.c.lifecycle, lifecycle), (ar_invoices.c.bp_id, bp_id), (ar_invoices.c.currency, currency),
(ar_invoices.c.source_kind, source_kind)):
if v is not None:
cond.append(col == v)
if issued_from:
cond.append(ar_invoices.c.issue_date >= issued_from)
if issued_to:
cond.append(ar_invoices.c.issue_date <= issued_to)
where = and_(*cond) if cond else None
total_q = select(func.count()).select_from(ar_invoices)
q = select(ar_invoices).order_by(ar_invoices.c.issue_date.desc(), ar_invoices.c.invoice_id).limit(limit).offset(offset)
if where is not None:
total_q, q = total_q.where(where), q.where(where)
return [dict(r._mapping) for r in self.conn.execute(q)], int(self.conn.execute(total_q).scalar() or 0)
def raised_periods(self, schedule_id: str) -> set:
"""The periods of a schedule that have an invoice (any state): derived from the keys, never counted."""
prefix = f"rec:{schedule_id}:"
rows = self.conn.execute(select(ar_invoices.c.invoice_id).where(ar_invoices.c.invoice_id.like(prefix + "%")))
out = set()
for (iid,) in rows:
tail = iid[len(prefix):]
if tail.isdigit():
out.add(int(tail))
return out
def issued_for(self, bp_id: int) -> list:
q = select(ar_invoices).where(and_(ar_invoices.c.bp_id == bp_id, ar_invoices.c.lifecycle == "issued"))
return [dict(r._mapping) for r in self.conn.execute(q)]
def recent_totals(self, bp_id: int, exclude: str, limit: int = 12) -> list:
q = (select(ar_invoices.c.total, ar_invoices.c.issue_date, ar_invoices.c.customer_po_ref).where(and_(
ar_invoices.c.bp_id == bp_id, ar_invoices.c.lifecycle == "issued", ar_invoices.c.invoice_id != exclude))
.order_by(ar_invoices.c.issue_date.desc()).limit(limit))
return [(Decimal(str(t)), d, po or "") for t, d, po in self.conn.execute(q)]
def in_period_unexported(self, start: date, end: date) -> list:
q = select(ar_invoices).where(and_(
ar_invoices.c.lifecycle.in_(("issued", "void")), ar_invoices.c.issue_date >= start, ar_invoices.c.issue_date <= end,
ar_invoices.c.source_kind != "odoo")).order_by(ar_invoices.c.issue_date, ar_invoices.c.invoice_no)
return [dict(r._mapping) for r in self.conn.execute(q) if not r._mapping["export_batch_id"]]
def ledger_states(self) -> list:
"""(invoice_id, lifecycle, issue_date, issue posted?, void posted?) of every issued or void invoice of this brick, in ONE
statement: the ledger pass knows which already have their journals without loading each (it used to: 6 reads apiece)."""
pi, pv = ar_postings.alias("p_issue"), ar_postings.alias("p_void")
q = (select(ar_invoices.c.invoice_id, ar_invoices.c.lifecycle, ar_invoices.c.issue_date,
pi.c.document_id.label("issue_posted"), pv.c.document_id.label("void_posted"))
.select_from(ar_invoices
.outerjoin(pi, and_(pi.c.document_id == ar_invoices.c.invoice_id, pi.c.event == "issue"))
.outerjoin(pv, and_(pv.c.document_id == ar_invoices.c.invoice_id, pv.c.event == "void")))
.where(and_(ar_invoices.c.lifecycle.in_(("issued", "void")), ar_invoices.c.source_kind != "odoo")))
return [(i, lc, day, ip is not None, vp is not None)
for i, lc, day, ip, vp in self.conn.execute(q)]
def unposted_ids(self, posted: set) -> list:
q = select(ar_invoices.c.invoice_id, ar_invoices.c.lifecycle).where(and_(
ar_invoices.c.lifecycle.in_(("issued", "void")), ar_invoices.c.source_kind != "odoo"))
return [(i, lc) for i, lc in self.conn.execute(q)]
def billed(self, sales_order_id_prefix: str) -> list:
"""(sales_order_id, so_line_no, quantity, net) of ISSUED invoice lines for an order, to net against credits."""
q = (select(ar_invoice_lines.c.sales_order_id, ar_invoice_lines.c.so_line_no, ar_invoice_lines.c.quantity,
ar_invoice_lines.c.net_amount)
.select_from(ar_invoice_lines.join(ar_invoices, ar_invoices.c.invoice_id == ar_invoice_lines.c.invoice_id))
.where(and_(ar_invoices.c.lifecycle == "issued", ar_invoice_lines.c.sales_order_id == sales_order_id_prefix)))
return [(a, b, _dec(c) or Decimal(0), _dec(d) or Decimal(0)) for a, b, c, d in self.conn.execute(q)]
def billed_lines(self, bp_id: int, issued_from, issued_to, include_drafts: bool) -> list:
"""v6: the lines of one customer's ISSUED invoices dated in the window, plus (include_drafts) every DRAFT invoice line, which
has no issue date and so ignores the window. Rows: (invoice_id, status, line_no, issue_date, product_id, net_amount,
tax_category, sales_order_id, so_line_no)."""
cond = [ar_invoices.c.lifecycle == "issued"]
if issued_from:
cond.append(ar_invoices.c.issue_date >= issued_from)
if issued_to:
cond.append(ar_invoices.c.issue_date <= issued_to)
where = and_(*cond)
if include_drafts:
where = or_(where, ar_invoices.c.lifecycle == "draft")
q = (select(ar_invoices.c.invoice_id, ar_invoices.c.lifecycle, ar_invoice_lines.c.line_no, ar_invoices.c.issue_date,
ar_invoice_lines.c.product_id, ar_invoice_lines.c.net_amount, ar_invoice_lines.c.tax_category,
ar_invoice_lines.c.sales_order_id, ar_invoice_lines.c.so_line_no)
.select_from(ar_invoice_lines.join(ar_invoices, ar_invoices.c.invoice_id == ar_invoice_lines.c.invoice_id))
.where(and_(ar_invoices.c.bp_id == bp_id, where, or_(ar_invoice_lines.c.deducts_invoice_id.is_(None),
ar_invoice_lines.c.deducts_invoice_id == ""))) # v7: a deduction is not a billed line
.order_by(ar_invoices.c.invoice_id, ar_invoice_lines.c.line_no))
return [tuple(r) for r in self.conn.execute(q)]
def migration_in_progress(self, bp_id: int) -> bool:
"""An unsigned migration run holds documents for this customer: invoicing takes writes for them once it is signed."""
q = (select(func.count()).select_from(ar_invoices.join(ar_migration_run, ar_migration_run.c.run_id == ar_invoices.c.migration_run_id))
.where(and_(ar_invoices.c.bp_id == bp_id, ar_migration_run.c.status.in_(("loaded", "reconciled")))))
return int(self.conn.execute(q).scalar() or 0) > 0
class CreditNoteRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, credit_id) -> Optional[tuple]:
rows = CREDIT_NOTE.load(self.conn, {"credit_id": credit_id})
if rows is None:
return None
self._snapshots[credit_id] = rows.snapshot
return self._build(rows)
def get_many(self, ids) -> dict:
"""{id: (aggregate, version)} for a set of ids: one query per table per 100 roots (a loop of get() is one round trip per root)."""
out = {}
for (rid,), rows in CREDIT_NOTE.load_many(self.conn, [{"credit_id": i} for i in dict.fromkeys(ids)]).items():
if rows is None:
continue
self._snapshots[rid] = rows.snapshot
out[rid] = self._build(rows)
return out
def _build(self, rows) -> tuple:
cn = _make(CreditNote, rows["ar_credit_notes"][0], ar_credit_notes)
_fix_money(cn, ("subtotal", "tax_total", "total"))
cn.lines = []
for r in sorted(rows["ar_credit_note_lines"], key=lambda r: r["line_no"]):
ln = _make(Line, r, ar_credit_note_lines)
_fix_money(ln, LINE_DEC)
cn.lines.append(ln)
cn.tax = []
for r in sorted(rows["ar_credit_note_tax"], key=lambda r: r["tax_code"]):
t = _make(TaxRow, r, ar_credit_note_tax)
_fix_money(t, TAX_DEC)
cn.tax.append(t)
return cn, CREDIT_NOTE.version_of(rows)
def save(self, cn: CreditNote, expected_version: int) -> int:
snapshot = self._snapshots.pop(cn.credit_id, None) # spent: after this save the rows are no longer what it read
return CREDIT_NOTE.save(self.conn, {
"ar_credit_notes": [{**_pick(cn, ar_credit_notes), "approved_at": _utc(cn.approved_at)}],
"ar_credit_note_lines": [{**{n: getattr(ln, n) for n in _names(ar_credit_note_lines) if n != "credit_id"},
"credit_id": cn.credit_id} for ln in cn.lines],
"ar_credit_note_tax": [{**_pick(t, ar_credit_note_tax), "credit_id": cn.credit_id} for t in cn.tax]},
expected_version=expected_version, snapshot=snapshot)
def exists(self, credit_id) -> bool:
return self.conn.execute(select(ar_credit_notes.c.credit_id).where(ar_credit_notes.c.credit_id == credit_id)).first() is not None
def credited_total(self, invoice_id: str) -> Decimal:
"""What approved credit notes already take off an invoice."""
q = select(func.coalesce(func.sum(ar_credit_notes.c.total), 0)).where(and_(
ar_credit_notes.c.invoice_id == invoice_id, ar_credit_notes.c.status == "approved"))
return Decimal(str(self.conn.execute(q).scalar() or 0))
def committed_count(self, invoice_id: str) -> int:
"""How many credit notes are approved against the invoice, as this transaction's snapshot has them."""
return len(self.approved_for(invoice_id))
def claim_slot(self, invoice_id: str, credit_id: str, at: datetime) -> int:
"""The invoice's next credit slot, n = committed credit notes + 1, with a PLAIN INSERT keyed (invoice, n): two approvals
that both saw k credits both insert n = k+1 and the second to commit fails 23505, is re-run, and meets the first's credit
in the within_invoice gate. The credits on one invoice are serialised by THIS key, not by a number series."""
n = self.committed_count(invoice_id) + 1
self.conn.execute(insert(ar_credit_claim).values(invoice_id=invoice_id, n=n, credit_id=credit_id, claimed_at=_utc(at), **_stamp()))
return n
def drafts_for(self, invoice_id: str) -> list:
"""Draft credit notes that name the invoice: voiding it would leave them approvable against nothing."""
q = select(ar_credit_notes.c.credit_id).where(and_(ar_credit_notes.c.invoice_id == invoice_id,
ar_credit_notes.c.status == "draft"))
return [r[0] for r in self.conn.execute(q)]
def approved_for(self, invoice_id: str) -> list:
q = select(ar_credit_notes.c.credit_id).where(and_(ar_credit_notes.c.invoice_id == invoice_id,
ar_credit_notes.c.status == "approved"))
return [r[0] for r in self.conn.execute(q)]
def page(self, *, status=None, bp_id=None, invoice_id=None, limit=50, offset=0) -> list:
cond = [c == v for c, v in ((ar_credit_notes.c.status, status), (ar_credit_notes.c.bp_id, bp_id),
(ar_credit_notes.c.invoice_id, invoice_id)) if v is not None]
q = select(ar_credit_notes).order_by(ar_credit_notes.c.credit_id).limit(limit).offset(offset)
if cond:
q = q.where(and_(*cond))
return [dict(r._mapping) for r in self.conn.execute(q)]
def in_period_unexported(self, start: date, end: date) -> list:
q = select(ar_credit_notes).where(and_(ar_credit_notes.c.status == "approved", ar_credit_notes.c.credit_date >= start,
ar_credit_notes.c.credit_date <= end, ar_credit_notes.c.source_kind != "odoo"))
return [dict(r._mapping) for r in self.conn.execute(q) if not r._mapping["export_batch_id"]]
def approved_ids(self) -> list:
q = select(ar_credit_notes.c.credit_id).where(and_(ar_credit_notes.c.status == "approved", ar_credit_notes.c.source_kind != "odoo"))
return [r[0] for r in self.conn.execute(q)]
def billed_lines(self, bp_id: int, issued_from, issued_to) -> list:
"""v6: the lines of one customer's APPROVED credit notes dated in the window (credit_date)."""
cond = [ar_credit_notes.c.bp_id == bp_id, ar_credit_notes.c.status == "approved"]
if issued_from:
cond.append(ar_credit_notes.c.credit_date >= issued_from)
if issued_to:
cond.append(ar_credit_notes.c.credit_date <= issued_to)
q = (select(ar_credit_notes.c.credit_id, ar_credit_note_lines.c.line_no, ar_credit_notes.c.credit_date,
ar_credit_note_lines.c.product_id, ar_credit_note_lines.c.net_amount, ar_credit_note_lines.c.tax_category,
ar_credit_note_lines.c.sales_order_id, ar_credit_note_lines.c.so_line_no)
.select_from(ar_credit_note_lines.join(ar_credit_notes, ar_credit_notes.c.credit_id == ar_credit_note_lines.c.credit_id))
.where(and_(*cond)).order_by(ar_credit_notes.c.credit_id, ar_credit_note_lines.c.line_no))
return [tuple(r) for r in self.conn.execute(q)]
def billed(self, sales_order_id: str) -> list:
q = (select(ar_credit_note_lines.c.sales_order_id, ar_credit_note_lines.c.so_line_no, ar_credit_note_lines.c.quantity,
ar_credit_note_lines.c.net_amount)
.select_from(ar_credit_note_lines.join(ar_credit_notes, ar_credit_notes.c.credit_id == ar_credit_note_lines.c.credit_id))
.where(and_(ar_credit_notes.c.status == "approved", ar_credit_note_lines.c.sales_order_id == sales_order_id)))
return [(a, b, _dec(c) or Decimal(0), _dec(d) or Decimal(0)) for a, b, c, d in self.conn.execute(q)]
class WriteoffRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, writeoff_id) -> Optional[tuple]:
rows = WRITEOFF.load(self.conn, {"writeoff_id": writeoff_id})
if rows is None:
return None
self._snapshots[writeoff_id] = rows.snapshot
w = _make(Writeoff, rows["ar_writeoffs"][0], ar_writeoffs)
w.amount = Decimal(str(w.amount))
return w, WRITEOFF.version_of(rows)
def save(self, w: Writeoff, expected_version: int) -> int:
snapshot = self._snapshots.pop(w.writeoff_id, None) # spent: after this save the rows are no longer what it read
return WRITEOFF.save(self.conn, {"ar_writeoffs": [{**_pick(w, ar_writeoffs), "approved_at": _utc(w.approved_at)}]},
expected_version=expected_version, snapshot=snapshot)
def for_invoice(self, invoice_id: str) -> list:
q = select(ar_writeoffs).where(ar_writeoffs.c.invoice_id == invoice_id).order_by(ar_writeoffs.c.writeoff_id)
return [dict(r._mapping) for r in self.conn.execute(q)]
def approved_ids(self) -> list:
return [r[0] for r in self.conn.execute(select(ar_writeoffs.c.writeoff_id).where(ar_writeoffs.c.status == "approved"))]
class ScheduleRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, schedule_id) -> Optional[tuple]:
rows = SCHEDULE.load(self.conn, {"schedule_id": schedule_id})
if rows is None:
return None
self._snapshots[schedule_id] = rows.snapshot
return self._build(rows)
def get_many(self, ids) -> dict:
"""{id: (aggregate, version)} for a set of ids: one query per table per 100 roots (a loop of get() is one round trip per root)."""
out = {}
for (rid,), rows in SCHEDULE.load_many(self.conn, [{"schedule_id": i} for i in dict.fromkeys(ids)]).items():
if rows is None:
continue
self._snapshots[rid] = rows.snapshot
out[rid] = self._build(rows)
return out
def _build(self, rows) -> tuple:
s = _make(Schedule, rows["ar_schedule"][0], ar_schedule)
for n in ("quantity", "unit_price", "discount_pct"):
v = rows["ar_schedule"][0].get(n)
setattr(s, n, None if v is None else Decimal(str(v)))
return s, SCHEDULE.version_of(rows)
def save(self, s: Schedule, expected_version: int) -> int:
snapshot = self._snapshots.pop(s.schedule_id, None) # spent: after this save the rows are no longer what it read
return SCHEDULE.save(self.conn, {"ar_schedule": [_pick(s, ar_schedule)]}, expected_version=expected_version, snapshot=snapshot)
def ids(self, *, status=None, source_id=None, limit=500, offset=0) -> list:
q = select(ar_schedule.c.schedule_id).order_by(ar_schedule.c.schedule_id).limit(limit).offset(offset)
cond = [c == v for c, v in ((ar_schedule.c.status, status), (ar_schedule.c.source_id, source_id)) if v is not None]
if cond:
q = q.where(and_(*cond))
return [r[0] for r in self.conn.execute(q)]
class CustomerRepository:
"""Billing profiles, as dated rows: only ever inserted."""
def __init__(self, conn):
self.conn = conn
def rows(self, bp_id: int) -> list:
"""The customer's dated profiles, read once per request (add() forgets them)."""
memo = self.__dict__.setdefault("_rows", {})
if bp_id not in memo:
q = select(ar_customer).where(ar_customer.c.bp_id == bp_id).order_by(ar_customer.c.valid_from)
memo[bp_id] = [dict(r._mapping) for r in self.conn.execute(q)]
return [_make(customer_mod.Profile, dict(m), ar_customer) for m in memo[bp_id]]
def add(self, p: customer_mod.Profile) -> None:
"""A plain INSERT: a second row for the same day fails on the key, and a row is never edited."""
self.__dict__.get("_rows", {}).pop(p.bp_id, None)
self.conn.execute(insert(ar_customer).values(**{**_pick(p, ar_customer), "set_at": _utc(p.set_at)}, **_stamp()))
class TermsRepository:
def __init__(self, conn):
self.conn = conn
def all(self, active_only: bool = False) -> list:
if self.__dict__.get("_all") is None: # read once per request (put() forgets it)
self.__dict__["_all"] = [dict(r._mapping) for r in self.conn.execute(select(ar_terms).order_by(ar_terms.c.terms_id))]
rows = [dict(r) for r in self.__dict__["_all"]]
for r in rows:
r["early_discount_pct"] = _dec(r["early_discount_pct"]) or Decimal(0)
r["early_discount_days"] = r.get("early_discount_days") or 0
return [r for r in rows if r["active"] or not active_only]
def get(self, terms_id: str) -> Optional[dict]:
return next((r for r in self.all() if r["terms_id"] == terms_id), None)
def put(self, row: dict) -> None:
self.__dict__["_all"] = None
upsert(self.conn, ar_terms, [row], ("terms_id",))
def used(self, terms_id: str) -> bool:
return self.conn.execute(select(ar_invoices.c.invoice_id).where(and_(
ar_invoices.c.terms_id == terms_id, ar_invoices.c.lifecycle == "issued")).limit(1)).first() is not None
class SettingsRepository:
"""Dated setting rows: the value in force on a day is the row with the latest valid_from on or before it."""
def __init__(self, conn):
self.conn = conn
def rows(self, key: Optional[str] = None) -> list:
"""Every setting row in ONE read, once per request (add() forgets them), filtered here."""
if self.__dict__.get("_all") is None:
q = select(ar_settings).order_by(ar_settings.c.setting_key, ar_settings.c.valid_from)
self.__dict__["_all"] = [{"setting_key": r._mapping["setting_key"], "valid_from": r._mapping["valid_from"],
"value": json.loads(r._mapping["value"]) if r._mapping["value"] is not None else None,
"set_by": r._mapping["set_by"]} for r in self.conn.execute(q)]
return [copy.deepcopy(r) for r in self.__dict__["_all"] if not key or r["setting_key"] == key]
def in_force(self, on: date) -> Settings:
values = {}
for r in self.rows():
if r["valid_from"] <= on and r["setting_key"] in SETTINGS:
values[r["setting_key"]] = r["value"] # rows come in valid_from order: the latest wins
return Settings.of(values)
def has_any(self, key: str) -> bool:
return bool(self.rows(key))
def add(self, key: str, valid_from: date, value, by: str, at: datetime) -> None:
"""A plain INSERT: a row is never edited; the same key and day twice fails on the primary key."""
self.__dict__["_all"] = None
self.conn.execute(insert(ar_settings).values(setting_key=key, valid_from=valid_from, value=json.dumps(value),
set_by=by, set_at=_utc(at), **_stamp()))
class AccountMapRepository:
def __init__(self, conn):
self.conn = conn
def rows(self) -> list:
if self.__dict__.get("_all") is None: # read once per request (add() forgets it)
self.__dict__["_all"] = [dict(r._mapping) for r in self.conn.execute(select(ar_account_map).order_by(
ar_account_map.c.purpose, ar_account_map.c.tax_code, ar_account_map.c.valid_from))]
return [dict(r) for r in self.__dict__["_all"]]
def has_any(self, purpose: str, tax_code: str) -> bool:
return any(r["purpose"] == purpose and r["tax_code"] == tax_code for r in self.rows())
def add(self, purpose: str, tax_code: str, valid_from: date, account_id: int, by: str) -> None:
self.__dict__["_all"] = None
self.conn.execute(insert(ar_account_map).values(purpose=purpose, tax_code=tax_code, valid_from=valid_from,
account_id=account_id, set_by=by, **_stamp()))
def account(self, purpose: str, tax_code: str, on: date) -> Optional[int]:
"""The account for a purpose (and tax code) in force on the document's own date."""
best = None
for r in self.rows():
if r["purpose"] == purpose and r["tax_code"] == (tax_code or "") and r["valid_from"] <= on:
best = r["account_id"]
return best
class NumberRepository:
def __init__(self, conn):
self.conn = conn
def claim(self, series_id: str, document_id: str, at: datetime, free=lambda seq: True, floor: int = 0) -> int:
"""The next counter in the series: the high-water mark, the highest claim or `floor` (the highest counter a number already
in use renders in this series - Odoo's carried ones included), plus one - so a new number never fills a gap below the
mark. Written with a PLAIN INSERT (never an upsert): the key is the uniqueness guard, so two issues racing for one
number cannot both commit."""
mark = self.conn.execute(select(ar_number_series.c.high_water_mark).where(ar_number_series.c.series_id == series_id)).scalar() or 0
top = self.conn.execute(select(func.max(ar_number_claims.c.seq)).where(ar_number_claims.c.series_id == series_id)).scalar() or 0
seq = max(int(mark), int(top), int(floor)) + 1
while not free(seq):
seq += 1
self.conn.execute(insert(ar_number_claims).values(series_id=series_id, seq=seq, document_id=document_id, claimed_at=_utc(at), **_stamp()))
upsert(self.conn, ar_number_series, [{"series_id": series_id, "high_water_mark": seq}], ("series_id",))
return seq
def numbers_like(self, prefix: str) -> list:
"""Every number in use (ours and carried) whose text starts with `prefix`."""
esc = prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
q = select(ar_invoice_no_claims.c.invoice_no).where(ar_invoice_no_claims.c.invoice_no.like(esc + "%", escape="\\"))
return [r[0] for r in self.conn.execute(q)]
def taken(self, number: str) -> bool:
return self.conn.execute(select(ar_invoice_no_claims.c.invoice_no).where(ar_invoice_no_claims.c.invoice_no == number)).first() is not None
def register(self, number: str, document_id: str, at: datetime) -> bool:
"""The number belongs to this document, with a PLAIN INSERT keyed on the number: no two documents share one. True when it is
(now) this document's, False when another document already holds it."""
row = self.conn.execute(select(ar_invoice_no_claims.c.document_id).where(ar_invoice_no_claims.c.invoice_no == number)).first()
if row is not None:
return row[0] == document_id
self.conn.execute(insert(ar_invoice_no_claims).values(invoice_no=number, document_id=document_id, claimed_at=_utc(at), **_stamp()))
return True
def raise_mark(self, series_id: str, seq: int) -> None:
"""A migrated number: the next issue continues the series after the highest carried."""
mark = self.conn.execute(select(ar_number_series.c.high_water_mark).where(ar_number_series.c.series_id == series_id)).scalar() or 0
if seq > int(mark):
upsert(self.conn, ar_number_series, [{"series_id": series_id, "high_water_mark": seq}], ("series_id",))
def mark(self, series_id: str) -> int:
return int(self.conn.execute(select(ar_number_series.c.high_water_mark).where(ar_number_series.c.series_id == series_id)).scalar() or 0)
class ExportRepository:
def __init__(self, conn):
self.conn = conn
def get(self, batch_id: str) -> Optional[dict]:
r = self.conn.execute(select(ar_export_batch).where(ar_export_batch.c.batch_id == batch_id)).first()
return dict(r._mapping) if r else None
def put(self, row: dict) -> None:
upsert(self.conn, ar_export_batch, [row], ("batch_id",))
def list(self) -> list:
q = select(ar_export_batch.c.batch_id, ar_export_batch.c.period, ar_export_batch.c.format,
ar_export_batch.c.document_count, ar_export_batch.c.file_key).order_by(ar_export_batch.c.batch_id)
return [dict(r._mapping) for r in self.conn.execute(q)]
class PostingRepository:
def __init__(self, conn):
self.conn = conn
def get(self, document_id: str, event: str) -> Optional[dict]:
r = self.conn.execute(select(ar_postings).where(and_(ar_postings.c.document_id == document_id,
ar_postings.c.event == event))).first()
return dict(r._mapping) if r else None
def record(self, document_id: str, event: str, journal_id: int, journal_date: date, by: str) -> None:
upsert(self.conn, ar_postings, [{"document_id": document_id, "event": event, "journal_id": journal_id,
"journal_date": journal_date, "posted_by": by}], ("document_id", "event"))
class MigrationRunRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, run_id: str) -> Optional[tuple]:
rows = MIGRATION_RUN.load(self.conn, {"run_id": run_id})
if rows is None:
return None
self._snapshots[run_id] = rows.snapshot
run = _make(MigrationRun, rows["ar_migration_run"][0], ar_migration_run)
run.checks = [MigrationCheck(**{n: r[n] for n in _names(ar_migration_check) if r.get(n) is not None})
for r in sorted(rows["ar_migration_check"], key=lambda r: (r["check_key"], r["slice"]))]
for c in run.checks:
c.equal = bool(c.equal)
return run, MIGRATION_RUN.version_of(rows)
def save(self, run: MigrationRun, expected_version: int) -> int:
snapshot = self._snapshots.pop(run.run_id, None) # spent: after this save the rows are no longer what it read
checks = [{**_pick(c, ar_migration_check), "run_id": run.run_id} for c in run.checks]
return MIGRATION_RUN.save(self.conn, {
"ar_migration_run": [{**_pick(run, ar_migration_run), "loaded_at": _utc(run.loaded_at), "signed_at": _utc(run.signed_at)}],
"ar_migration_check": checks}, expected_version=expected_version, snapshot=snapshot)
def list(self) -> list:
q = select(ar_migration_run.c.run_id, ar_migration_run.c.status, ar_migration_run.c.cutover_date).order_by(ar_migration_run.c.run_id)
return [dict(r._mapping) for r in self.conn.execute(q)]
def invoice_ids(self, run_id: str) -> list:
return [r[0] for r in self.conn.execute(select(ar_invoices.c.invoice_id).where(ar_invoices.c.migration_run_id == run_id))]
def credit_ids(self, run_id: str) -> list:
return [r[0] for r in self.conn.execute(select(ar_credit_notes.c.credit_id).where(ar_credit_notes.c.migration_run_id == run_id))]
def remove_documents(self, run_id: str) -> int:
"""Discard: the run's documents go (nothing was ever issued from them by this brick)."""
n = 0
for iid in self.invoice_ids(run_id):
for t in (ar_invoice_lines, ar_invoice_tax, ar_sends, ar_ai_finding):
self.conn.execute(delete(t).where(t.c.invoice_id == iid))
self.conn.execute(delete(ar_invoices).where(ar_invoices.c.invoice_id == iid))
self.conn.execute(delete(INVOICE.versions).where(INVOICE.versions.c.invoice_id == iid))
n += 1
for cid in self.credit_ids(run_id):
for t in (ar_credit_note_lines, ar_credit_note_tax):
self.conn.execute(delete(t).where(t.c.credit_id == cid))
self.conn.execute(delete(ar_credit_notes).where(ar_credit_notes.c.credit_id == cid))
self.conn.execute(delete(CREDIT_NOTE.versions).where(CREDIT_NOTE.versions.c.credit_id == cid))
n += 1
return n
class MigrationMapRepository:
"""(odoo db, odoo model, odoo id) -> our natural key (shared.yaml conventions.odoo_migration.id_map)."""
def __init__(self, conn):
self.conn = conn
def put(self, db: str, model: str, odoo_id, natural_key: str, run_id: str) -> None:
upsert(self.conn, ar_migration_map, [{"odoo_db": db, "odoo_model": model, "odoo_id": str(odoo_id),
"natural_key": natural_key, "run_id": run_id}], ("odoo_db", "odoo_model", "odoo_id"))
def get(self, db: str, model: str, odoo_id) -> Optional[str]:
return self.conn.execute(select(ar_migration_map.c.natural_key).where(and_(
ar_migration_map.c.odoo_db == db, ar_migration_map.c.odoo_model == model,
ar_migration_map.c.odoo_id == str(odoo_id)))).scalar()
def existing(self, db: str, model: str) -> dict:
q = select(ar_migration_map.c.odoo_id, ar_migration_map.c.natural_key).where(and_(
ar_migration_map.c.odoo_db == db, ar_migration_map.c.odoo_model == model))
return {r[0]: r[1] for r in self.conn.execute(q)}
class PaymentRequestRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, request_id) -> Optional[tuple]:
rows = PAYMENT_REQUEST.load(self.conn, {"request_id": request_id})
if rows is None:
return None
self._snapshots[request_id] = rows.snapshot
pr = _make(PaymentRequest, rows["ar_payment_request"][0], ar_payment_request)
_fix_money(pr, ("net_amount", "tax_amount_display", "total"))
return pr, PAYMENT_REQUEST.version_of(rows)
def save(self, pr: PaymentRequest, expected_version: int) -> int:
snapshot = self._snapshots.pop(pr.request_id, None) # spent: after this save the rows are no longer what it read
return PAYMENT_REQUEST.save(self.conn, {"ar_payment_request": [_pick(pr, ar_payment_request)]}, expected_version=expected_version, snapshot=snapshot)
def received(self, request_id: str) -> Decimal:
q = select(func.coalesce(func.sum(ar_request_settlement.c.amount), 0)).where(ar_request_settlement.c.request_id == request_id)
return Decimal(str(self.conn.execute(q).scalar() or 0))
def settlements(self, request_id: str) -> list:
q = select(ar_request_settlement).where(ar_request_settlement.c.request_id == request_id).order_by(ar_request_settlement.c.n)
return [dict(r._mapping) for r in self.conn.execute(q)]
def settlement_of(self, payment_id: str) -> Optional[dict]:
r = self.conn.execute(select(ar_request_settlement).where(ar_request_settlement.c.payment_id == payment_id)).first()
return dict(r._mapping) if r else None
def record_settlement(self, payment_id: str, request_id: str, n: int, received_on: date, amount: Decimal, invoice_id: str,
at: datetime) -> None:
"""A PLAIN INSERT keyed on the payment, after the caller has checked it is not there: never swallowed."""
self.conn.execute(insert(ar_request_settlement).values(payment_id=payment_id, request_id=request_id, n=n, received_on=received_on,
amount=amount, deposit_invoice_id=invoice_id, settled_at=_utc(at), **_stamp()))
def page(self, *, bp_id=None, limit=100) -> list:
q = select(ar_payment_request.c.request_id)
if bp_id is not None:
q = q.where(ar_payment_request.c.bp_id == bp_id)
return [r[0] for r in self.conn.execute(q.order_by(ar_payment_request.c.request_id).limit(limit))]
class DepositRepository:
"""What a deposit invoice has already given up to final invoices (claims) and to credit notes."""
def __init__(self, conn):
self.conn = conn
def claimed(self, deposit_id: str) -> Decimal:
"""The deductions that stand: claims of finals that are ISSUED - a void final gives its deduction back."""
q = (select(func.coalesce(func.sum(ar_deposit_claim.c.net), 0))
.select_from(ar_deposit_claim.join(ar_invoices, ar_invoices.c.invoice_id == ar_deposit_claim.c.final_invoice_id))
.where(and_(ar_deposit_claim.c.deposit_invoice_id == deposit_id, ar_invoices.c.lifecycle == "issued")))
return Decimal(str(self.conn.execute(q).scalar() or 0))
def standing(self, deposit_id: str) -> list:
"""The issued finals whose deduction stands against the deposit: [(final invoice id, net)]."""
q = (select(ar_deposit_claim.c.final_invoice_id, ar_deposit_claim.c.net)
.select_from(ar_deposit_claim.join(ar_invoices, ar_invoices.c.invoice_id == ar_deposit_claim.c.final_invoice_id))
.where(and_(ar_deposit_claim.c.deposit_invoice_id == deposit_id, ar_invoices.c.lifecycle == "issued")))
return [(r[0], Decimal(str(r[1]))) for r in self.conn.execute(q)]
def count(self, deposit_id: str) -> int:
q = select(func.count()).select_from(ar_deposit_claim).where(ar_deposit_claim.c.deposit_invoice_id == deposit_id)
return int(self.conn.execute(q).scalar() or 0)
def claim(self, deposit_id: str, final_id: str, net: Decimal, at: datetime, count: Optional[int] = None) -> int:
"""Slot n = deductions so far + 1, a PLAIN INSERT keyed (deposit, n): two finals that both saw k deductions both insert
n = k+1 and the second to commit fails 23505, is re-run, and meets the first in the deposit_remaining gate."""
n = (self.count(deposit_id) if count is None else count) + 1
self.conn.execute(insert(ar_deposit_claim).values(deposit_invoice_id=deposit_id, n=n, final_invoice_id=final_id, net=net,
claimed_at=_utc(at), **_stamp()))
return n
def credited_net(self, deposit_id: str) -> Decimal:
q = select(func.coalesce(func.sum(ar_credit_notes.c.subtotal), 0)).where(and_(
ar_credit_notes.c.invoice_id == deposit_id, ar_credit_notes.c.status == "approved"))
return Decimal(str(self.conn.execute(q).scalar() or 0))
def deduction_of(self, final_id: str) -> list:
q = select(ar_deposit_claim).where(ar_deposit_claim.c.final_invoice_id == final_id)
return [dict(r._mapping) for r in self.conn.execute(q)]
class InterestRateRepository:
"""The statutory reference rate, as dated rows (from 30 June and 31 December)."""
def __init__(self, conn):
self.conn = conn
def add(self, valid_from: date, reference_pct: Decimal, margin_pct: Decimal, source: str, by: str) -> None:
self.conn.execute(insert(ar_interest_rate).values(valid_from=valid_from, reference_rate_pct=reference_pct, margin_pct=margin_pct,
source=source, set_by=by, **_stamp()))
def all(self) -> list:
return [{**dict(r._mapping), "reference_rate_pct": Decimal(str(r.reference_rate_pct)), "margin_pct": Decimal(str(r.margin_pct))}
for r in self.conn.execute(select(ar_interest_rate).order_by(ar_interest_rate.c.valid_from))]
def in_force(self, on: date) -> Optional[dict]:
rows = [r for r in self.all() if r["valid_from"] <= on]
return rows[-1] if rows else None
class HandoverRepository:
def __init__(self, conn):
self.conn = conn
def get(self, document_id: str) -> Optional[dict]:
r = self.conn.execute(select(ar_ic_handover).where(ar_ic_handover.c.document_id == document_id)).first()
return dict(r._mapping) if r else None
def record(self, document_id: str, bill_id: str, partner: str, at: datetime) -> None:
self.conn.execute(insert(ar_ic_handover).values(document_id=document_id, bill_id=bill_id, partner_company=partner,
handed_at=_utc(at), **_stamp()))
def credits_for(self, bp_id: int) -> list:
return []
class SettlementDiscountRepository:
def __init__(self, conn):
self.conn = conn
def get(self, payment_id: str) -> Optional[dict]:
r = self.conn.execute(select(ar_settlement_discount).where(ar_settlement_discount.c.payment_id == payment_id)).first()
return dict(r._mapping) if r else None
def record(self, payment_id: str, invoice_id: str, credit_id: str, at: datetime) -> None:
self.conn.execute(insert(ar_settlement_discount).values(payment_id=payment_id, invoice_id=invoice_id, credit_id=credit_id,
claimed_at=_utc(at), **_stamp()))
class ReleaseRepository:
"""Deferred revenue released per (invoice, line, period) and taken back by credit notes. Plain INSERTs: the key is the guard."""
def __init__(self, conn):
self.conn = conn
def get(self, invoice_id: str, line_no: int, period_id: str) -> Optional[dict]:
r = self.conn.execute(select(ar_deferral_release).where(and_(
ar_deferral_release.c.invoice_id == invoice_id, ar_deferral_release.c.line_no == line_no,
ar_deferral_release.c.period_id == period_id))).first()
return dict(r._mapping) if r else None
def add(self, invoice_id: str, line_no: int, period_id: str, kind: str, amount: Decimal, journal_date: date, at: datetime) -> None:
self.conn.execute(insert(ar_deferral_release).values(invoice_id=invoice_id, line_no=line_no, period_id=period_id, kind=kind,
amount=amount, journal_date=journal_date, released_at=_utc(at), **_stamp()))
def rows(self, invoice_id: str) -> list:
q = select(ar_deferral_release).where(ar_deferral_release.c.invoice_id == invoice_id).order_by(
ar_deferral_release.c.line_no, ar_deferral_release.c.period_id)
return [{**dict(r._mapping), "amount": Decimal(str(r.amount))} for r in self.conn.execute(q)]
def any_for(self, invoice_id: str) -> bool:
return self.conn.execute(select(ar_deferral_release.c.invoice_id).where(and_(
ar_deferral_release.c.invoice_id == invoice_id, ar_deferral_release.c.kind != "guard"))).first() is not None
def guard(self, invoice_id: str, at: datetime) -> int:
"""The invoice's next serialisation slot, n = committed guards + 1, a PLAIN INSERT keyed (invoice, 0, "guard:n"): a release, a
credit take-back and a void of ONE invoice all claim it, so two that overlap both insert n = k+1 and the second to commit fails
23505 (-> 409) and is re-run against what the first left."""
q = select(func.count()).select_from(ar_deferral_release).where(and_(
ar_deferral_release.c.invoice_id == invoice_id, ar_deferral_release.c.kind == "guard"))
n = int(self.conn.execute(q).scalar() or 0) + 1
self.add(invoice_id, 0, f"guard:{n:06d}", "guard", Decimal("0"), _utc(at).date(), at)
return n
def totals(self, invoice_id: str) -> dict:
"""{line_no: [released, taken back]} read straight from the claims (never a cached snapshot)."""
out: dict = {}
q = select(ar_deferral_release.c.line_no, ar_deferral_release.c.kind, func.coalesce(func.sum(ar_deferral_release.c.amount), 0)).where(
ar_deferral_release.c.invoice_id == invoice_id).group_by(ar_deferral_release.c.line_no, ar_deferral_release.c.kind)
for ln, kind, amt in self.conn.execute(q):
if kind in ("release", "credit"):
out.setdefault(ln, [Decimal(0), Decimal(0)])[0 if kind == "release" else 1] = Decimal(str(amt))
return out
def of_credit(self, credit_id: str) -> list:
q = select(ar_deferral_release).where(ar_deferral_release.c.period_id == f"cn:{credit_id}").order_by(ar_deferral_release.c.line_no)
return [{**dict(r._mapping), "amount": Decimal(str(r.amount))} for r in self.conn.execute(q)]
def candidates(self, last_day: date) -> list:
"""Issued invoices with a line whose service has begun by `last_day` and runs past the issue month: the release pass reads them."""
q = (select(ar_invoices.c.invoice_id).distinct().select_from(ar_invoices.join(
ar_invoice_lines, ar_invoice_lines.c.invoice_id == ar_invoices.c.invoice_id)).where(and_(
ar_invoices.c.lifecycle == "issued", ar_invoices.c.source_kind != "odoo", ar_invoice_lines.c.service_from <= last_day,
ar_invoice_lines.c.service_to.isnot(None))).order_by(ar_invoices.c.invoice_id))
return [r[0] for r in self.conn.execute(q)]
class ProvisionRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, run_id: str) -> Optional[tuple]:
rows = PROVISION_RUN.load(self.conn, {"run_id": run_id})
if rows is None:
return None
self._snapshots[run_id] = rows.snapshot
run = _make(ProvisionRun, rows["ar_provision_run"][0], ar_provision_run)
_fix_money(run, ("total", "previous_total", "movement"))
return run, PROVISION_RUN.version_of(rows)
def save(self, run: ProvisionRun, expected_version: int) -> int:
snapshot = self._snapshots.pop(run.run_id, None) # spent: after this save the rows are no longer what it read
return PROVISION_RUN.save(self.conn, {"ar_provision_run": [{**_pick(run, ar_provision_run), "approved_at": _utc(run.approved_at)}]},
expected_version=expected_version, snapshot=snapshot)
def ids(self, limit: int = 200) -> list:
return [r[0] for r in self.conn.execute(select(ar_provision_run.c.run_id).order_by(ar_provision_run.c.run_id.desc()).limit(limit))]
def approved_from(self, company_code: str, period_id: str) -> Optional[str]:
"""The latest APPROVED run's period at or after `period_id` (the allowance already carries that month or a later one)."""
q = select(ar_provision_run.c.period_id).where(and_(ar_provision_run.c.company_code == company_code, ar_provision_run.c.lifecycle == "approved",
ar_provision_run.c.period_id >= period_id)).order_by(ar_provision_run.c.period_id.desc()).limit(1)
r = self.conn.execute(q).first()
return r[0] if r else None
def last_approved_before(self, company_code: str, period_id: str) -> Optional[ProvisionRun]:
"""The allowance the books carry going into `period_id`: the latest APPROVED run of an earlier period."""
q = select(ar_provision_run.c.run_id).where(and_(ar_provision_run.c.company_code == company_code, ar_provision_run.c.lifecycle == "approved",
ar_provision_run.c.period_id < period_id)).order_by(ar_provision_run.c.period_id.desc()).limit(1)
r = self.conn.execute(q).first()
return self.get(r[0])[0] if r else None
class ProvisionRateRepository:
"""The doubtful-debt percentage per aging bucket, as dated rows: only ever inserted."""
def __init__(self, conn):
self.conn = conn
def rows(self) -> list:
return [{**dict(r._mapping), "rate_pct": Decimal(str(r.rate_pct))} for r in self.conn.execute(
select(ar_provision_rate).order_by(ar_provision_rate.c.bucket, ar_provision_rate.c.valid_from))]
def has_any(self, bucket: str) -> bool:
return any(r["bucket"] == bucket for r in self.rows())
def add(self, bucket: str, valid_from: date, rate_pct: Decimal, by: str) -> None:
self.conn.execute(insert(ar_provision_rate).values(bucket=bucket, valid_from=valid_from, rate_pct=rate_pct, set_by=by, **_stamp()))
def in_force(self, bucket: str, on: date) -> Optional[Decimal]:
best = None
for r in self.rows():
if r["bucket"] == bucket and r["valid_from"] <= on:
best = r["rate_pct"]
return best
class EinvoiceRepository:
def __init__(self, conn):
self.conn = conn
def message(self, message_id: str) -> Optional[dict]:
r = self.conn.execute(select(ar_einvoice_message).where(ar_einvoice_message.c.message_id == message_id)).first()
return dict(r._mapping) if r else None
def add_message(self, message_id: str, document_id: str, kind: str, provider: str, at: datetime) -> None:
self.conn.execute(insert(ar_einvoice_message).values(message_id=message_id, document_id=document_id, document_kind=kind,
provider=provider, sent_at=_utc(at), **_stamp()))
def message_of(self, document_id: str) -> Optional[dict]:
r = self.conn.execute(select(ar_einvoice_message).where(ar_einvoice_message.c.document_id == document_id).order_by(
ar_einvoice_message.c.sent_at.desc()).limit(1)).first()
return dict(r._mapping) if r else None
def status(self, message_id: str, status: str) -> Optional[dict]:
r = self.conn.execute(select(ar_einvoice_status).where(and_(ar_einvoice_status.c.message_id == message_id,
ar_einvoice_status.c.status == status))).first()
return dict(r._mapping) if r else None
def add_status(self, message_id: str, status: str, at: datetime, reason: str) -> None:
self.conn.execute(insert(ar_einvoice_status).values(message_id=message_id, status=status, status_at=_utc(at), reason=reason, **_stamp()))
def statuses(self, message_id: str) -> list:
rows = [dict(r._mapping) for r in self.conn.execute(select(ar_einvoice_status).where(ar_einvoice_status.c.message_id == message_id))]
for r in rows:
r["status_at"] = _utc(r["status_at"]) # aware on the pg wire, naive on SQLite: compare as naive UTC
return sorted(rows, key=lambda r: (r["status_at"] is None, r["status_at"], r["status"]))
class Repository:
"""Every aggregate's repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.invoices = InvoiceRepository(conn)
self.credits = CreditNoteRepository(conn)
self.writeoffs = WriteoffRepository(conn)
self.schedules = ScheduleRepository(conn)
self.customers = CustomerRepository(conn)
self.terms = TermsRepository(conn)
self.settings = SettingsRepository(conn)
self.accounts = AccountMapRepository(conn)
self.numbers = NumberRepository(conn)
self.exports = ExportRepository(conn)
self.postings = PostingRepository(conn)
self.runs = MigrationRunRepository(conn)
self.maps = MigrationMapRepository(conn)
self.requests = PaymentRequestRepository(conn)
self.deposits = DepositRepository(conn)
self.rates = InterestRateRepository(conn)
self.handovers = HandoverRepository(conn)
self.discounts = SettlementDiscountRepository(conn)
self.releases = ReleaseRepository(conn)
self.provisions = ProvisionRepository(conn)
self.provision_rates = ProvisionRateRepository(conn)
self.einvoices = EinvoiceRepository(conn)
def seed(self) -> None:
"""The rows a new bucket starts with: one payment terms row (net 30 days)."""
if self.terms.get("NET30") is None:
self.terms.put({"terms_id": "NET30", "days": 30, "basis": "invoice_date", "description": "30 days net",
"early_discount_pct": Decimal(0), "active": True})