dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/infrastructure/tables.py
"""The tables, and one repository per aggregate. Every table is prefixed `pay_`, 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 equal.
Rows that are FACTS (a receipt, an allocation, a collection event, a limit, an override, a statement) are written
with a plain INSERT: a second writer of the same key is rejected, never silently overwritten (DataK3's
ON CONFLICT DO UPDATE rewrites every column in VALUES). Only the few rows that move FORWARD - a receipt's reversal,
a refund's status, a suggestion's decision, a reminder draft sent - are written with the keyed upsert.
"""
import copy
import json
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal
from typing import Iterable, Optional
from zoneinfo import ZoneInfo
from erp_platform import work
from erp_platform.repository import upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
func,
insert,
select,
union_all,
)
from brick_payments.domain import collection as coll
from brick_payments.domain.credit import Limit, Override
from brick_payments.domain.receipt import Receipt
from brick_payments.domain.refund import Refund
from brick_payments.domain.settlement import CREDIT_PREFIX, Allocation
from brick_payments.domain.values import SETTING_DEFS, Settings
md = MetaData()
RATE = Numeric(19, 4)
MONEY = Numeric(18, 2)
def _cols(table: Table, *skip) -> tuple:
return tuple(c.name for c in table.columns if c.name not in ("created_at", "updated_at", *skip))
pay_receipt = Table("pay_receipt", md,
Column("payment_id", String, primary_key=True),
Column("bp_id", BigInteger),
Column("currency", String),
Column("amount", MONEY),
Column("received_on", Date),
Column("method", String),
Column("bank_account_id", String),
Column("bank_reference", String),
Column("remittance_text", String),
Column("status", String),
Column("batch_id", String),
Column("fx_rate", RATE),
Column("returned_on", Date),
Column("bank_return_ref", String),
Column("reversed_on", Date),
Column("origin", String),
Column("recorded_by", String),
Column("recorded_at", TIMESTAMP),
Column("cleared_by", String),
Column("recorded_on", Date), # the business day it was recorded (as-at views judge a receipt by it)
Column("return_reason_code", String), # v4: the ARUDD reason of a returned direct debit
Column("statement_line_id", String), # v4: the statement line this receipt came from
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A bank reference may be on ONE receipt per bank account: claimed with a plain INSERT, so two cashiers recording
# the same transfer at once cannot both win.
pay_receipt_ref = Table("pay_receipt_ref", md,
Column("bank_account_id", String, primary_key=True),
Column("bank_reference", String, primary_key=True),
Column("payment_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_allocation = Table("pay_allocation", md,
Column("source_kind", String, primary_key=True),
Column("source_id", String, primary_key=True),
Column("invoice_id", String, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("amount", MONEY),
Column("currency", String),
Column("allocated_on", Date),
Column("allocated_by", String),
Column("reason", String),
Column("reverses_seq", Integer),
Column("suggestion_id", String),
Column("decided_how", String),
Column("origin", String),
Column("bp_id", BigInteger),
Column("source_amount", MONEY),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# Claim rows (audit 1). DataK3 detects a write conflict only on the SAME key, and an allocation's key is (source, invoice,
# seq) - two writers spending one invoice from two receipts, or one receipt over two invoices, never share it. So every
# allocation also claims, with a plain INSERT in the same transaction, the next slot of each invoice it settles and of each
# source it draws on: n = the committed rows already there + 1. Two writers who read the same state claim the same key,
# and the second is refused (409, retry-safe).
pay_invoice_claim = Table("pay_invoice_claim", md,
Column("invoice_id", String, primary_key=True),
Column("n", Integer, primary_key=True),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_source_claim = Table("pay_source_claim", md,
Column("source_kind", String, primary_key=True),
Column("source_id", String, primary_key=True),
Column("n", Integer, primary_key=True),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# business_timezone as DATED rows: a change takes effect from the NEXT business date (audit 1), so no date already used
# is ever re-counted in another zone. pay_settings holds the zone before the first row (the default until one is set).
pay_timezone = Table("pay_timezone", md,
Column("valid_from", Date, primary_key=True),
Column("zone", String),
Column("set_by", String),
Column("set_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_match_suggestion = Table("pay_match_suggestion", md,
Column("suggestion_id", String, primary_key=True),
Column("payment_id", String),
Column("proposed_lines", String),
Column("evidence", String),
Column("confidence", Numeric(5, 2)),
Column("model", String),
Column("exact", Integer),
Column("status", String),
Column("decided_by", String),
Column("decided_at", TIMESTAMP),
Column("decided_lines", String),
Column("reviewed_by", String),
Column("reviewed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_refund = Table("pay_refund", md,
Column("refund_id", String, primary_key=True),
Column("bp_id", BigInteger),
Column("currency", String),
Column("amount", MONEY),
Column("status", String),
Column("requested_by", String),
Column("requested_at", TIMESTAMP),
Column("approved_by", String),
Column("approved_at", TIMESTAMP),
Column("approved_on", Date),
Column("reason", String),
Column("bank_ref", String),
Column("paid_on", Date),
Column("origin", String),
Column("requested_sub", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_collection = Table("pay_collection", md,
Column("bp_id", BigInteger, primary_key=True),
Column("kind", String, primary_key=True),
Column("level", Integer, primary_key=True),
Column("ref_id", String, primary_key=True),
Column("invoice_id", String),
Column("occurred_on", Date),
Column("promise_date", Date),
Column("reason", String),
Column("outcome", String),
Column("note", String),
Column("draft_text", String),
Column("status", String),
Column("origin", String),
Column("recorded_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_statement = Table("pay_statement", md,
Column("bp_id", BigInteger, primary_key=True),
Column("as_at", Date, primary_key=True),
Column("currency", String, primary_key=True),
Column("closing_balance", MONEY),
Column("sent_at", TIMESTAMP),
Column("sent_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_credit = Table("pay_credit", md,
Column("bp_id", BigInteger, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("credit_limit_amount", MONEY),
Column("currency", String),
Column("reason", String),
Column("set_by", String),
Column("set_at", TIMESTAMP),
Column("origin", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_credit_override = Table("pay_credit_override", md,
Column("bp_id", BigInteger, primary_key=True),
Column("kind", String, primary_key=True),
Column("starts_at", TIMESTAMP, primary_key=True),
Column("ends_on", Date),
Column("reason", String),
Column("set_by", String),
Column("origin", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_bank_account = Table("pay_bank_account", md,
Column("bank_account_id", String, primary_key=True),
Column("name", String),
Column("currency", String),
Column("iban_or_sort_code_masked", String),
Column("gl_account_id", String),
Column("origin", String),
Column("house_bank_id", String),
Column("company_code", String),
Column("purposes", String),
Column("account_name", String),
Column("status", String),
Column("closed_from", Date),
Column("account_ref", String), # the full account number / IBAN: for matching our own accounts and the Bacs file; never shown
Column("iban", String), # v4 slice B: the IBAN of an account that has one (served unmasked to purchasing's machine identity only)
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# v4 slice C. The realised FX difference posted for one allocation row (PAY-33), keyed on the row: the rates and the amount are kept,
# so the undo reverses exactly what was posted and never re-reads a rate.
pay_fx_posting = Table("pay_fx_posting", md,
Column("payment_id", String, primary_key=True),
Column("invoice_id", String, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("event", String),
Column("direction", String),
Column("source_document", String),
Column("journal_id", String),
Column("amount", MONEY),
Column("currency", String),
Column("receipt_rate", RATE),
Column("invoice_rate", RATE),
Column("posted_on", Date),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# An open-banking feed's consent (PAY-35): read-only, granted by the owner for at most 90 days. A new grant is a new row.
pay_feed_consent = Table("pay_feed_consent", md,
Column("bank_account_id", String, primary_key=True),
Column("granted_on", Date, primary_key=True),
Column("provider", String),
Column("expires_on", Date),
Column("recorded_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_reminder_level = Table("pay_reminder_level", md,
Column("policy_id", String, primary_key=True),
Column("level", Integer, primary_key=True),
Column("days_after_due", Integer),
Column("channel", String),
Column("template", String),
Column("mode", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_settings = Table("pay_settings", md,
Column("setting_key", String, primary_key=True),
Column("value", String),
Column("changes", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# Who changed a setting, from what, to what and when (PAY-13): appended, never edited.
pay_settings_log = Table("pay_settings_log", md,
Column("setting_key", String, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("old_value", String),
Column("new_value", String),
Column("changed_by", String),
Column("changed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The Odoo migration (PAY-44): what each Odoo record became here, so a second run matches every row.
pay_odoo_map = Table("pay_odoo_map", md,
Column("odoo_ref", String, primary_key=True),
Column("target_table", String),
Column("target_key", String),
Column("migrated_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One row per migration: the Odoo side's figures as read at the cut-over, and the owner's signature on them (PAY-15).
pay_migration_run = Table("pay_migration_run", md,
Column("run_id", String, primary_key=True),
Column("params", String),
Column("odoo_figures", String),
Column("counts", String),
Column("ran_by", String),
Column("ran_at", TIMESTAMP),
Column("signed_by", String),
Column("signed_at", TIMESTAMP),
Column("signed_figures", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# ═══ v4 slice A: statements, rules, remittances, direct debits ═══════════════════════════════════════════
# A generic claim row (audit habit): ONE guard key per bounded thing, written with a plain INSERT by EVERY operation that touches
# it, in its own transaction: n = the committed claims already on the key + 1, so two writers who read the same state claim the
# same (scope, key, n) and the second is refused (409). scope 'line' = a statement line; 'counterpart' = what a line is matched to;
# 'ddinv' = an invoice in a direct-debit run; 'discount' = an invoice's settlement discount; 'dd_report' = one ARUDD/ADDACS line.
pay_claim = Table("pay_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))
# Objects (the Bacs / pain.008 files): kept by key, never edited.
pay_object = Table("pay_object", md,
Column("object_key", String, primary_key=True),
Column("content_type", String),
Column("sha256", String),
Column("body", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_house_bank = Table("pay_house_bank", md,
Column("house_bank_id", String, primary_key=True),
Column("name", String),
Column("bic", String),
Column("sort_code", String),
Column("address", String),
Column("sun", String),
Column("creditor_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_bank_statement = Table("pay_bank_statement", md,
Column("statement_id", String, primary_key=True),
Column("bank_account_id", String),
Column("from_date", Date),
Column("to_date", Date),
Column("opening_balance", MONEY),
Column("closing_balance", MONEY),
Column("format", String),
Column("file_key", String),
Column("lifecycle", String),
Column("reconciled_by", String),
Column("reconciled_at", TIMESTAMP),
Column("imported_by", String),
Column("imported_on", Date),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_statement_line = Table("pay_statement_line", md,
Column("line_id", String, primary_key=True),
Column("statement_id", String),
Column("bank_account_id", String),
Column("booked_on", Date),
Column("value_on", Date),
Column("amount", MONEY), # signed: a credit is positive, a debit negative
Column("currency", String),
Column("counterparty_name", String),
Column("counterparty_account_masked", String),
Column("text", String),
Column("bank_tx_code", String),
Column("lifecycle", String),
Column("matched_kind", String),
Column("matched_ref", String),
Column("account_id", String),
Column("tax_code", String),
Column("cost_centre", String),
Column("journal_id", String),
Column("decided_how", String),
Column("decided_by", String),
Column("decided_on", Date),
Column("receipt_image", String),
Column("vat_amount", MONEY),
Column("vat_reclaimed", String),
Column("reversal_reason", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_line_suggestion = Table("pay_line_suggestion", md,
Column("suggestion_id", String, primary_key=True),
Column("line_id", String),
Column("account_id", String),
Column("tax_code", String),
Column("cost_centre", String),
Column("receipt_needed", String),
Column("evidence", String),
Column("confidence", String),
Column("model", String),
Column("status", String),
Column("decided_by", String),
Column("decided_on", Date),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_reconciliation_rule = Table("pay_reconciliation_rule", md,
Column("rule_id", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("bank_account_id", String),
Column("direction", String),
Column("text_contains", String),
Column("counterparty_contains", String),
Column("amount_min", MONEY),
Column("amount_max", MONEY),
Column("account_id", String),
Column("tax_code", String),
Column("cost_centre", String),
Column("mode", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_remittance = Table("pay_remittance", md,
Column("remittance_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("file_ref", String),
Column("format", String),
Column("payer", String),
Column("payment_id", String),
Column("bp_id", BigInteger),
Column("invoice_ref", String),
Column("invoice_id", String),
Column("amount", MONEY),
Column("currency", String),
Column("deduction_reason", String),
Column("status", String),
Column("suggestion_id", String),
Column("imported_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_dd_mandate = Table("pay_dd_mandate", md,
Column("sun", String, primary_key=True),
Column("mandate_ref", String, primary_key=True),
Column("bp_id", BigInteger),
Column("bank_account_id", String),
Column("payer_sort_code", String),
Column("payer_account_masked", String),
Column("payer_account_name", String),
Column("scheme", String),
Column("lifecycle", String),
Column("lodged_on", Date),
Column("auddis_result", String),
Column("cancelled_reason", String),
Column("cancel_lodged_on", Date),
Column("last_collected_on", Date),
Column("recorded_on", Date),
Column("recorded_by", String),
Column("notice_days", Integer),
Column("ends_on", Date),
Column("disputed_on", Date),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The payer's full bank details, DATED (an ADDACS change starts from its date; a row is never edited). Read only to build the Bacs file.
pay_dd_mandate_details = Table("pay_dd_mandate_details", md,
Column("sun", String, primary_key=True),
Column("mandate_ref", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("sort_code", String),
Column("account_number", String),
Column("source", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_dd_run = Table("pay_dd_run", md,
Column("run_id", String, primary_key=True),
Column("bank_account_id", String),
Column("collection_date", Date),
Column("input_day", Date),
Column("lifecycle", String),
Column("prepared_by", String),
Column("prepared_by_sub", String),
Column("approved_by", String),
Column("approved_by_sub", String),
Column("file_key", String),
Column("submission_ref", String),
Column("total", MONEY),
Column("count", Integer),
Column("currency", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_dd_collection = Table("pay_dd_collection", md,
Column("run_id", String, primary_key=True),
Column("mandate_ref", String, primary_key=True),
Column("bp_id", BigInteger),
Column("sun", String),
Column("invoice_ids", String),
Column("amount", MONEY),
Column("transaction_code", String),
Column("notice_sent_on", Date),
Column("status", String),
Column("payment_id", String),
Column("return_reason_code", String),
Column("returned_on", Date),
Column("represented", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# ── v4 slice B: default receiving accounts, the unmasked-read audit, cash, deposit batches, forecast snapshots ──────────────────
pay_default_bank_account = Table("pay_default_bank_account", md,
Column("company_code", String, primary_key=True),
Column("currency", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("bank_account_id", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One row per answered read of an unmasked account (PAY-27): written in the read's own transaction, before the answer.
pay_bank_account_read = Table("pay_bank_account_read", md,
Column("read_id", String, primary_key=True),
Column("bank_account_id", String),
Column("caller", String),
Column("requested_at", TIMESTAMP),
Column("purpose", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_cash_journal = Table("pay_cash_journal", md,
Column("cash_journal_id", String, primary_key=True),
Column("company_code", String),
Column("currency", String),
Column("gl_account_id", String),
Column("name", String),
Column("status", String),
Column("opening_balance", MONEY),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_cash_day = Table("pay_cash_day", md,
Column("cash_journal_id", String, primary_key=True),
Column("day", Date, primary_key=True),
Column("lifecycle", String),
Column("opening_balance", MONEY),
Column("expected_balance", MONEY),
Column("counted_amount", MONEY),
Column("difference", MONEY),
Column("counted_by", String),
Column("approved_by", String),
Column("approval_reason", String),
Column("closed_by", String),
Column("closed_at", TIMESTAMP),
Column("difference_journal_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_cash_entry = Table("pay_cash_entry", md,
Column("cash_journal_id", String, primary_key=True),
Column("day", Date, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("kind", String),
Column("amount", MONEY),
Column("currency", String),
Column("bp_id", BigInteger),
Column("payment_id", String),
Column("account_id", String),
Column("tax_code", String),
Column("description", String),
Column("receipt_image", String),
Column("vat_amount", MONEY),
Column("vat_reclaimed", String),
Column("journal_id", String),
Column("statement_line_id", String),
Column("recorded_by", String),
Column("recorded_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_cash_count = Table("pay_cash_count", md,
Column("cash_journal_id", String, primary_key=True),
Column("day", Date, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("counted_amount", MONEY),
Column("denominations", String),
Column("counted_by", String),
Column("counted_by_ids", String),
Column("counted_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_deposit_batch = Table("pay_deposit_batch", md,
Column("batch_id", String, primary_key=True),
Column("bank_account_id", String),
Column("currency", String),
Column("slip_no", String),
Column("lifecycle", String),
Column("total", MONEY),
Column("cheque_count", Integer),
Column("deposited_on", Date),
Column("statement_line_id", String),
Column("final_batch_id", String),
Column("opened_by", String),
Column("kind", String), # v6: cheque | cash (PAY-49: a bag another module sent to the bank)
Column("batch_ref", String), # v6: the handing module's key, dep:<module>:<shop_id>:<bag_no>
Column("source_module", String),
Column("clearing_purpose", String), # v6: cash_in_transit
Column("expected_on", Date),
Column("company_code", String),
Column("difference", MONEY), # v6: the counted difference an approver accepted
Column("accepted_by", String),
Column("accepted_sub", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# v6 (PAY-50): what a card provider owes us for one terminal batch, matched to its net credit.
pay_card_settlement = Table("pay_card_settlement", md,
Column("settlement_ref", String, primary_key=True),
Column("provider", String),
Column("company_code", String),
Column("currency", String),
Column("gross", MONEY),
Column("business_date", Date),
Column("transaction_count", Integer),
Column("source_module", String),
Column("lifecycle", String), # expected | matched | differs
Column("bank_account_id", String),
Column("statement_line_id", String),
Column("fees", MONEY),
Column("fee_pct", RATE),
Column("matched_by", String),
Column("matched_sub", String),
Column("accepted_by", String),
Column("accepted_sub", String),
Column("accept_reason", String),
Column("claim", String), # "" | fees | claim: where an accepted difference went
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# v6 (PAY-51): money another module owes a person, paid from our bank in a payout run.
pay_payout = Table("pay_payout", md,
Column("payout_ref", String, primary_key=True),
Column("payout_id", String),
Column("payee_bp_id", BigInteger),
Column("person_id", String),
Column("company_code", String),
Column("currency", String),
Column("amount", MONEY),
Column("due_on", Date),
Column("reference", String),
Column("source_module", String),
Column("source_document", String),
Column("clearing_purpose", String),
Column("lifecycle", String), # requested | approved | exported | paid | returned | cancelled
Column("bank_account_id", String),
Column("requested_by", String),
Column("approved_by", String),
Column("approved_sub", String),
Column("approved_on", Date),
Column("exported_by", String),
Column("exported_sub", String),
Column("exported_on", Date),
Column("export_key", String),
Column("paid_on", Date),
Column("bank_ref", String),
Column("statement_line_id", String),
Column("journal_id", String),
Column("returned_on", Date),
Column("cancel_reason", String),
Column("cancelled_by", String),
Column("told_source_at", TIMESTAMP), # when the source module acknowledged 'paid' (retried until it does)
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pay_forecast_snapshot = Table("pay_forecast_snapshot", md,
Column("snapshot_id", String, primary_key=True),
Column("saved_on", Date),
Column("saved_by", String),
Column("body", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
TABLE_ORDER = [t.name for t in md.sorted_tables]
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 _dec(v) -> Optional[Decimal]:
return None if v is None else Decimal(str(v))
def _j(v) -> str:
return json.dumps(v, default=str, sort_keys=True)
class ReceiptRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _make(m) -> Receipt:
return Receipt(m["payment_id"], m["bp_id"], m["currency"], _dec(m["amount"]), m["received_on"],
m["method"] or "", m["bank_account_id"] or "", m["bank_reference"] or "", m["remittance_text"] or "",
m["status"], m["batch_id"] or "", _dec(m["fx_rate"]), m["returned_on"], m["bank_return_ref"] or "",
m["reversed_on"], m["origin"] or "", m["recorded_by"] or "", m["recorded_at"],
recorded_on=m["recorded_on"] or (m["recorded_at"].date() if m["recorded_at"] else m["received_on"]),
return_reason_code=m["return_reason_code"] or "", statement_line_id=m["statement_line_id"] or "")
@staticmethod
def _row(r: Receipt) -> dict:
return {"payment_id": r.payment_id, "bp_id": r.bp_id, "currency": r.currency, "amount": r.amount,
"received_on": r.received_on, "method": r.method, "bank_account_id": r.bank_account_id,
"bank_reference": r.bank_reference, "remittance_text": r.remittance_text, "status": r.status,
"batch_id": r.batch_id, "fx_rate": r.fx_rate, "returned_on": r.returned_on,
"bank_return_ref": r.bank_return_ref, "reversed_on": r.reversed_on, "origin": r.origin,
"recorded_by": r.recorded_by, "recorded_at": _utc(r.recorded_at), "recorded_on": r.recorded_on,
"return_reason_code": r.return_reason_code, "statement_line_id": r.statement_line_id}
def get(self, payment_id: str) -> Optional[Receipt]:
m = self.conn.execute(select(pay_receipt).where(pay_receipt.c.payment_id == payment_id)).mappings().first()
return self._make(m) if m else None
def get_many(self, ids: Iterable[str]) -> dict:
"""{payment_id: Receipt} for a set of receipts in one query, however many there are."""
ids = list(dict.fromkeys(ids))
if not ids:
return {}
q = select(pay_receipt).where(pay_receipt.c.payment_id.in_(ids))
return {m["payment_id"]: self._make(m) for m in self.conn.execute(q).mappings()}
def not_on_a_statement(self, bank_account_id: str, up_to: date) -> list:
"""Live receipts typed in against this account that no statement line stands for yet, received on or before `up_to`."""
q = select(pay_receipt).where((pay_receipt.c.bank_account_id == bank_account_id) & (pay_receipt.c.status == "received") &
(pay_receipt.c.received_on <= up_to) & ((pay_receipt.c.statement_line_id == "") | pay_receipt.c.statement_line_id.is_(None)))
return [self._make(m) for m in self.conn.execute(q.order_by(pay_receipt.c.received_on, pay_receipt.c.payment_id)).mappings()]
def existing(self, ids: Iterable[str]) -> set:
ids = list(ids)
if not ids:
return set()
return {r[0] for r in self.conn.execute(select(pay_receipt.c.payment_id).where(pay_receipt.c.payment_id.in_(ids)))}
def add(self, r: Receipt) -> None:
"""A new receipt: a plain INSERT, so the same payment_id from two writers is rejected, not overwritten."""
self.conn.execute(insert(pay_receipt).values(**self._row(r), created_at=_utc(r.recorded_at), updated_at=_utc(r.recorded_at)))
def save(self, r: Receipt) -> None:
"""A receipt that moves forward (reversed, or a suspense customer cleared)."""
upsert(self.conn, pay_receipt, [self._row(r)], ("payment_id",))
def save_many(self, receipts: Iterable[Receipt]) -> None:
"""Receipts that move forward together: one keyed upsert for all of them."""
rows = [self._row(r) for r in receipts]
if rows:
upsert(self.conn, pay_receipt, rows, ("payment_id",))
def clear_suspense(self, payment_id: str, bp_id: int, by: str) -> None:
upsert(self.conn, pay_receipt, [{"payment_id": payment_id, "bp_id": bp_id, "cleared_by": by}], ("payment_id",))
def cleared_by(self, payment_id: str) -> str:
return self.conn.execute(select(pay_receipt.c.cleared_by).where(pay_receipt.c.payment_id == payment_id)).scalar() or ""
def search(self, *, bp_id: Optional[int] = None, status: Optional[str] = None, currency: Optional[str] = None,
limit: int = 200, offset: int = 0) -> list:
q = select(pay_receipt).order_by(pay_receipt.c.received_on.desc(), pay_receipt.c.payment_id)
for col, v in ((pay_receipt.c.bp_id, bp_id), (pay_receipt.c.status, status), (pay_receipt.c.currency, currency)):
if v is not None:
q = q.where(col == v)
return [self._make(m) for m in self.conn.execute(q.limit(limit).offset(offset)).mappings()]
def live_of(self, bp_id: int) -> list:
return self.search(bp_id=bp_id, limit=100000)
def ref_holder(self, bank_account_id: str, bank_reference: str) -> Optional[str]:
return self.conn.execute(select(pay_receipt_ref.c.payment_id).where(
(pay_receipt_ref.c.bank_account_id == bank_account_id) &
(pay_receipt_ref.c.bank_reference == bank_reference))).scalar()
def ref_holder_anywhere(self, bank_reference: str) -> Optional[str]:
"""The receipt that claimed this reference on ANY bank account, or on none: a transfer typed in with no account
and the same transfer imported later on a bank account are one transfer."""
return self.conn.execute(select(pay_receipt_ref.c.payment_id).where(
pay_receipt_ref.c.bank_reference == bank_reference)).scalars().first()
def claim_ref(self, bank_account_id: str, bank_reference: str, payment_id: str, at: datetime) -> None:
self.conn.execute(insert(pay_receipt_ref).values(
bank_account_id=bank_account_id, bank_reference=bank_reference, payment_id=payment_id,
created_at=_utc(at), updated_at=_utc(at)))
class AllocationRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _make(m) -> Allocation:
return Allocation(m["source_kind"], m["source_id"], m["invoice_id"], m["seq"], _dec(m["amount"]), m["currency"],
m["allocated_on"], m["allocated_by"] or "", m["reason"] or "", m["reverses_seq"],
m["suggestion_id"] or "", m["decided_how"] or "manual", m["origin"] or "", m["bp_id"],
_dec(m["source_amount"]))
def _where(self, cond, limit: Optional[int] = None) -> list:
q = select(pay_allocation).where(cond).order_by(pay_allocation.c.source_kind, pay_allocation.c.source_id,
pay_allocation.c.invoice_id, pay_allocation.c.seq)
return [self._make(m) for m in self.conn.execute(q).mappings()]
def for_invoices(self, ids: Iterable[str]) -> list:
"""Rows that settle these invoices (refund rows name a credit, never an invoice)."""
ids = list(ids)
return self._where(pay_allocation.c.invoice_id.in_(ids)) if ids else []
def for_source(self, kind: str, source_id: str) -> list:
return self._where((pay_allocation.c.source_kind == kind) & (pay_allocation.c.source_id == source_id))
def for_credit(self, kind: str, source_id: str) -> list:
"""Everything that drew on a receipt or credit note: its own allocations AND the refunds paid out of it."""
from brick_payments.domain.settlement import credit_ref
ref = credit_ref(kind, source_id)
return self._where(((pay_allocation.c.source_kind == kind) & (pay_allocation.c.source_id == source_id)) |
((pay_allocation.c.source_kind == "refund") & (pay_allocation.c.invoice_id == ref)))
def for_bp(self, bp_id: int) -> list:
return self._where(pay_allocation.c.bp_id == bp_id)
def for_bps(self, bp_ids: Iterable[int]) -> dict:
"""{bp_id: [Allocation]} for a set of customers in one query (a loop of for_bp() is one round trip per customer)."""
ids = list(dict.fromkeys(bp_ids))
out = {i: [] for i in ids}
for a in (self._where(pay_allocation.c.bp_id.in_(ids)) if ids else []):
out.setdefault(a.bp_id, []).append(a)
return out
def all(self) -> list:
return self._where(pay_allocation.c.seq >= 0) + self._where(pay_allocation.c.seq < 0)
def get(self, kind: str, source_id: str, invoice_id: str, seq: int) -> Optional[Allocation]:
rows = self._where((pay_allocation.c.source_kind == kind) & (pay_allocation.c.source_id == source_id) &
(pay_allocation.c.invoice_id == invoice_id) & (pay_allocation.c.seq == seq))
return rows[0] if rows else None
def pair(self, kind: str, source_id: str, invoice_id: str) -> list:
return self._where((pay_allocation.c.source_kind == kind) & (pay_allocation.c.source_id == source_id) &
(pay_allocation.c.invoice_id == invoice_id))
def next_seq(self, kind: str, source_id: str, invoice_id: str) -> int:
return max((r.seq for r in self.pair(kind, source_id, invoice_id)), default=0) + 1
def next_seqs(self, kind: str, source_id: str, invoice_ids: Iterable[str]) -> dict:
"""{invoice_id: next seq} of one source against several invoices, in ONE query (next_seq() of each, without a read each)."""
ids = list(dict.fromkeys(invoice_ids))
if not ids:
return {}
rows = self._where((pay_allocation.c.source_kind == kind) & (pay_allocation.c.source_id == source_id) &
pay_allocation.c.invoice_id.in_(ids))
return {i: max((r.seq for r in rows if r.invoice_id == i), default=0) + 1 for i in ids}
def existing_keys(self, keys: list) -> set:
found = set()
for k in keys:
if self.get(*k) is not None:
found.add(tuple(k))
return found
def add(self, rows: Iterable[Allocation], at: datetime) -> None:
"""Plain INSERTs: a racing writer of the same (source, invoice, seq) is rejected (409), never overwritten."""
values = [dict(source_kind=r.source_kind, source_id=r.source_id, invoice_id=r.invoice_id, seq=r.seq, amount=r.amount,
currency=r.currency, allocated_on=r.allocated_on, allocated_by=r.allocated_by, reason=r.reason,
reverses_seq=r.reverses_seq, suggestion_id=r.suggestion_id, decided_how=r.decided_how, origin=r.origin,
bp_id=r.bp_id, source_amount=r.source_amount, created_at=_utc(at), updated_at=_utc(at)) for r in rows]
if values: # one statement: a racer's key fails it whole, as the first row of a loop would
self.conn.execute(insert(pay_allocation).values(values))
def claim(self, *, invoices: Iterable[str], sources: Iterable[tuple], at: datetime) -> None:
"""Claim the next slot of each invoice and each credit (receipt / credit note / write-off) about to be drawn on. Call
BEFORE add(): the count is the committed rows, and a concurrent writer that read the same rows claims the same key."""
invs = sorted(set(invoices))
if invs: # every invoice's committed rows in one read, every claim in one INSERT
held: dict = {}
for a in self.for_invoices(invs):
held[a.invoice_id] = held.get(a.invoice_id, 0) + 1
self.conn.execute(insert(pay_invoice_claim).values([
dict(invoice_id=inv, n=held.get(inv, 0) + 1, claimed_at=_utc(at), created_at=_utc(at), updated_at=_utc(at))
for inv in invs]))
for kind, source_id in sorted(set(sources)):
self.conn.execute(insert(pay_source_claim).values(
source_kind=kind, source_id=source_id,
n=len(self.for_credit(kind, source_id) if kind in CREDIT_PREFIX else self.for_source(kind, source_id)) + 1,
claimed_at=_utc(at), created_at=_utc(at), updated_at=_utc(at)))
def with_received(self, cond, order=True) -> list:
"""(allocation, received_on) pairs: a receipt's rows carry the receipt's own received_on (PAY-46), any other source's is None."""
q = (select(pay_allocation, pay_receipt.c.received_on.label("_received_on"))
.select_from(pay_allocation.outerjoin(pay_receipt, (pay_allocation.c.source_kind == "receipt") &
(pay_receipt.c.payment_id == pay_allocation.c.source_id)))
.where(cond).order_by(pay_allocation.c.source_kind, pay_allocation.c.source_id, pay_allocation.c.invoice_id, pay_allocation.c.seq))
return [(self._make(m), m["_received_on"]) for m in self.conn.execute(q).mappings()]
def auto_on(self, day: date) -> list:
return self._where((pay_allocation.c.decided_how == "auto") & (pay_allocation.c.allocated_on == day) &
(pay_allocation.c.amount > 0))
class SuggestionRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _view(m) -> dict:
return {"suggestion_id": m["suggestion_id"], "payment_id": m["payment_id"],
"proposed_lines": json.loads(m["proposed_lines"] or "[]"), "evidence": json.loads(m["evidence"] or "{}"),
"confidence": str(_dec(m["confidence"])), "model": m["model"], "exact": bool(m["exact"]), "status": m["status"],
"decided_by": m["decided_by"] or "", "decided_at": m["decided_at"].isoformat() if m["decided_at"] else None,
"decided_lines": json.loads(m["decided_lines"]) if m["decided_lines"] else None,
"reviewed_by": m["reviewed_by"] or "", "created_at": m["created_at"].isoformat() if m["created_at"] else None}
def get(self, suggestion_id: str) -> Optional[dict]:
m = self.conn.execute(select(pay_match_suggestion).where(
pay_match_suggestion.c.suggestion_id == suggestion_id)).mappings().first()
return self._view(m) if m else None
def get_many(self, suggestion_ids: Iterable[str]) -> dict:
"""{suggestion_id: view} for a set of suggestions in one query."""
ids = list(dict.fromkeys(suggestion_ids))
if not ids:
return {}
q = select(pay_match_suggestion).where(pay_match_suggestion.c.suggestion_id.in_(ids))
return {m["suggestion_id"]: self._view(m) for m in self.conn.execute(q).mappings()}
def of_receipt(self, payment_id: str) -> list:
q = select(pay_match_suggestion).where(pay_match_suggestion.c.payment_id == payment_id).order_by(
pay_match_suggestion.c.suggestion_id)
views = [self._view(m) for m in self.conn.execute(q).mappings()]
return sorted(views, key=lambda v: int(v["suggestion_id"].rsplit("#", 1)[1]))
def open_of_many(self, payment_ids: Iterable[str]) -> dict:
"""{payment_id: its latest open suggestion or None} for a set of receipts in one query."""
ids = list(dict.fromkeys(payment_ids))
out = {i: None for i in ids}
if ids:
q = select(pay_match_suggestion).where((pay_match_suggestion.c.payment_id.in_(ids)) & (pay_match_suggestion.c.status == "open"))
for v in sorted((self._view(m) for m in self.conn.execute(q).mappings()), key=lambda v: int(v["suggestion_id"].rsplit("#", 1)[1])):
out[v["payment_id"]] = v
return out
def open_of(self, payment_id: str) -> Optional[dict]:
return next((v for v in reversed(self.of_receipt(payment_id)) if v["status"] == "open"), None)
def add(self, view: dict, at: datetime) -> None:
self.conn.execute(insert(pay_match_suggestion).values(
suggestion_id=view["suggestion_id"], payment_id=view["payment_id"], proposed_lines=_j(view["proposed_lines"]),
evidence=_j(view["evidence"]), confidence=view["confidence"], model=view["model"], exact=int(view["exact"]),
status=view["status"], created_at=_utc(at), updated_at=_utc(at)))
def move(self, suggestion_id: str, **columns) -> None:
"""A suggestion's status moves forward; nothing else about it changes."""
cols = {k: (_utc(v) if isinstance(v, datetime) else (_j(v) if k == "decided_lines" else v)) for k, v in columns.items()}
upsert(self.conn, pay_match_suggestion, [{"suggestion_id": suggestion_id, **cols}], ("suggestion_id",))
def all_open(self) -> list:
q = select(pay_match_suggestion).where(pay_match_suggestion.c.status == "open")
return [self._view(m) for m in self.conn.execute(q).mappings()]
class RefundRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _make(m) -> Refund:
return Refund(m["refund_id"], m["bp_id"], m["currency"], _dec(m["amount"]), m["status"], m["requested_by"] or "",
m["requested_at"], m["approved_by"] or "", m["approved_at"], m["approved_on"], m["reason"] or "",
m["bank_ref"] or "", m["paid_on"], m["origin"] or "", m["requested_sub"] or "")
@staticmethod
def _row(r: Refund) -> dict:
return {"refund_id": r.refund_id, "bp_id": r.bp_id, "currency": r.currency, "amount": r.amount, "status": r.status,
"requested_by": r.requested_by, "requested_at": _utc(r.requested_at), "approved_by": r.approved_by,
"approved_at": _utc(r.approved_at), "approved_on": r.approved_on, "reason": r.reason,
"bank_ref": r.bank_ref, "paid_on": r.paid_on, "origin": r.origin, "requested_sub": r.requested_sub}
def get(self, refund_id: str) -> Optional[Refund]:
m = self.conn.execute(select(pay_refund).where(pay_refund.c.refund_id == refund_id)).mappings().first()
return self._make(m) if m else None
def add(self, r: Refund) -> None:
self.conn.execute(insert(pay_refund).values(**self._row(r), created_at=_utc(r.requested_at), updated_at=_utc(r.requested_at)))
def save(self, r: Refund) -> None:
upsert(self.conn, pay_refund, [self._row(r)], ("refund_id",))
def search(self, *, bp_id: Optional[int] = None, status: Optional[str] = None, limit: int = 200) -> list:
q = select(pay_refund).order_by(pay_refund.c.requested_at.desc(), pay_refund.c.refund_id)
if bp_id is not None:
q = q.where(pay_refund.c.bp_id == bp_id)
if status is not None:
q = q.where(pay_refund.c.status == status)
return [self._make(m) for m in self.conn.execute(q.limit(limit)).mappings()]
def reserved(self, bp_id: int, currency: str) -> Decimal:
"""Refunds asked for and not yet paid: they hold credit so two requests cannot spend the same money."""
q = select(func.coalesce(func.sum(pay_refund.c.amount), 0)).where(
(pay_refund.c.bp_id == bp_id) & (pay_refund.c.currency == currency) &
pay_refund.c.status.in_(["requested", "approved"]))
return Decimal(str(self.conn.execute(q).scalar() or 0))
class CollectionRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _event(m) -> coll.Event:
return coll.Event(m["kind"], m["level"], m["ref_id"], m["invoice_id"] or "", m["occurred_on"], m["promise_date"],
m["status"] or "", m["reason"] or "", m["outcome"] or "")
def _q(self, cond=None) -> list:
q = select(pay_collection)
if cond is not None:
q = q.where(cond)
return list(self.conn.execute(q).mappings())
def events_of(self, bp_id: int) -> list:
return [self._event(m) for m in self._q(pay_collection.c.bp_id == bp_id)]
def all_events(self) -> dict:
out: dict = {}
for m in self._q():
out.setdefault(m["bp_id"], []).append(self._event(m))
return out
def rows_of(self, bp_id: int, kind: Optional[str] = None) -> list:
cond = pay_collection.c.bp_id == bp_id
if kind:
cond = cond & (pay_collection.c.kind == kind)
return [dict(m) for m in self._q(cond)]
def get(self, bp_id: int, kind: str, level: int, ref_id: str) -> Optional[dict]:
rows = self._q((pay_collection.c.bp_id == bp_id) & (pay_collection.c.kind == kind) &
(pay_collection.c.level == level) & (pay_collection.c.ref_id == ref_id))
return dict(rows[0]) if rows else None
def next_level(self, bp_id: int, kind: str) -> int:
return max((m["level"] for m in self._q((pay_collection.c.bp_id == bp_id) & (pay_collection.c.kind == kind))), default=0) + 1
def add(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(pay_collection).values(**row, created_at=_utc(at), updated_at=_utc(at)))
def move(self, bp_id: int, kind: str, level: int, ref_id: str, **columns) -> None:
upsert(self.conn, pay_collection, [{"bp_id": bp_id, "kind": kind, "level": level, "ref_id": ref_id, **columns}],
("bp_id", "kind", "level", "ref_id"))
def drafts(self) -> list:
return [dict(m) for m in self._q((pay_collection.c.kind == "reminder") & (pay_collection.c.status == "draft"))]
class StatementRepository:
def __init__(self, conn):
self.conn = conn
def get(self, bp_id: int, as_at: date, currency: str) -> Optional[dict]:
m = self.conn.execute(select(pay_statement).where(
(pay_statement.c.bp_id == bp_id) & (pay_statement.c.as_at == as_at) &
(pay_statement.c.currency == currency))).mappings().first()
return dict(m) if m else None
def of(self, bp_id: int) -> list:
return [dict(m) for m in self.conn.execute(select(pay_statement).where(pay_statement.c.bp_id == bp_id).order_by(
pay_statement.c.as_at, pay_statement.c.currency)).mappings()]
def add(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(pay_statement).values(**row, created_at=_utc(at), updated_at=_utc(at)))
class CreditRepository:
def __init__(self, conn):
self.conn = conn
def limits(self, bp_id: int) -> list:
q = select(pay_credit).where(pay_credit.c.bp_id == bp_id).order_by(pay_credit.c.valid_from)
return [Limit(m["valid_from"], _dec(m["credit_limit_amount"]), m["currency"]) for m in self.conn.execute(q).mappings()]
def limit_rows(self, bp_id: int) -> list:
q = select(pay_credit).where(pay_credit.c.bp_id == bp_id).order_by(pay_credit.c.valid_from)
return [dict(m) for m in self.conn.execute(q).mappings()]
def has(self, bp_id: int, valid_from: date) -> bool:
return self.conn.execute(select(pay_credit.c.bp_id).where(
(pay_credit.c.bp_id == bp_id) & (pay_credit.c.valid_from == valid_from))).first() is not None
def add_limit(self, row: dict, at: datetime) -> None:
self.conn.execute(insert(pay_credit).values(**row, created_at=_utc(at), updated_at=_utc(at)))
def overrides(self, bp_id: int) -> list:
q = select(pay_credit_override).where(pay_credit_override.c.bp_id == bp_id).order_by(pay_credit_override.c.starts_at)
return [Override(m["kind"], m["starts_at"], m["ends_on"], m["reason"] or "", m["set_by"] or "")
for m in self.conn.execute(q).mappings()]
def has_override(self, bp_id: int, kind: str, starts_at: datetime) -> bool:
return self.conn.execute(select(pay_credit_override.c.bp_id).where(
(pay_credit_override.c.bp_id == bp_id) & (pay_credit_override.c.kind == kind) &
(pay_credit_override.c.starts_at == _utc(starts_at)))).first() is not None
def add_override(self, row: dict, at: datetime) -> None:
row = {**row, "starts_at": _utc(row["starts_at"])}
self.conn.execute(insert(pay_credit_override).values(**row, created_at=_utc(at), updated_at=_utc(at)))
def bp_ids(self) -> list:
return sorted({r[0] for r in self.conn.execute(select(pay_credit.c.bp_id))} |
{r[0] for r in self.conn.execute(select(pay_credit_override.c.bp_id))})
class ConfigRepository:
"""Bank accounts and reminder levels: configuration, keyed, edited by upsert."""
def __init__(self, conn):
self.conn = conn
def banks(self) -> list:
return [dict(m) for m in self.conn.execute(select(pay_bank_account).order_by(pay_bank_account.c.bank_account_id)).mappings()]
def bank(self, bank_account_id: str) -> Optional[dict]:
return next((b for b in self.banks() if b["bank_account_id"] == bank_account_id), None)
def put_bank(self, row: dict) -> None:
upsert(self.conn, pay_bank_account, [row], ("bank_account_id",))
def levels(self, policy_id: str = "default") -> list:
q = select(pay_reminder_level).where(pay_reminder_level.c.policy_id == policy_id).order_by(pay_reminder_level.c.level)
return [coll.Level(m["level"], m["days_after_due"], m["channel"] or "email", m["template"] or "", m["mode"] or "suggest")
for m in self.conn.execute(q).mappings()]
def put_level(self, policy_id: str, lv: coll.Level) -> None:
upsert(self.conn, pay_reminder_level, [{"policy_id": policy_id, "level": lv.level, "days_after_due": lv.days_after_due,
"channel": lv.channel, "template": lv.template, "mode": lv.mode}],
("policy_id", "level"))
def _as_date(v) -> Optional[date]:
"""A max() of a DATE column through UNION ALL: a date on Postgres, an ISO string on SQLite (no column type to convert by)."""
if v is None or isinstance(v, date):
return v.date() if isinstance(v, datetime) else v
return date.fromisoformat(str(v)[:10])
class SettingsRepository:
"""pay_settings: one row per setting (JSON value; NULL = the default). Every change is logged."""
def __init__(self, conn):
self.conn = conn
def _raw(self) -> dict:
"""Every setting in ONE read, once per request (set() and seed() forget it)."""
if self.__dict__.get("_raw_memo") is None:
self.__dict__["_raw_memo"] = {m["setting_key"]: (json.loads(m["value"]) if m["value"] is not None else None)
for m in self.conn.execute(select(pay_settings)).mappings()}
return copy.deepcopy(self.__dict__["_raw_memo"])
def _forget(self) -> None:
self.__dict__["_raw_memo"] = self.__dict__["_zones_memo"] = None
def all(self) -> list:
raw = self._raw()
zones = self._zone_rows()
if zones:
raw["business_timezone"] = zones[-1]["zone"] # what is set (it may start tomorrow)
return [{"setting_key": k, "value": raw.get(k), "default": d[1], "kind": d[0], "controls": d[2], "changes": d[3]}
for k, d in SETTING_DEFS.items()]
def get(self, key: str) -> Optional[dict]:
return next((s for s in self.all() if s["setting_key"] == key), None)
def current(self) -> Settings:
return Settings.from_values({**self._raw(), "business_timezone": self.timezone()})
def _zone_rows(self) -> list:
if self.__dict__.get("_zones_memo") is None: # read once per request (a zone change forgets it)
self.__dict__["_zones_memo"] = [dict(m) for m in self.conn.execute(select(pay_timezone).order_by(pay_timezone.c.valid_from)).mappings()]
return [dict(z) for z in self.__dict__["_zones_memo"]]
def timezone(self) -> str:
"""THE port for business_timezone (shared.yaml conventions): 'today' in every brick is counted in ONE zone per
bucket - the one in force NOW. The zone is a dated control: a 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). Every caller reads the zone here and nowhere else."""
name = self._raw().get("business_timezone") or SETTING_DEFS["business_timezone"][1]
now = work.now()
for row in self._zone_rows():
if row["valid_from"] <= now.astimezone(ZoneInfo(name)).date():
name = row["zone"] or name
else:
break
return name
def used_through(self) -> Optional[date]:
"""The latest business date already stamped on a row of this brick (an allocation, a reversal, an approval): 'today'
is never below it, whatever zone is in force, so nothing is ever dated before something already recorded."""
if not hasattr(self, "_used"): # the three latest dates in ONE statement
q = union_all(*(select(func.max(col).label("d")) for col in
(pay_allocation.c.allocated_on, pay_receipt.c.reversed_on, pay_refund.c.approved_on)))
days = [_as_date(r[0]) for r in self.conn.execute(q)]
self._used = max((d for d in days if d is not None), default=None)
return self._used
def _schedule_zone(self, value, by: str, at: datetime) -> dict:
"""A new zone starts on the NEXT business date (counted in the zone in force now). Changing it again before then
replaces the pending change."""
zone = value or SETTING_DEFS["business_timezone"][1]
self._forget()
start = work.now().astimezone(ZoneInfo(self.timezone())).date() + timedelta(days=1)
upsert(self.conn, pay_timezone, [{"valid_from": start, "zone": zone, "set_by": by, "set_at": _utc(at)}], ("valid_from",))
return {"effective_from": start.isoformat()}
def set(self, key: str, value, by: str, at: datetime) -> dict:
zones = self._zone_rows() if key == "business_timezone" else []
old = zones[-1]["zone"] if zones else self._raw().get(key)
self._forget()
seq = (self.conn.execute(select(func.coalesce(func.max(pay_settings_log.c.seq), 0)).where(
pay_settings_log.c.setting_key == key)).scalar() or 0) + 1
extra = {}
if key == "business_timezone":
extra = self._schedule_zone(value, by, at)
else:
upsert(self.conn, pay_settings, [{"setting_key": key, "value": None if value is None else json.dumps(value),
"changes": SETTING_DEFS[key][3]}], ("setting_key",))
self.conn.execute(insert(pay_settings_log).values(
setting_key=key, seq=seq, old_value=None if old is None else json.dumps(old),
new_value=None if value is None else json.dumps(value), changed_by=by, changed_at=_utc(at),
created_at=_utc(at), updated_at=_utc(at)))
return {"setting_key": key, "value": value, "was": old, **extra}
def history(self, key: Optional[str] = None) -> list:
q = select(pay_settings_log).order_by(pay_settings_log.c.setting_key, pay_settings_log.c.seq)
if key:
q = q.where(pay_settings_log.c.setting_key == key)
return [{"setting_key": m["setting_key"], "seq": m["seq"], "from": json.loads(m["old_value"]) if m["old_value"] else None,
"to": json.loads(m["new_value"]) if m["new_value"] else None, "changed_by": m["changed_by"],
"changed_at": m["changed_at"].isoformat() if m["changed_at"] else None}
for m in self.conn.execute(q).mappings()]
def seed(self) -> None:
have = set(self._raw())
self._forget()
for key, d in SETTING_DEFS.items():
if key not in have:
upsert(self.conn, pay_settings, [{"setting_key": key, "value": None, "changes": d[3]}], ("setting_key",))
class MigrationRepository:
def __init__(self, conn):
self.conn = conn
def mapped(self, refs: Iterable[str]) -> set:
refs = list(refs)
if not refs:
return set()
out = set()
for i in range(0, len(refs), 500):
out |= {r[0] for r in self.conn.execute(select(pay_odoo_map.c.odoo_ref).where(pay_odoo_map.c.odoo_ref.in_(refs[i:i + 500])))}
return out
def targets(self, refs: list) -> dict:
"""odoo_ref -> target_key for the refs this migration mapped."""
out = {}
for i in range(0, len(refs), 500):
q = select(pay_odoo_map.c.odoo_ref, pay_odoo_map.c.target_key).where(pay_odoo_map.c.odoo_ref.in_(refs[i:i + 500]))
out |= {r[0]: r[1] for r in self.conn.execute(q)}
return out
def all_refs(self) -> dict:
return {m["odoo_ref"]: (m["target_table"], m["target_key"]) for m in self.conn.execute(select(pay_odoo_map)).mappings()}
def add(self, ref: str, table: str, key: str, at: datetime) -> None:
self.conn.execute(insert(pay_odoo_map).values(odoo_ref=ref, target_table=table, target_key=key,
migrated_at=_utc(at), created_at=_utc(at), updated_at=_utc(at)))
def run(self) -> Optional[dict]:
m = self.conn.execute(select(pay_migration_run).where(pay_migration_run.c.run_id == "odoo")).mappings().first()
if not m:
return None
return {"params": json.loads(m["params"] or "{}"), "odoo_figures": json.loads(m["odoo_figures"] or "{}"),
"counts": json.loads(m["counts"] or "{}"), "ran_by": m["ran_by"], "signed_by": m["signed_by"] or "",
"signed_at": m["signed_at"].isoformat() if m["signed_at"] else None,
"signed_figures": json.loads(m["signed_figures"]) if m["signed_figures"] else None}
def put_run(self, params: dict, figures: dict, counts: dict, by: str, at: datetime) -> None:
upsert(self.conn, pay_migration_run, [{"run_id": "odoo", "params": _j(params), "odoo_figures": _j(figures),
"counts": _j(counts), "ran_by": by, "ran_at": _utc(at)}], ("run_id",))
def add_attempt(self, params: dict, figures: dict, counts: dict, by: str, at: datetime) -> None:
"""A run after the first: its own row (odoo#2, odoo#3 ...); the first run's record is never rewritten."""
n = self.conn.execute(select(func.count()).select_from(pay_migration_run).where(
pay_migration_run.c.run_id.like("odoo#%"))).scalar() or 0
self.conn.execute(insert(pay_migration_run).values(
run_id=f"odoo#{n + 2}", params=_j(params), odoo_figures=_j(figures), counts=_j(counts), ran_by=by, ran_at=_utc(at),
created_at=_utc(at), updated_at=_utc(at)))
def sign(self, by: str, at: datetime, figures: list) -> None:
upsert(self.conn, pay_migration_run, [{"run_id": "odoo", "signed_by": by, "signed_at": _utc(at),
"signed_figures": _j(figures)}], ("run_id",))
# ═══ v4 slice A repositories ═════════════════════════════════════════════════════════════════════════════
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."""
row = {k: v for k, v in row.items() if k not in ("created_at", "updated_at")} # a row copied from a read carries its old stamps
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()]
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(pay_claim).where(
(pay_claim.c.claim_scope == scope) & (pay_claim.c.claim_key == key))).scalar() or 0
def take(self, scope: str, key: str, at: datetime, by: str = "", ref: str = "") -> int:
n = self.count(scope, key) + 1
_plain(self.conn, pay_claim, {"claim_scope": scope, "claim_key": key, "n": n, "ref": ref, "claimed_by": by, "claimed_at": _utc(at)}, at)
return n
def ref_of(self, scope: str, key: str) -> Optional[str]:
return self.conn.execute(select(pay_claim.c.ref).where((pay_claim.c.claim_scope == scope) & (pay_claim.c.claim_key == key))
.order_by(pay_claim.c.n.desc()).limit(1)).scalar()
class ObjectRepository:
def __init__(self, conn):
self.conn = conn
def put(self, key: str, body: str, content_type: str, sha256: str, at: datetime) -> None:
_plain(self.conn, pay_object, {"object_key": key, "content_type": content_type, "sha256": sha256, "body": body}, at)
def get(self, key: str) -> Optional[str]:
return self.conn.execute(select(pay_object.c.body).where(pay_object.c.object_key == key)).scalar()
class BankRepository:
"""House banks, statements, statement lines, the suggestions on them, and the rules."""
def __init__(self, conn):
self.conn = conn
# house banks
def house_banks(self) -> list:
return _all(self.conn, pay_house_bank, order=(pay_house_bank.c.house_bank_id,))
def house_bank(self, house_bank_id: str) -> Optional[dict]:
return next((h for h in self.house_banks() if h["house_bank_id"] == house_bank_id), None)
def put_house_bank(self, row: dict) -> None:
upsert(self.conn, pay_house_bank, [row], ("house_bank_id",))
# statements
def statement(self, statement_id: str) -> Optional[dict]:
rows = _all(self.conn, pay_bank_statement, pay_bank_statement.c.statement_id == statement_id)
return rows[0] if rows else None
def statements(self, bank_account_id: Optional[str] = None) -> list:
cond = pay_bank_statement.c.bank_account_id == bank_account_id if bank_account_id else None
return _all(self.conn, pay_bank_statement, cond, (pay_bank_statement.c.to_date, pay_bank_statement.c.statement_id))
def add_statement(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_bank_statement, row, at)
def save_statement(self, row: dict) -> None:
"""A statement moves forward (imported -> balanced -> reconciled): the WHOLE row, keyed; callers hold the statement's claim."""
upsert(self.conn, pay_bank_statement, [row], ("statement_id",))
def previous_statement(self, bank_account_id: str, before: date, not_id: str) -> Optional[dict]:
rows = [r for r in self.statements(bank_account_id) if r["statement_id"] != not_id and r["to_date"] is not None and r["to_date"] <= before]
return rows[-1] if rows else None
# lines
def line(self, line_id: str) -> Optional[dict]:
rows = _all(self.conn, pay_statement_line, pay_statement_line.c.line_id == line_id)
return rows[0] if rows else None
def existing_lines(self, ids: Iterable[str]) -> set:
ids = list(ids)
if not ids:
return set()
return {r[0] for r in self.conn.execute(select(pay_statement_line.c.line_id).where(pay_statement_line.c.line_id.in_(ids)))}
def add_line(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_statement_line, row, at)
def save_line(self, row: dict) -> None:
"""A line that moves (open -> matched | posted -> open): the WHOLE row, keyed; callers hold the line's claim."""
upsert(self.conn, pay_statement_line, [row], ("line_id",))
def lines(self, *, statement_id: Optional[str] = None, bank_account_id: Optional[str] = None, lifecycle: Optional[str] = None,
limit: int = 500, offset: int = 0) -> list:
q = select(pay_statement_line).order_by(pay_statement_line.c.booked_on, pay_statement_line.c.line_id)
for col, v in ((pay_statement_line.c.statement_id, statement_id), (pay_statement_line.c.bank_account_id, bank_account_id),
(pay_statement_line.c.lifecycle, lifecycle)):
if v:
q = q.where(col == v)
return [dict(m) for m in self.conn.execute(q.limit(limit).offset(offset)).mappings()]
def posted_by_rule_on(self, day: date) -> list:
return _all(self.conn, pay_statement_line, (pay_statement_line.c.decided_how == "auto") & (pay_statement_line.c.decided_on == day) &
(pay_statement_line.c.lifecycle == "posted"), (pay_statement_line.c.line_id,))
def holder_of(self, kind: str, ref: str) -> Optional[str]:
"""The line that holds this counterpart NOW (matched or posted to it); None when it is free."""
return self.conn.execute(select(pay_statement_line.c.line_id).where(
(pay_statement_line.c.matched_kind == kind) & (pay_statement_line.c.matched_ref == ref) &
(pay_statement_line.c.lifecycle != "open")).limit(1)).scalar()
def posted_history(self, limit: int = 2000) -> list:
return _all(self.conn, pay_statement_line, (pay_statement_line.c.lifecycle == "posted") & (pay_statement_line.c.account_id != ""),
(pay_statement_line.c.booked_on.desc(),))[:limit]
# suggestions on lines
@staticmethod
def _sug(m) -> dict:
d = dict(m)
d["evidence"] = json.loads(d["evidence"]) if d.get("evidence") else {}
return d
def suggestion(self, suggestion_id: str) -> Optional[dict]:
rows = _all(self.conn, pay_line_suggestion, pay_line_suggestion.c.suggestion_id == suggestion_id)
return self._sug(rows[0]) if rows else None
def suggestions_of(self, line_id: str) -> list:
return [self._sug(r) for r in _all(self.conn, pay_line_suggestion, pay_line_suggestion.c.line_id == line_id,
(pay_line_suggestion.c.suggestion_id,))]
def open_suggestions_of(self, line_id: str) -> list:
return [s for s in self.suggestions_of(line_id) if s["status"] == "open"]
def add_suggestion(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_line_suggestion, {**row, "evidence": _j(row.get("evidence") or {})}, at)
def move_suggestion(self, suggestion_id: str, **columns) -> None:
upsert(self.conn, pay_line_suggestion, [{"suggestion_id": suggestion_id, **columns}], ("suggestion_id",))
# rules (dated rows; the effective row of a rule is its latest valid_from <= the day)
def add_rule(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_reconciliation_rule, row, at)
def rule_rows(self) -> list:
return _all(self.conn, pay_reconciliation_rule, order=(pay_reconciliation_rule.c.rule_id, pay_reconciliation_rule.c.valid_from))
def rules_on(self, day: date) -> list:
latest: dict = {}
for r in self.rule_rows():
if r["valid_from"] <= day:
latest[r["rule_id"]] = r
return list(latest.values())
class DirectDebitRepository:
"""Mandates (and their dated bank details), runs and collections."""
def __init__(self, conn):
self.conn = conn
def mandate(self, sun: str, mandate_ref: str) -> Optional[dict]:
rows = _all(self.conn, pay_dd_mandate, (pay_dd_mandate.c.sun == sun) & (pay_dd_mandate.c.mandate_ref == mandate_ref))
return rows[0] if rows else None
def mandates(self, *, sun: Optional[str] = None, lifecycle: Optional[str] = None, bp_id: Optional[int] = None) -> list:
cond = None
for col, v in ((pay_dd_mandate.c.sun, sun), (pay_dd_mandate.c.lifecycle, lifecycle), (pay_dd_mandate.c.bp_id, bp_id)):
if v is not None:
cond = (col == v) if cond is None else cond & (col == v)
return _all(self.conn, pay_dd_mandate, cond, (pay_dd_mandate.c.sun, pay_dd_mandate.c.mandate_ref))
def add_mandate(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_dd_mandate, row, at)
def save_mandate(self, row: dict) -> None:
upsert(self.conn, pay_dd_mandate, [row], ("sun", "mandate_ref"))
def add_details(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_dd_mandate_details, row, at)
def details_at(self, sun: str, mandate_ref: str, on: date) -> Optional[dict]:
rows = [r for r in _all(self.conn, pay_dd_mandate_details, (pay_dd_mandate_details.c.sun == sun) &
(pay_dd_mandate_details.c.mandate_ref == mandate_ref), (pay_dd_mandate_details.c.valid_from,))
if r["valid_from"] <= on]
return rows[-1] if rows else None
def run(self, run_id: str) -> Optional[dict]:
rows = _all(self.conn, pay_dd_run, pay_dd_run.c.run_id == run_id)
return rows[0] if rows else None
def runs(self) -> list:
return _all(self.conn, pay_dd_run, order=(pay_dd_run.c.collection_date, pay_dd_run.c.run_id))
def add_run(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_dd_run, row, at)
def save_run(self, row: dict) -> None:
upsert(self.conn, pay_dd_run, [row], ("run_id",))
def collections(self, run_id: str) -> list:
return _all(self.conn, pay_dd_collection, pay_dd_collection.c.run_id == run_id, (pay_dd_collection.c.mandate_ref,))
def collections_of_mandate(self, sun: str, mandate_ref: str) -> list:
return _all(self.conn, pay_dd_collection, (pay_dd_collection.c.sun == sun) & (pay_dd_collection.c.mandate_ref == mandate_ref))
def add_collection(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_dd_collection, row, at)
def save_collection(self, row: dict) -> None:
upsert(self.conn, pay_dd_collection, [row], ("run_id", "mandate_ref"))
def invoices_in_flight(self) -> dict:
"""invoice_id -> run_id, for every invoice named by a collection that is still to be collected (not deferred, skipped,
collected or returned) in a run that is not finished."""
out = {}
for c in _all(self.conn, pay_dd_collection, pay_dd_collection.c.status == "planned"):
for iid in json.loads(c["invoice_ids"] or "[]"):
out[iid] = c["run_id"]
return out
class RemittanceRepository:
def __init__(self, conn):
self.conn = conn
def lines(self, remittance_id: str) -> list:
return _all(self.conn, pay_remittance, pay_remittance.c.remittance_id == remittance_id, (pay_remittance.c.line_no,))
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_remittance, row, at)
def move(self, remittance_id: str, line_no: int, **columns) -> None:
upsert(self.conn, pay_remittance, [{"remittance_id": remittance_id, "line_no": line_no, **columns}], ("remittance_id", "line_no"))
def of_payment(self, payment_id: str) -> list:
return _all(self.conn, pay_remittance, pay_remittance.c.payment_id == payment_id, (pay_remittance.c.remittance_id, pay_remittance.c.line_no))
class AccountRepository:
"""Default receiving accounts (dated rows), the audit of unmasked reads, and saved forecasts."""
def __init__(self, conn):
self.conn = conn
def defaults(self, company_code: Optional[str] = None, currency: Optional[str] = None) -> list:
cond = None
for col, v in ((pay_default_bank_account.c.company_code, company_code), (pay_default_bank_account.c.currency, currency)):
if v:
cond = (col == v) if cond is None else cond & (col == v)
return _all(self.conn, pay_default_bank_account, cond, (pay_default_bank_account.c.company_code, pay_default_bank_account.c.currency,
pay_default_bank_account.c.valid_from))
def add_default(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_default_bank_account, row, at)
def log_read(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_bank_account_read, row, at)
def reads(self, bank_account_id: Optional[str] = None) -> list:
cond = pay_bank_account_read.c.bank_account_id == bank_account_id if bank_account_id else None
return _all(self.conn, pay_bank_account_read, cond, (pay_bank_account_read.c.requested_at, pay_bank_account_read.c.read_id))
def add_snapshot(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_forecast_snapshot, row, at)
def snapshots(self) -> list:
return _all(self.conn, pay_forecast_snapshot, None, (pay_forecast_snapshot.c.saved_on.desc(), pay_forecast_snapshot.c.snapshot_id))
class CashRepository:
"""Cash journals, their days, entries and counts."""
def __init__(self, conn):
self.conn = conn
def journal(self, cash_journal_id: str) -> Optional[dict]:
rows = _all(self.conn, pay_cash_journal, pay_cash_journal.c.cash_journal_id == cash_journal_id)
return rows[0] if rows else None
def journals(self) -> list:
return _all(self.conn, pay_cash_journal, None, (pay_cash_journal.c.cash_journal_id,))
def put_journal(self, row: dict) -> None:
upsert(self.conn, pay_cash_journal, [row], ("cash_journal_id",))
def day(self, cash_journal_id: str, day: date) -> Optional[dict]:
rows = _all(self.conn, pay_cash_day, (pay_cash_day.c.cash_journal_id == cash_journal_id) & (pay_cash_day.c.day == day))
return rows[0] if rows else None
def days(self, cash_journal_id: str) -> list:
return _all(self.conn, pay_cash_day, pay_cash_day.c.cash_journal_id == cash_journal_id, (pay_cash_day.c.day,))
def add_day(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_cash_day, row, at)
def save_day(self, row: dict) -> None:
"""A day moves forward (open -> closed, an approval): the WHOLE row, keyed; callers hold the day's claim."""
upsert(self.conn, pay_cash_day, [row], ("cash_journal_id", "day"))
def entries(self, cash_journal_id: str, day: Optional[date] = None) -> list:
cond = pay_cash_entry.c.cash_journal_id == cash_journal_id
if day:
cond = cond & (pay_cash_entry.c.day == day)
return _all(self.conn, pay_cash_entry, cond, (pay_cash_entry.c.day, pay_cash_entry.c.seq))
def entry(self, cash_journal_id: str, day: date, seq: int) -> Optional[dict]:
rows = _all(self.conn, pay_cash_entry, (pay_cash_entry.c.cash_journal_id == cash_journal_id) & (pay_cash_entry.c.day == day) &
(pay_cash_entry.c.seq == seq))
return rows[0] if rows else None
def add_entry(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_cash_entry, row, at)
def save_entry(self, row: dict) -> None:
upsert(self.conn, pay_cash_entry, [row], ("cash_journal_id", "day", "seq"))
def counts(self, cash_journal_id: str, day: date) -> list:
return _all(self.conn, pay_cash_count, (pay_cash_count.c.cash_journal_id == cash_journal_id) & (pay_cash_count.c.day == day),
(pay_cash_count.c.seq,))
def add_count(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_cash_count, row, at)
class DepositRepository:
"""Deposit batches of cheques."""
def __init__(self, conn):
self.conn = conn
def get(self, batch_id: str) -> Optional[dict]:
rows = _all(self.conn, pay_deposit_batch, pay_deposit_batch.c.batch_id == batch_id)
return rows[0] if rows else None
def search(self, lifecycle: Optional[str] = None, bank_account_id: Optional[str] = None) -> list:
cond = None
for col, v in ((pay_deposit_batch.c.lifecycle, lifecycle), (pay_deposit_batch.c.bank_account_id, bank_account_id)):
if v:
cond = (col == v) if cond is None else cond & (col == v)
return _all(self.conn, pay_deposit_batch, cond, (pay_deposit_batch.c.batch_id,))
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_deposit_batch, row, at)
def save(self, row: dict) -> None:
upsert(self.conn, pay_deposit_batch, [row], ("batch_id",))
def cheques(self, batch_id: str) -> list:
return [dict(m) for m in self.conn.execute(select(pay_receipt).where(pay_receipt.c.batch_id == batch_id)
.order_by(pay_receipt.c.payment_id)).mappings()]
def live_receipts_of_account(self, bank_account_id: str, after: Optional[date]) -> list:
q = select(pay_receipt).where((pay_receipt.c.bank_account_id == bank_account_id) & (pay_receipt.c.status == "received") &
((pay_receipt.c.statement_line_id == "") | pay_receipt.c.statement_line_id.is_(None)))
if after:
q = q.where(pay_receipt.c.received_on > after)
return [dict(m) for m in self.conn.execute(q).mappings()]
class ExpectedRepository:
"""v6: card settlements (PAY-50) and payouts (PAY-51), keyed on the handing module's own reference."""
def __init__(self, conn):
self.conn = conn
def card(self, settlement_ref: str) -> Optional[dict]:
rows = _all(self.conn, pay_card_settlement, pay_card_settlement.c.settlement_ref == settlement_ref)
return rows[0] if rows else None
def cards(self, lifecycle: Optional[str] = None) -> list:
return _all(self.conn, pay_card_settlement, (pay_card_settlement.c.lifecycle == lifecycle) if lifecycle else None,
(pay_card_settlement.c.settlement_ref,))
def add_card(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_card_settlement, row, at)
def save_card(self, row: dict) -> None:
upsert(self.conn, pay_card_settlement, [row], ("settlement_ref",))
def payout(self, payout_ref: str) -> Optional[dict]:
rows = _all(self.conn, pay_payout, pay_payout.c.payout_ref == payout_ref)
return rows[0] if rows else None
def payouts(self, lifecycle: Optional[str] = None) -> list:
return _all(self.conn, pay_payout, (pay_payout.c.lifecycle == lifecycle) if lifecycle else None, (pay_payout.c.payout_ref,))
def add_payout(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_payout, row, at)
def save_payout(self, row: dict) -> None:
upsert(self.conn, pay_payout, [row], ("payout_ref",))
def unannounced(self) -> list:
"""Paid payouts whose source module has not yet acknowledged them."""
return _all(self.conn, pay_payout, (pay_payout.c.lifecycle == "paid") & pay_payout.c.told_source_at.is_(None), (pay_payout.c.payout_ref,))
class FxRepository:
"""The FX difference posted per allocation row (PAY-33)."""
def __init__(self, conn):
self.conn = conn
def get(self, payment_id: str, invoice_id: str, seq: int) -> Optional[dict]:
rows = _all(self.conn, pay_fx_posting, (pay_fx_posting.c.payment_id == payment_id) & (pay_fx_posting.c.invoice_id == invoice_id)
& (pay_fx_posting.c.seq == seq))
return rows[0] if rows else None
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_fx_posting, row, at)
def for_payment(self, payment_id: str) -> list:
return _all(self.conn, pay_fx_posting, pay_fx_posting.c.payment_id == payment_id,
(pay_fx_posting.c.invoice_id, pay_fx_posting.c.seq))
class FeedRepository:
"""Open-banking feed consents (PAY-35)."""
def __init__(self, conn):
self.conn = conn
def consents(self, bank_account_id: Optional[str] = None) -> list:
cond = pay_feed_consent.c.bank_account_id == bank_account_id if bank_account_id else None
return _all(self.conn, pay_feed_consent, cond, (pay_feed_consent.c.bank_account_id, pay_feed_consent.c.granted_on))
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, pay_feed_consent, row, at)
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.receipts = ReceiptRepository(conn)
self.allocations = AllocationRepository(conn)
self.suggestions = SuggestionRepository(conn)
self.refunds = RefundRepository(conn)
self.collections = CollectionRepository(conn)
self.statements = StatementRepository(conn)
self.credit = CreditRepository(conn)
self.config = ConfigRepository(conn)
self.settings = SettingsRepository(conn)
self.migration = MigrationRepository(conn)
self.claims = ClaimRepository(conn)
self.objects = ObjectRepository(conn)
self.bank = BankRepository(conn)
self.dd = DirectDebitRepository(conn)
self.remittances = RemittanceRepository(conn)
self.accounts = AccountRepository(conn)
self.cash = CashRepository(conn)
self.deposits = DepositRepository(conn)
self.expected = ExpectedRepository(conn)
self.fx = FxRepository(conn)
self.feeds = FeedRepository(conn)
def seed(self) -> None:
self.settings.seed()