dodil/erp-brick-expenses
DODIL ERP brick: expenses
src/brick_expenses/infrastructure/tables.py
"""The tables, and one repository per concern. Every table is prefixed `exp_`, has a natural primary key (never an
autoincrement), and money is NUMERIC(18,2). schema.sql is written from the same column model; tests/test_schema.py keeps
the two (and the migrations) equal.
Rows that are FACTS (a line, an approval step, a claim, a leaver notice) are written with a plain INSERT: a second writer of
the same key is rejected (23505), never silently overwritten - DataK3's ON CONFLICT DO UPDATE rewrites every column in VALUES.
Only the few rows that move FORWARD (the report's status, an outbox row's status) are written with the keyed upsert.
Statements per request do not grow with the rows around them: a report loads in three queries however many lines it has,
lines are written in ONE statement, and every list is one query.
"""
import json
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from zoneinfo import ZoneInfo
from erp_platform import work
from erp_platform.repository import upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
delete,
func,
insert,
or_,
select,
)
from brick_expenses.domain import policy as pol
from brick_expenses.domain import rates as rt
from brick_expenses.domain import report as rp
from brick_expenses.domain.errors import RuleBroken
from brick_expenses.domain.money import money
from brick_expenses.domain.values import DEFAULT_CATEGORIES, DEFAULT_RATES, SETTING_DEFS
md = MetaData()
MONEY = Numeric(18, 2)
exp_report = Table("exp_report", md,
Column("report_id", String, primary_key=True),
Column("person_id", BigInteger),
Column("company_code", String),
Column("bp_id", BigInteger),
Column("claimant_email", String),
Column("claimant_sub", String),
Column("revision", Integer),
Column("title", String),
Column("currency", String),
Column("status", String),
Column("submitted_at", TIMESTAMP),
Column("submitted_by", String),
Column("submitted_sub", String),
Column("submitted_on", Date),
Column("chain_json", String),
Column("waiting_level", Integer),
Column("waiting_kind", String),
Column("waiting_email", String),
Column("waiting_sub", String),
Column("journal_id", BigInteger),
Column("payout_ref", String),
Column("approved_on", Date),
Column("paid_on", Date),
Column("payment_id", String),
Column("fail_reason", String),
Column("total_amount", MONEY),
Column("reimbursable_amount", MONEY),
Column("netted_amount", MONEY),
Column("waiting_since", Date),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
exp_line = Table("exp_line", md,
Column("report_id", String, primary_key=True),
Column("revision", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("kind", String),
Column("category_code", String),
Column("expense_date", Date),
Column("merchant", String),
Column("receipt_amount", MONEY),
Column("receipt_currency", String),
Column("amount", MONEY),
Column("paid_with", String),
Column("vat_amount", MONEY),
Column("cost_centre", String),
Column("receipt_key", String),
Column("description", String),
Column("city", String),
Column("nights", Integer),
Column("attendees_json", String),
Column("justification", String),
Column("policy_result_json", String),
Column("ai_fields_json", String),
Column("stale", Boolean),
Column("stale_reason", String),
Column("card_txn_id", String),
Column("recomputed_json", String),
Column("fx_rate", Numeric(19, 6)),
Column("treatment_code", String),
Column("contents_json", String),
Column("tax_code", String),
Column("vat_reclaimable", MONEY),
Column("vat_non_reclaimable", MONEY),
Column("tax_json", String),
Column("receipt_vat", MONEY),
Column("possible_duplicate_of", String),
Column("miles_or_km", Numeric(19, 4)),
Column("trip_from", TIMESTAMP),
Column("trip_to", TIMESTAMP),
Column("meals_provided_json", String),
Column("calc_json", String),
Column("rate_country", String),
Column("rate_variant", String),
Column("rebill_bp_id", BigInteger),
Column("rebill_markup_pct", Numeric(19, 4)),
Column("rebill_tax_category", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
exp_approval_step = Table("exp_approval_step", md,
Column("report_id", String, primary_key=True),
Column("revision", Integer, primary_key=True),
Column("level", Integer, primary_key=True),
Column("kind", String),
Column("approver_person_id", BigInteger),
Column("decision", String),
Column("decided_by", String),
Column("decided_sub", String),
Column("decided_at", TIMESTAMP),
Column("reason", String),
Column("delegate_for", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One guard row per bounded operation on money or on a person: the operation takes the NEXT n for its key in its own
# transaction, so two racers cannot both succeed and a loser is a 409. scopes: series, submit, approval, close, complete,
# paid, leaver.
exp_claim = Table("exp_claim", md,
Column("claim_scope", String, primary_key=True),
Column("claim_key", String, primary_key=True),
Column("n", Integer, primary_key=True),
Column("ref", String),
Column("claimed_by", String),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A side effect waiting for a neighbour, written in the SAME transaction as the state change that needs it and sent only
# after that commit. target: gl | payments.
exp_outbox = Table("exp_outbox", md,
Column("target", String, primary_key=True),
Column("ref", String, primary_key=True),
Column("report_id", String),
Column("payload_json", String),
Column("status", String),
Column("attempts", Integer),
Column("last_error", String),
Column("sent_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
exp_category = Table("exp_category", md,
Column("company_code", String, primary_key=True),
Column("category_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("name", String),
Column("gl_purpose", String),
Column("tax_category", String),
Column("receipt_required", Boolean),
Column("kind", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
exp_settings = Table("exp_settings", md,
Column("setting_key", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# hr told us a person left (POST /employees/{person_id}/leaver): one row per hr source_ref, and the answer given.
exp_leaver = Table("exp_leaver", md,
Column("source_ref", String, primary_key=True),
Column("person_id", BigInteger),
Column("last_day", Date),
Column("claims_until", Date),
Column("open_reports", Integer),
Column("cards_stopped", Integer),
Column("rerouted_json", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A limit or a ban on a category: dated rows finance edits (EXP-07, EXP-23). The row in force on a line's date is the latest valid_from on or
# before it; a company's own row beats the '*' default of the same rule_code.
exp_policy_rule = Table("exp_policy_rule", md,
Column("company_code", String, primary_key=True),
Column("rule_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("category_code", String),
Column("scope", String),
Column("city_or_country", String),
Column("limit_amount", MONEY),
Column("currency", String),
Column("mode", String),
Column("requires", String),
Column("message", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A manager's absence cover (EXP-09): the delegate may decide the manager level of the delegator's claims between two dates. Identity is kept
# as both names (email, sub): the same person under either login is the same person.
exp_delegation = Table("exp_delegation", md,
Column("delegator", String, primary_key=True),
Column("delegate", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("delegator_email", String),
Column("delegator_sub", String),
Column("delegate_email", String),
Column("delegate_sub", String),
Column("valid_to", Date),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A company card and who holds it (EXP-15, EXP-19). status: active | stopped (a leaver's cards stop the day after their last day).
exp_card = Table("exp_card", md,
Column("card_id", String, primary_key=True),
Column("company_code", String),
Column("holder_person_id", BigInteger),
Column("provider", String),
Column("currency", String),
Column("account_ref", String),
Column("status", String),
Column("stopped_from", Date),
Column("holder_email", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One transaction of a card feed, keyed card:<card_id>:<provider txn ref>. status: imported -> matched | personal -> recovered.
exp_card_txn = Table("exp_card_txn", md,
Column("card_txn_id", String, primary_key=True),
Column("card_id", String),
Column("company_code", String),
Column("holder_person_id", BigInteger),
Column("posted_on", Date),
Column("merchant", String),
Column("billed_amount", MONEY),
Column("billed_currency", String),
Column("original_amount", MONEY),
Column("original_currency", String),
Column("status", String),
Column("statement_ref", String),
Column("matched_report_id", String),
Column("matched_line_no", Integer),
Column("recovery_journal_id", BigInteger),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The row two operations on one card transaction race for: match, personal, recover (netting AND repayment share `recover`, so one wins).
# A plain INSERT; the loser is a 409.
exp_card_claim = Table("exp_card_claim", md,
Column("card_txn_id", String, primary_key=True),
Column("operation", String, primary_key=True),
Column("report_id", String),
Column("payout_ref", String),
Column("payment_id", String),
Column("claimed_by", String),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The photo or PDF of a receipt, kept as evidence for the retention period. object_key is raw/expenses/<report_id>/<line_no>/<sha256>.<ext>: one row
# per stored image (the same bytes are the same key). GAP: the spec's `embedding` (a VECTOR for near-duplicate search) is not here - no image
# embedding model is reachable from the platform yet; the same-digest duplicate check is.
exp_receipt_image = Table("exp_receipt_image", md,
Column("object_key", String, primary_key=True),
Column("report_id", String),
Column("line_no", Integer),
Column("sha256", String),
Column("content_type", String),
Column("size_bytes", Integer),
Column("uploaded_by", String),
Column("uploaded_at", TIMESTAMP),
Column("retain_until", Date),
Column("possible_duplicate_of", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A mileage or daily-allowance rate for a country, vehicle or destination, from a date (EXP-03, EXP-04, EXP-23). rate_amount is per mile / km or per
# 24 hours; a mileage band ends at band_upto cumulative miles in the tax year (NULL: no end); year_start is MM-DD.
exp_rate = Table("exp_rate", md,
Column("kind", String, primary_key=True),
Column("country", String, primary_key=True),
Column("variant", String, primary_key=True),
Column("band", Integer, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("rate_amount", Numeric(19, 4)),
Column("currency", String),
Column("band_upto", Numeric(19, 4)),
Column("unit", String),
Column("partial_day_amount", Numeric(19, 4)),
Column("breakfast_pct", Numeric(19, 4)),
Column("lunch_pct", Numeric(19, 4)),
Column("dinner_pct", Numeric(19, 4)),
Column("year_start", String),
Column("source", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# An employee's yes or no to an AI step reading their receipts: append-only, the latest row decides. GDPR arts. 6 and 13: what is processed, where,
# and which text they saw (text_version). Nobody consents for someone else: the row is written for the caller's own person.
exp_consent = Table("exp_consent", md,
Column("person_id", BigInteger, primary_key=True),
Column("step_key", String, primary_key=True),
Column("given_at", TIMESTAMP, primary_key=True),
Column("decision", String),
Column("text_version", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# What the model read from a receipt and what the person did with each field (the audit of the read_receipt step).
exp_ai_finding = Table("exp_ai_finding", md,
Column("report_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("finding_no", Integer, primary_key=True),
Column("step_key", String),
Column("model", String),
Column("inputs_digest", String),
Column("fields_json", String),
Column("confirmed_json", String),
Column("decided_by", String),
Column("decided_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One Odoo migration run (EXP-24): loaded -> reconciled -> signed | discarded. checks_json holds the reconciliation (each check with Odoo's number, ours and the
# owner's explanation); a SIGNED run is frozen.
exp_migration_run = Table("exp_migration_run", md,
Column("run_id", String, primary_key=True),
Column("odoo_db", String),
Column("cutover_date", Date),
Column("company_code", String),
Column("status", String),
Column("doors_json", String),
Column("checks_json", String),
Column("summary_json", String),
Column("loaded_by", String),
Column("signed_by", String),
Column("signed_sub", String),
Column("signed_on", Date),
Column("discarded_by", String),
Column("discard_reason", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# (Odoo db, Odoo model, Odoo id) -> our natural key (shared.yaml conventions.odoo_migration): what a later brick resolves through, and how a re-load matches.
exp_migration_map = Table("exp_migration_map", md,
Column("odoo_db", String, primary_key=True),
Column("odoo_model", String, primary_key=True),
Column("odoo_id", String, primary_key=True),
Column("ours", String),
Column("run_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
class AlreadyDone(RuleBroken):
"""The one-shot claim was already taken: a 409 with these words."""
def _utc(dt: Optional[datetime]) -> Optional[datetime]:
"""Stored timestamps are naive UTC."""
if dt is None:
return None
return dt.astimezone(timezone.utc).replace(tzinfo=None) if dt.tzinfo else dt
def _plain(conn, table: Table, row: dict, at: datetime) -> None:
"""A FACT row: a plain INSERT - a second writer of the same key is rejected (23505), never overwritten."""
conn.execute(insert(table).values(**row, created_at=_utc(at), updated_at=_utc(at)))
def _all(conn, table: Table, cond=None, order=()) -> list:
q = select(table)
if cond is not None:
q = q.where(cond)
return [dict(m) for m in conn.execute(q.order_by(*order)).mappings()]
def _dec(v) -> Optional[Decimal]:
return None if v is None else money(v)
def _no_created(row: dict) -> dict:
"""An update keeps the row's created_at: the keyed upsert writes every column it is given."""
return {k: v for k, v in row.items() if k != "created_at"}
def line_row(report_id: str, revision: int, ln: rp.Line, at: datetime) -> dict:
return {"report_id": report_id, "revision": revision, "line_no": ln.line_no, "kind": ln.kind, "category_code": ln.category_code,
"expense_date": ln.expense_date, "merchant": ln.merchant, "receipt_amount": ln.receipt_amount,
"receipt_currency": ln.receipt_currency or None, "amount": ln.amount, "paid_with": ln.paid_with,
"vat_amount": ln.vat_amount, "cost_centre": ln.cost_centre or None, "receipt_key": ln.receipt_key or None,
"description": ln.description or None, "city": ln.city or None, "nights": ln.nights,
"attendees_json": json.dumps(ln.attendees) if ln.attendees else None, "justification": ln.justification or None,
"policy_result_json": json.dumps(ln.policy_results, sort_keys=True) if ln.policy_results else None,
"ai_fields_json": json.dumps({"unconfirmed": ln.unconfirmed, **ln.ai}, sort_keys=True) if (ln.unconfirmed or ln.ai) else None,
"stale": bool(ln.stale), "stale_reason": ln.stale_reason or None, "card_txn_id": ln.card_txn_id or None,
"recomputed_json": json.dumps(ln.recomputed, sort_keys=True) if ln.recomputed else None, "fx_rate": ln.fx_rate,
"treatment_code": ln.treatment_code or None, "contents_json": json.dumps(ln.contents, sort_keys=True) if ln.contents else None,
"tax_code": ln.tax_code or None, "vat_reclaimable": ln.vat_reclaimable, "vat_non_reclaimable": ln.vat_non_reclaimable,
"tax_json": json.dumps(ln.tax, sort_keys=True, default=str) if ln.tax else None, "receipt_vat": ln.receipt_vat,
"possible_duplicate_of": ln.possible_duplicate_of or None,
"miles_or_km": ln.miles, "trip_from": ln.trip_from, "trip_to": ln.trip_to,
"meals_provided_json": json.dumps(ln.meals) if ln.meals else None, "calc_json": json.dumps(ln.calc, sort_keys=True, default=str) if ln.calc else None,
"rate_country": ln.rate_country or None, "rate_variant": ln.rate_variant or None, "rebill_bp_id": ln.rebill_bp_id,
"rebill_markup_pct": ln.rebill_markup_pct if ln.rebill_bp_id else None, "rebill_tax_category": ln.rebill_tax_category or None, "created_at": _utc(at), "updated_at": _utc(at)}
def line_from_row(r: dict) -> rp.Line:
ai = json.loads(r["ai_fields_json"]) if r["ai_fields_json"] else {}
return rp.Line(
line_no=r["line_no"], category_code=r["category_code"], expense_date=r["expense_date"], merchant=r["merchant"] or "", amount=money(r["amount"]),
vat_amount=money(r["vat_amount"] or 0), paid_with=r["paid_with"], kind=r["kind"], cost_centre=r["cost_centre"] or "", receipt_key=r["receipt_key"] or "",
description=r["description"] or "", receipt_currency=r["receipt_currency"] or "", receipt_amount=_dec(r["receipt_amount"]), city=r["city"] or "",
nights=r["nights"], attendees=json.loads(r["attendees_json"]) if r["attendees_json"] else [], justification=r["justification"] or "",
policy_results=json.loads(r["policy_result_json"]) if r["policy_result_json"] else [], unconfirmed=ai.get("unconfirmed", []),
ai={k: v for k, v in ai.items() if k != "unconfirmed"},
stale=bool(r["stale"]), stale_reason=r["stale_reason"] or "", recomputed=json.loads(r["recomputed_json"]) if r["recomputed_json"] else [],
card_txn_id=r["card_txn_id"] or "", fx_rate=Decimal(str(r["fx_rate"])) if r["fx_rate"] is not None else None, treatment_code=r["treatment_code"] or "",
contents=json.loads(r["contents_json"]) if r["contents_json"] else {}, tax_code=r["tax_code"] or "", vat_reclaimable=money(r["vat_reclaimable"] or 0),
vat_non_reclaimable=money(r["vat_non_reclaimable"] or 0), tax=json.loads(r["tax_json"]) if r["tax_json"] else {},
miles=Decimal(str(r["miles_or_km"])) if r["miles_or_km"] is not None else None, trip_from=r["trip_from"], trip_to=r["trip_to"],
meals=json.loads(r["meals_provided_json"]) if r["meals_provided_json"] else [], calc=json.loads(r["calc_json"]) if r["calc_json"] else {},
rate_country=r["rate_country"] or "", rate_variant=r["rate_variant"] or "", possible_duplicate_of=r["possible_duplicate_of"] or "",
rebill_bp_id=r["rebill_bp_id"], rebill_markup_pct=Decimal(str(r["rebill_markup_pct"])) if r["rebill_markup_pct"] is not None else money(0), rebill_tax_category=r["rebill_tax_category"] or "",
receipt_vat=_dec(r["receipt_vat"]))
ROOT_COLS = tuple(c.name for c in exp_report.columns if c.name not in ("created_at", "updated_at"))
class ReportRepository:
"""A report whole: the root, the CURRENT revision's lines and its decisions - three queries however many lines."""
def __init__(self, conn):
self.conn = conn
def get(self, report_id: str) -> Optional[rp.Report]:
root = self.conn.execute(select(exp_report).where(exp_report.c.report_id == report_id)).mappings().first()
if root is None:
return None
rev = root["revision"]
lines = _all(self.conn, exp_line, (exp_line.c.report_id == report_id) & (exp_line.c.revision == rev), (exp_line.c.line_no,))
steps = _all(self.conn, exp_approval_step, (exp_approval_step.c.report_id == report_id) & (exp_approval_step.c.revision == rev),
(exp_approval_step.c.level,))
return self._make(dict(root), lines, steps)
@staticmethod
def _make(root: dict, lines: list, steps: list) -> rp.Report:
return rp.Report(
report_id=root["report_id"], person_id=root["person_id"], company_code=root["company_code"], bp_id=root["bp_id"],
claimant=rp.Identity(root["claimant_email"] or "", root["claimant_sub"] or ""), currency=root["currency"],
title=root["title"] or "", revision=root["revision"], status=root["status"], submitted_on=root["submitted_on"],
submitted_at=root["submitted_at"], chain=rp.chain_from_json(root["chain_json"]), waiting_level=root["waiting_level"],
journal_id=root["journal_id"], payout_ref=root["payout_ref"], approved_on=root["approved_on"], paid_on=root["paid_on"],
payment_id=root["payment_id"], fail_reason=root["fail_reason"], total_amount=_dec(root["total_amount"]),
reimbursable_amount=_dec(root["reimbursable_amount"]), netted_amount=_dec(root["netted_amount"]) or money(0), waiting_since=root["waiting_since"],
lines=[line_from_row(r) for r in lines],
steps=[rp.Step(r["level"], r["kind"], r["decision"], r["decided_by"] or "", r["decided_sub"] or "", r["decided_at"], r["reason"] or "",
r["approver_person_id"], r["delegate_for"] or "") for r in steps])
def _root_row(self, r: rp.Report) -> dict:
waiting = r.level_row(r.waiting_level) if r.waiting_level else None
return {"report_id": r.report_id, "person_id": r.person_id, "company_code": r.company_code, "bp_id": r.bp_id,
"claimant_email": r.claimant.email, "claimant_sub": r.claimant.sub, "revision": r.revision, "title": r.title,
"currency": r.currency, "status": r.status, "submitted_at": _utc(r.submitted_at), "submitted_by": r.claimant.email,
"submitted_sub": r.claimant.sub if r.submitted_on else None, "submitted_on": r.submitted_on,
"chain_json": rp.chain_json(r.chain) if r.chain else None, "waiting_level": r.waiting_level,
"waiting_kind": waiting.kind if waiting else None, "waiting_email": (waiting.email or None) if waiting else None,
"waiting_sub": (waiting.sub or None) if waiting else None, "journal_id": r.journal_id, "payout_ref": r.payout_ref,
"approved_on": r.approved_on, "paid_on": r.paid_on, "payment_id": r.payment_id, "fail_reason": r.fail_reason,
"total_amount": r.total_amount, "reimbursable_amount": r.reimbursable_amount,
"netted_amount": r.netted_amount if r.netted_amount else None, "waiting_since": r.waiting_since}
def create(self, r: rp.Report, at: datetime) -> None:
_plain(self.conn, exp_report, self._root_row(r), at)
def save(self, r: rp.Report, at: datetime) -> None:
"""The root moves forward (status, chain, totals): a keyed upsert of the whole row."""
upsert(self.conn, exp_report, [{**self._root_row(r), "updated_at": _utc(at)}], ("report_id",))
def add_lines(self, report_id: str, revision: int, lines: list, at: datetime) -> None:
"""New lines in ONE statement, however many."""
if not lines:
return
self.conn.execute(insert(exp_line), [line_row(report_id, revision, ln, at) for ln in lines])
def save_lines(self, report_id: str, revision: int, lines: list, at: datetime) -> None:
"""Lines that already exist, written whole (a keyed upsert) in ONE statement however many: a confirmed field, a recomputed amount,
a stale mark. Only a DRAFT's lines, or a submitted line's derived marks, are ever written this way - never a posted line."""
if lines:
upsert(self.conn, exp_line, [_no_created(line_row(report_id, revision, ln, at)) for ln in lines],
("report_id", "revision", "line_no"))
def remove_line(self, report_id: str, revision: int, line_no: int) -> None:
self.conn.execute(delete(exp_line).where((exp_line.c.report_id == report_id) & (exp_line.c.revision == revision)
& (exp_line.c.line_no == line_no)))
def add_step(self, report_id: str, revision: int, s: rp.Step, at: datetime) -> None:
_plain(self.conn, exp_approval_step, {"report_id": report_id, "revision": revision, "level": s.level, "kind": s.kind,
"approver_person_id": s.approver_person_id, "decision": s.decision, "decided_by": s.decided_by,
"decided_sub": s.decided_sub, "decided_at": _utc(s.decided_at), "reason": s.reason or None,
"delegate_for": s.delegate_for or None}, at)
# ── lists: one query each ────────────────────────────────────────────────────────────────
def summaries(self, *, mine: Optional[rp.Identity] = None, waiting_for: Optional[rp.Identity] = None, finance_queue: bool = False,
everything: bool = False, person_id: Optional[int] = None, limit: int = 200) -> list:
"""The caller's claims, the claims waiting for them, or (everything) every claim: root rows only."""
conds = []
if everything:
conds.append(exp_report.c.report_id.isnot(None))
if mine is not None:
own = [exp_report.c.claimant_sub == mine.sub] if mine.sub else []
own += [exp_report.c.claimant_email == mine.email.lower()] if mine.email else []
conds.append(or_(*own) if own else exp_report.c.report_id.is_(None))
if waiting_for is not None: # one person, or a list: the caller and whoever they cover for
w = []
for who in (waiting_for if isinstance(waiting_for, (list, tuple)) else [waiting_for]):
w += [exp_report.c.waiting_sub == who.sub] if who.sub else []
w += [exp_report.c.waiting_email == who.email.lower()] if who.email else []
conds.append((exp_report.c.status == "submitted") & (or_(*w) if w else exp_report.c.report_id.is_(None)))
if finance_queue:
conds.append((exp_report.c.status == "submitted") & (exp_report.c.waiting_kind == "finance"))
if person_id is not None:
conds.append(exp_report.c.person_id == person_id)
if not conds:
return []
rows = self.conn.execute(select(exp_report).where(or_(*conds)).order_by(exp_report.c.report_id).limit(limit)).mappings()
return [dict(r) for r in rows]
def open_of_person(self, person_id: int) -> list:
"""The person's claims that are not finished: draft or submitted."""
rows = self.conn.execute(select(exp_report).where((exp_report.c.person_id == person_id)
& exp_report.c.status.in_(("draft", "submitted")))
.order_by(exp_report.c.report_id)).mappings()
return [dict(r) for r in rows]
def waiting_all(self, limit: int = 5000) -> list:
"""Every claim waiting for a decision (root rows): the daily pass reads them in ONE query."""
rows = self.conn.execute(select(exp_report).where((exp_report.c.status == "submitted") & exp_report.c.waiting_level.isnot(None))
.order_by(exp_report.c.report_id).limit(limit)).mappings()
return [dict(r) for r in rows]
def stuck_completions(self) -> list:
"""Submitted claims whose every level is decided but that were never completed (a crash between two transactions)."""
return [r["report_id"] for r in self.conn.execute(select(exp_report.c.report_id).where(
(exp_report.c.status == "submitted") & exp_report.c.waiting_level.is_(None) & exp_report.c.chain_json.isnot(None))).mappings()]
def open_lines(self, *, company_code: Optional[str] = None, category_codes: Optional[list] = None, kinds: Optional[list] = None,
person_id: Optional[int] = None, statuses: tuple = ("draft", "submitted"), exclude_report: Optional[str] = None) -> list:
"""[(report_id, revision, Line)] of the CURRENT revision of every not-yet-approved claim (draft or submitted) - what a changed rule can
make stale. ONE query, joined to the report root; never an approved or posted line."""
cond = exp_report.c.status.in_(statuses) & (exp_line.c.revision == exp_report.c.revision)
if company_code and company_code != "*":
cond &= exp_report.c.company_code == company_code
if category_codes:
cond &= exp_line.c.category_code.in_(category_codes)
if kinds:
cond &= exp_line.c.kind.in_(kinds)
if person_id is not None:
cond &= exp_report.c.person_id == person_id
if exclude_report:
cond &= exp_report.c.report_id != exclude_report
q = select(exp_line).select_from(exp_line.join(exp_report, exp_line.c.report_id == exp_report.c.report_id)).where(cond).order_by(
exp_line.c.report_id, exp_line.c.line_no)
return [(r["report_id"], r["revision"], line_from_row(dict(r))) for r in self.conn.execute(q).mappings()]
def approved_miles(self, person_id: int, since: date, until: date, country: str, variant: str, exclude_report: Optional[str] = None) -> Decimal:
"""The miles (or km) of this person's mileage lines of one country and variant in [since, until] on claims APPROVED, posted or paid: what the band
a new line starts in is derived from. ONE SUM; an unapproved line never counts, so two claims approved in either order end at the same total."""
cond = (exp_report.c.person_id == person_id) & exp_report.c.status.in_(("approved", "posted", "paid")) & (exp_line.c.revision == exp_report.c.revision) \
& (exp_line.c.kind == "mileage") & (exp_line.c.rate_country == country) & (exp_line.c.rate_variant == variant) \
& (exp_line.c.expense_date >= since) & (exp_line.c.expense_date <= until)
if exclude_report:
cond &= exp_line.c.report_id != exclude_report
got = self.conn.execute(select(func.coalesce(func.sum(exp_line.c.miles_or_km), 0)).select_from(
exp_line.join(exp_report, exp_line.c.report_id == exp_report.c.report_id)).where(cond)).scalar()
return Decimal(str(got or 0))
def save_marked(self, marked: list, at: datetime) -> None:
"""Lines of several claims, written whole in ONE statement: [(report_id, revision, Line)]."""
if marked:
upsert(self.conn, exp_line, [_no_created(line_row(rid, rev, ln, at)) for rid, rev, ln in marked],
("report_id", "revision", "line_no"))
def lines_of(self, report_id: str, revision: int) -> list:
return _all(self.conn, exp_line, (exp_line.c.report_id == report_id) & (exp_line.c.revision == revision), (exp_line.c.line_no,))
def revisions(self, report_id: str) -> list:
"""[(revision, last decision)] of every revision that has a decision, oldest first: how a returned claim reads."""
rows = self.conn.execute(select(exp_approval_step.c.revision, exp_approval_step.c.level, exp_approval_step.c.decision)
.where(exp_approval_step.c.report_id == report_id)
.order_by(exp_approval_step.c.revision, exp_approval_step.c.level)).mappings()
out: dict = {}
for r in rows:
out[r["revision"]] = r["decision"]
return sorted(out.items())
class ClaimRepository:
"""The one guard key per bounded thing: every operation that touches it takes the next claim in ITS transaction."""
def __init__(self, conn):
self.conn = conn
def count(self, scope: str, key: str) -> int:
return self.conn.execute(select(func.count()).select_from(exp_claim).where(
(exp_claim.c.claim_scope == scope) & (exp_claim.c.claim_key == key))).scalar() or 0
def take(self, scope: str, key: str, at: datetime, by: str = "", ref: str = "") -> int:
"""The next n for a key that counts (a report number): n = committed claims + 1; a racer with the same n loses at the INSERT."""
n = self.count(scope, key) + 1
_plain(self.conn, exp_claim, {"claim_scope": scope, "claim_key": key, "n": n, "ref": ref, "claimed_by": by, "claimed_at": _utc(at)}, at)
return n
def once(self, scope: str, key: str, at: datetime, by: str = "", ref: str = "") -> None:
"""A key that can be claimed exactly ONCE (one decision of one level, one payout paid): already claimed -> the operation was done a moment
ago; two racers both see none, both insert n=1, and the second is rejected at the INSERT (23505)."""
if self.count(scope, key):
raise AlreadyDone(f"Someone else did the same a moment ago ({scope} {key}) - load the claim again")
_plain(self.conn, exp_claim, {"claim_scope": scope, "claim_key": key, "n": 1, "ref": ref, "claimed_by": by, "claimed_at": _utc(at)}, at)
class OutboxRepository:
def __init__(self, conn):
self.conn = conn
def add(self, target: str, ref: str, report_id: str, payload: dict, at: datetime) -> None:
_plain(self.conn, exp_outbox, {"target": target, "ref": ref, "report_id": report_id,
"payload_json": json.dumps(payload, sort_keys=True), "status": "pending", "attempts": 0}, at)
def get(self, target: str, ref: str) -> Optional[dict]:
row = self.conn.execute(select(exp_outbox).where((exp_outbox.c.target == target) & (exp_outbox.c.ref == ref))).mappings().first()
return dict(row) if row else None
def many(self, target: str, refs: list) -> dict:
"""ref -> row for a set of refs: one query per 500."""
got: dict = {}
refs = sorted(set(refs))
for i in range(0, len(refs), 500):
for r in _all(self.conn, exp_outbox, (exp_outbox.c.target == target) & exp_outbox.c.ref.in_(refs[i:i + 500])):
got[r["ref"]] = r
return got
def listed(self, target: str, status: Optional[str] = None, limit: int = 200) -> list:
cond = exp_outbox.c.target == target
if status:
cond &= exp_outbox.c.status == status
return [dict(r) for r in self.conn.execute(select(exp_outbox).where(cond).order_by(exp_outbox.c.created_at.desc(), exp_outbox.c.ref).limit(limit)).mappings()]
def pending(self, report_id: Optional[str] = None) -> list:
q = select(exp_outbox).where(exp_outbox.c.status == "pending")
if report_id:
q = q.where(exp_outbox.c.report_id == report_id)
return [dict(r) for r in self.conn.execute(q.order_by(exp_outbox.c.created_at, exp_outbox.c.target, exp_outbox.c.ref)).mappings()]
def of_report(self, report_id: str) -> list:
return _all(self.conn, exp_outbox, exp_outbox.c.report_id == report_id, (exp_outbox.c.target, exp_outbox.c.ref))
def reset(self, row: dict, payload: dict, at: datetime) -> None:
"""A row ar (or any neighbour) refused, asked again with a corrected payload: the same key, pending, a fresh count."""
upsert(self.conn, exp_outbox, [{"target": row["target"], "ref": row["ref"], "report_id": row["report_id"],
"payload_json": json.dumps(payload, sort_keys=True), "status": "pending", "attempts": 0, "last_error": None,
"sent_at": None, "updated_at": _utc(at)}], ("target", "ref"))
def mark(self, row: dict, status: str, error: str, at: datetime, payload: Optional[dict] = None) -> None:
keep = {k: row[k] for k in ("target", "ref", "report_id", "payload_json")}
if payload is not None:
keep["payload_json"] = json.dumps(payload, sort_keys=True)
upsert(self.conn, exp_outbox, [{**keep, "status": status, "attempts": (row["attempts"] or 0) + 1, "last_error": error or None,
"sent_at": _utc(at) if status == "sent" else None, "updated_at": _utc(at)}], ("target", "ref"))
class CategoryRepository:
def __init__(self, conn):
self.conn = conn
_memo: Optional[dict] = None
def in_force(self, company_code: str, pairs: set) -> dict:
"""(code, date) -> the category row in force on that date: the company's own rows over the '*' default, then the latest
valid_from on or before the date. ONE query however many lines and dates - and one per unit of work for the same question
(the gates and the tax answer ask it twice)."""
if not pairs:
return {}
key = (company_code, frozenset(pairs))
if self._memo is None:
self._memo = {}
if key not in self._memo:
self._memo[key] = self._in_force(company_code, pairs)
return self._memo[key]
def _in_force(self, company_code: str, pairs: set) -> dict:
codes = sorted({c for c, _ in pairs})
rows = _all(self.conn, exp_category, exp_category.c.company_code.in_((company_code, "*")) & exp_category.c.category_code.in_(codes)
& (exp_category.c.valid_from <= max(d for _, d in pairs)), (exp_category.c.valid_from,))
out: dict = {}
for code, on in pairs:
mine = [r for r in rows if r["category_code"] == code and r["valid_from"] <= on]
own = [r for r in mine if r["company_code"] != "*"]
pick = (own or mine)[-1:]
if pick:
out[(code, on)] = pick[0]
return out
def of_key(self, company_code: str, category_code: str) -> list:
"""Every row of one company's code (not the '*' default's): what a dated change is checked against."""
return _all(self.conn, exp_category, (exp_category.c.company_code == company_code) & (exp_category.c.category_code == category_code),
(exp_category.c.valid_from,))
def all_in_force(self, company_code: str, on: date) -> list:
"""Every category in force on a date for a company: the company's own row of a code over the '*' default, the latest valid_from. ONE query."""
rows = _all(self.conn, exp_category, exp_category.c.company_code.in_((company_code, "*")) & (exp_category.c.valid_from <= on),
(exp_category.c.category_code, exp_category.c.valid_from))
best: dict = {}
for r in rows:
mine = r["company_code"] != "*"
if r["category_code"] not in best or (mine, r["valid_from"]) > (best[r["category_code"]]["company_code"] != "*", best[r["category_code"]]["valid_from"]):
best[r["category_code"]] = r
return [best[c] for c in sorted(best)]
def put(self, company_code: str, code: str, name: str, gl_purpose: str, tax_category: str, valid_from: date, at: datetime,
receipt_required: bool = True, kind: str = "receipt") -> None:
self._memo = None
_plain(self.conn, exp_category, {"company_code": company_code, "category_code": code, "valid_from": valid_from, "name": name,
"gl_purpose": gl_purpose, "tax_category": tax_category, "receipt_required": receipt_required,
"kind": kind}, at)
class SettingsRepository:
"""Dated settings. The business date is counted in `business_timezone` - read HERE and nowhere else."""
def __init__(self, conn):
self.conn = conn
self._rows: Optional[list] = None
def rows(self) -> list:
if self._rows is None:
self._rows = _all(self.conn, exp_settings, None, (exp_settings.c.setting_key, exp_settings.c.valid_from))
return self._rows
def on(self, key: str, day: date) -> str:
"""The value in force on `day`: the latest row dated on or before it, else the default."""
value = SETTING_DEFS[key][0]
for r in self.rows():
if r["setting_key"] == key and r["valid_from"] <= day:
value = r["value"]
return value
def timezone(self) -> str:
"""A zone row takes over once the business date, counted in the zone BEFORE it, has reached its valid_from - never by the
UTC date, which could put a row dated tomorrow in force at once."""
name = SETTING_DEFS["business_timezone"][0]
now = work.now()
for r in self.rows():
if r["setting_key"] == "business_timezone" and r["valid_from"] <= now.astimezone(ZoneInfo(name)).date():
name = r["value"] or name
return name
def put(self, key: str, value: str, valid_from: date, by: str, at: datetime) -> None:
_plain(self.conn, exp_settings, {"setting_key": key, "valid_from": valid_from, "value": str(value), "set_by": by}, at)
self._rows = None
class LeaverRepository:
def __init__(self, conn):
self.conn = conn
def get(self, source_ref: str) -> Optional[dict]:
row = self.conn.execute(select(exp_leaver).where(exp_leaver.c.source_ref == source_ref)).mappings().first()
return dict(row) if row else None
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, exp_leaver, row, at)
def claims_until(self, person_id: int) -> Optional[date]:
"""The latest day this person may still claim, over everything hr has told us about them."""
return self.conn.execute(select(func.max(exp_leaver.c.claims_until)).where(exp_leaver.c.person_id == person_id)).scalar()
class PolicyRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _rule(r: dict) -> pol.Rule:
return pol.Rule(r["company_code"], r["rule_code"], r["valid_from"], r["category_code"], r["scope"], r["city_or_country"] or "",
money(r["limit_amount"] or 0), r["currency"] or "", r["mode"], r["requires"] or "", r["message"] or "")
def rows(self, company_code: str, up_to: date) -> list:
"""Every rule row of a company (and the '*' defaults) dated on or before `up_to`: ONE query. policy.in_force picks the row per date."""
rows = _all(self.conn, exp_policy_rule, exp_policy_rule.c.company_code.in_((company_code, "*")) & (exp_policy_rule.c.valid_from <= up_to),
(exp_policy_rule.c.rule_code, exp_policy_rule.c.valid_from))
return [self._rule(r) for r in rows]
def of_key(self, company_code: str, rule_code: str) -> list:
rows = _all(self.conn, exp_policy_rule, (exp_policy_rule.c.company_code == company_code) & (exp_policy_rule.c.rule_code == rule_code),
(exp_policy_rule.c.valid_from,))
return [self._rule(r) for r in rows]
def put(self, r: pol.Rule, by: str, at: datetime) -> None:
_plain(self.conn, exp_policy_rule, {"company_code": r.company_code, "rule_code": r.rule_code, "valid_from": r.valid_from,
"category_code": r.category_code, "scope": r.scope, "city_or_country": r.city_or_country or None,
"limit_amount": r.limit_amount, "currency": r.currency or None, "mode": r.mode,
"requires": r.requires or None, "message": r.message or None, "set_by": by}, at)
class DelegationRepository:
def __init__(self, conn):
self.conn = conn
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, exp_delegation, row, at)
def active_for(self, delegate: rp.Identity, on: date) -> list:
"""The covers this person holds on a date: every delegation naming their email OR their sub, in force. ONE query."""
names = [exp_delegation.c.delegate_sub == delegate.sub] if delegate.sub else []
names += [exp_delegation.c.delegate_email == delegate.email.lower()] if delegate.email else []
if not names:
return []
return _all(self.conn, exp_delegation, or_(*names) & (exp_delegation.c.valid_from <= on)
& (exp_delegation.c.valid_to.is_(None) | (exp_delegation.c.valid_to >= on)), (exp_delegation.c.valid_from,))
def end(self, row: dict, at: datetime) -> None:
upsert(self.conn, exp_delegation, [{**{k: v for k, v in row.items() if k != "created_at"}, "updated_at": _utc(at)}], ("delegator", "delegate", "valid_from"))
def of_delegator(self, delegator: rp.Identity) -> list:
names = [exp_delegation.c.delegator_sub == delegator.sub] if delegator.sub else []
names += [exp_delegation.c.delegator_email == delegator.email.lower()] if delegator.email else []
return _all(self.conn, exp_delegation, or_(*names), (exp_delegation.c.valid_from,)) if names else []
class CardRepository:
"""Cards, their transactions and the claim rows. Every list is ONE query; a card's or a person's list holds only their own rows."""
def __init__(self, conn):
self.conn = conn
def get_card(self, card_id: str) -> Optional[dict]:
row = self.conn.execute(select(exp_card).where(exp_card.c.card_id == card_id)).mappings().first()
return dict(row) if row else None
def cards(self, ids: list) -> dict:
"""card_id -> row for a set of ids: ONE query."""
if not ids:
return {}
return {r["card_id"]: r for r in _all(self.conn, exp_card, exp_card.c.card_id.in_(sorted(set(ids))))}
def cards_of(self, person_id: int) -> list:
return _all(self.conn, exp_card, exp_card.c.holder_person_id == person_id, (exp_card.c.card_id,))
def put_card(self, row: dict, at: datetime) -> None:
_plain(self.conn, exp_card, row, at)
def save_card(self, row: dict, at: datetime) -> None:
upsert(self.conn, exp_card, [{**row, "updated_at": _utc(at)}], ("card_id",))
def stop_cards(self, person_id: int, from_day: date, at: datetime) -> int:
"""Stop every active card a person holds (the day after their last day): one read, one write however many cards."""
live = [r for r in self.cards_of(person_id) if r["status"] == "active"]
if live:
upsert(self.conn, exp_card, [{**{k: v for k, v in r.items() if k not in ("created_at",)}, "status": "stopped", "stopped_from": from_day,
"updated_at": _utc(at)} for r in live], ("card_id",))
return len(live)
def txn(self, card_txn_id: str) -> Optional[dict]:
row = self.conn.execute(select(exp_card_txn).where(exp_card_txn.c.card_txn_id == card_txn_id)).mappings().first()
return dict(row) if row else None
def txns(self, ids: list) -> dict:
have: dict = {}
ids = sorted(set(ids))
for i in range(0, len(ids), 500):
for r in _all(self.conn, exp_card_txn, exp_card_txn.c.card_txn_id.in_(ids[i:i + 500])):
have[r["card_txn_id"]] = r
return have
def add_txns(self, rows: list, at: datetime) -> None:
"""A whole feed in ONE statement per 500 rows."""
for i in range(0, len(rows), 500):
self.conn.execute(insert(exp_card_txn), [{**r, "created_at": _utc(at), "updated_at": _utc(at)} for r in rows[i:i + 500]])
def save_txn(self, row: dict, at: datetime) -> None:
upsert(self.conn, exp_card_txn, [{**{k: v for k, v in row.items() if k != "created_at"}, "updated_at": _utc(at)}], ("card_txn_id",))
def of_holder(self, person_id: int, *, status: Optional[str] = None, since: Optional[date] = None, until: Optional[date] = None,
card_id: Optional[str] = None, limit: int = 500) -> list:
"""A person's card transactions (optionally of one status, one card, a date range): ONE query, never anyone else's."""
cond = exp_card_txn.c.holder_person_id == person_id
if status:
cond &= exp_card_txn.c.status == status
if since:
cond &= exp_card_txn.c.posted_on >= since
if until:
cond &= exp_card_txn.c.posted_on <= until
if card_id:
cond &= exp_card_txn.c.card_id == card_id
return [dict(r) for r in self.conn.execute(select(exp_card_txn).where(cond).order_by(exp_card_txn.c.posted_on, exp_card_txn.c.card_txn_id)
.limit(limit)).mappings()]
def listed(self, *, status: Optional[str] = None, card_id: Optional[str] = None, company_code: Optional[str] = None,
posted_before: Optional[date] = None, limit: int = 500) -> list:
"""Finance's list across holders: ONE query, filtered by status / card / company / posted-before."""
cond = exp_card_txn.c.card_txn_id.isnot(None)
if status:
cond &= exp_card_txn.c.status == status
if card_id:
cond &= exp_card_txn.c.card_id == card_id
if company_code:
cond &= exp_card_txn.c.company_code == company_code
if posted_before:
cond &= exp_card_txn.c.posted_on <= posted_before
return [dict(r) for r in self.conn.execute(select(exp_card_txn).where(cond).order_by(exp_card_txn.c.posted_on, exp_card_txn.c.card_txn_id)
.limit(limit)).mappings()]
def claim(self, card_txn_id: str, operation: str, at: datetime, by: str = "", report_id: str = "", payout_ref: str = "",
payment_id: str = "") -> None:
"""Take (card_txn_id, operation): a plain INSERT, so exactly one of two racers wins; the other is told it was done a moment ago."""
have = self.conn.execute(select(func.count()).select_from(exp_card_claim).where(
(exp_card_claim.c.card_txn_id == card_txn_id) & (exp_card_claim.c.operation == operation))).scalar()
if have:
raise AlreadyDone(f"Someone else did the same a moment ago ({operation} of {card_txn_id}) - load it again")
_plain(self.conn, exp_card_claim, {"card_txn_id": card_txn_id, "operation": operation, "report_id": report_id or None,
"payout_ref": payout_ref or None, "payment_id": payment_id or None, "claimed_by": by,
"claimed_at": _utc(at)}, at)
def claims_of_report(self, report_id: str, operation: str) -> list:
return _all(self.conn, exp_card_claim, (exp_card_claim.c.report_id == report_id) & (exp_card_claim.c.operation == operation),
(exp_card_claim.c.card_txn_id,))
class ReceiptRepository:
def __init__(self, conn):
self.conn = conn
def get(self, object_key: str) -> Optional[dict]:
row = self.conn.execute(select(exp_receipt_image).where(exp_receipt_image.c.object_key == object_key)).mappings().first()
return dict(row) if row else None
def of_sha(self, digest: str) -> list:
"""Every stored image with these bytes (any claim, any person): ONE query on the digest."""
return _all(self.conn, exp_receipt_image, exp_receipt_image.c.sha256 == digest, (exp_receipt_image.c.uploaded_at, exp_receipt_image.c.object_key))
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, exp_receipt_image, row, at)
class RateRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _rate(r: dict) -> rt.Rate:
d = lambda v: Decimal(str(v)) if v is not None else None # noqa: E731
return rt.Rate(r["kind"], r["country"], r["variant"], r["band"], r["valid_from"], d(r["rate_amount"]), r["currency"], d(r["band_upto"]), r["unit"] or "",
d(r["partial_day_amount"]), d(r["breakfast_pct"]) or Decimal(0), d(r["lunch_pct"]) or Decimal(0), d(r["dinner_pct"]) or Decimal(0),
r["year_start"] or "01-01", r["source"] or "")
def rows(self, kind: str, country: str, variant: str, up_to: date) -> list:
"""The rate rows of one (kind, country, variant) dated on or before `up_to`: ONE query. rates.in_force picks the band rows per date."""
return [self._rate(r) for r in _all(self.conn, exp_rate, (exp_rate.c.kind == kind) & (exp_rate.c.country == country) & (exp_rate.c.variant == variant)
& (exp_rate.c.valid_from <= up_to), (exp_rate.c.band, exp_rate.c.valid_from))]
def of_key(self, kind: str, country: str, variant: str, band: int) -> list:
return [self._rate(r) for r in _all(self.conn, exp_rate, (exp_rate.c.kind == kind) & (exp_rate.c.country == country) & (exp_rate.c.variant == variant)
& (exp_rate.c.band == band), (exp_rate.c.valid_from,))]
def listed(self, kind: Optional[str] = None, country: Optional[str] = None) -> list:
cond = exp_rate.c.kind.isnot(None)
if kind:
cond &= exp_rate.c.kind == kind
if country:
cond &= exp_rate.c.country == country
return [self._rate(r) for r in _all(self.conn, exp_rate, cond, (exp_rate.c.kind, exp_rate.c.country, exp_rate.c.variant, exp_rate.c.band, exp_rate.c.valid_from))]
def put(self, r: rt.Rate, by: str, at: datetime) -> None:
_plain(self.conn, exp_rate, {"kind": r.kind, "country": r.country, "variant": r.variant, "band": r.band, "valid_from": r.valid_from,
"rate_amount": r.rate_amount, "currency": r.currency, "band_upto": r.band_upto, "unit": r.unit or None,
"partial_day_amount": r.partial_day_amount, "breakfast_pct": r.breakfast_pct, "lunch_pct": r.lunch_pct,
"dinner_pct": r.dinner_pct, "year_start": r.year_start, "source": r.source or None, "set_by": by}, at)
class AiRepository:
def __init__(self, conn):
self.conn = conn
def current_consent(self, person_id: int, step_key: str) -> Optional[dict]:
"""The latest consent row of a person for a step: ONE query."""
row = self.conn.execute(select(exp_consent).where((exp_consent.c.person_id == person_id) & (exp_consent.c.step_key == step_key))
.order_by(exp_consent.c.given_at.desc()).limit(1)).mappings().first()
return dict(row) if row else None
def add_consent(self, row: dict, at: datetime) -> None:
_plain(self.conn, exp_consent, row, at)
def add_finding(self, row: dict, at: datetime) -> None:
_plain(self.conn, exp_ai_finding, row, at)
def finding(self, report_id: str, line_no: int, finding_no: int) -> Optional[dict]:
row = self.conn.execute(select(exp_ai_finding).where((exp_ai_finding.c.report_id == report_id) & (exp_ai_finding.c.line_no == line_no)
& (exp_ai_finding.c.finding_no == finding_no))).mappings().first()
return dict(row) if row else None
def of_report(self, report_id: str) -> list:
return _all(self.conn, exp_ai_finding, exp_ai_finding.c.report_id == report_id, (exp_ai_finding.c.line_no, exp_ai_finding.c.finding_no))
def save_finding(self, row: dict, at: datetime) -> None:
upsert(self.conn, exp_ai_finding, [{**{k: v for k, v in row.items() if k != "created_at"}, "updated_at": _utc(at)}], ("report_id", "line_no", "finding_no"))
class MigrationRepository:
def __init__(self, conn):
self.conn = conn
def run(self, run_id: str) -> Optional[dict]:
row = self.conn.execute(select(exp_migration_run).where(exp_migration_run.c.run_id == run_id)).mappings().first()
return dict(row) if row else None
def runs(self) -> list:
return _all(self.conn, exp_migration_run, None, (exp_migration_run.c.run_id,))
def add_run(self, row: dict, at: datetime) -> None:
_plain(self.conn, exp_migration_run, row, at)
def save_run(self, row: dict, at: datetime) -> None:
upsert(self.conn, exp_migration_run, [{**{k: v for k, v in row.items() if k != "created_at"}, "updated_at": _utc(at)}], ("run_id",))
def mapped(self, db: str, model: str, ids: list) -> dict:
"""odoo_id -> our key for a set of Odoo ids: one query per 500."""
got: dict = {}
ids = sorted({str(i) for i in ids})
for i in range(0, len(ids), 500):
for r in _all(self.conn, exp_migration_map, (exp_migration_map.c.odoo_db == db) & (exp_migration_map.c.odoo_model == model)
& exp_migration_map.c.odoo_id.in_(ids[i:i + 500])):
got[r["odoo_id"]] = r
return got
def add_map(self, rows: list, at: datetime) -> None:
for i in range(0, len(rows), 500):
self.conn.execute(insert(exp_migration_map), [{**r, "created_at": _utc(at), "updated_at": _utc(at)} for r in rows[i:i + 500]])
def of_run(self, run_id: str, model: Optional[str] = None) -> list:
cond = exp_migration_map.c.run_id == run_id
if model:
cond &= exp_migration_map.c.odoo_model == model
return _all(self.conn, exp_migration_map, cond, (exp_migration_map.c.odoo_model, exp_migration_map.c.odoo_id))
def delete_run(self, run_id: str, report_ids: list) -> None:
"""Take a discarded run's rows back out: its claims, their lines and receipt rows, and its map. Only ever for a run nothing has left for."""
for i in range(0, len(report_ids), 400):
chunk = report_ids[i:i + 400]
for t in (exp_line, exp_receipt_image, exp_outbox):
self.conn.execute(delete(t).where(t.c.report_id.in_(chunk)))
self.conn.execute(delete(exp_report).where(exp_report.c.report_id.in_(chunk)))
self.conn.execute(delete(exp_migration_map).where(exp_migration_map.c.run_id == run_id))
def add_reports(self, reports: list, at: datetime) -> None:
"""Migrated claims in bulk (a handful of statements, not two per sheet)."""
rows = [{**ReportRepository._root_row(None, r), "created_at": _utc(at), "updated_at": _utc(at)} for r in reports]
for i in range(0, len(rows), 200):
self.conn.execute(insert(exp_report), rows[i:i + 200])
lines = [line_row(r.report_id, r.revision, ln, at) for r in reports for ln in r.lines]
for i in range(0, len(lines), 300):
self.conn.execute(insert(exp_line), lines[i:i + 300])
class Repository:
"""Every repository on one connection (one unit of work)."""
def __init__(self, conn):
self.conn = conn
self.reports = ReportRepository(conn)
self.claims = ClaimRepository(conn)
self.outbox = OutboxRepository(conn)
self.categories = CategoryRepository(conn)
self.settings = SettingsRepository(conn)
self.leavers = LeaverRepository(conn)
self.policy = PolicyRepository(conn)
self.delegations = DelegationRepository(conn)
self.cards = CardRepository(conn)
self.receipts = ReceiptRepository(conn)
self.rates = RateRepository(conn)
self.ai = AiRepository(conn)
self.migration = MigrationRepository(conn)
def seed(self, at: Optional[datetime] = None) -> None:
"""The categories every company starts with, once (company '*'). Idempotent; the same rows migrations 00002, 00003 and 00005 insert."""
at = at or work.now()
have = {(r["category_code"], r["valid_from"]) for r in _all(self.conn, exp_category, exp_category.c.company_code == "*")}
for code, name, purpose, tax, receipt, kind, since in DEFAULT_CATEGORIES:
if (code, date.fromisoformat(since)) not in have:
self.categories.put("*", code, name, purpose, tax, date.fromisoformat(since), at, receipt, kind)
have_rates = {(r.kind, r.country, r.variant, r.band, r.valid_from) for r in self.rates.listed()}
for kind, country, variant, band, since, amount, cur, upto, unit, partial, b, lu, di, year, source in DEFAULT_RATES:
if (kind, country, variant, band, date.fromisoformat(since)) not in have_rates:
D = lambda v: Decimal(v) if v is not None else None # noqa: E731
self.rates.put(rt.Rate(kind, country, variant, band, date.fromisoformat(since), D(amount), cur, D(upto), unit, D(partial), D(b), D(lu), D(di), year, source),
"migration", at)