dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/infrastructure/tables.py
"""The tables, and one repository per aggregate that loads and saves it whole.
Every table is prefixed `tax_`, has a natural primary key (never SERIAL) and, for a part, a key that starts with its root's.
A rate is NUMERIC(19,4), money NUMERIC(18,2) (shared.yaml). A dated row (a rate, a rule, a registration, a customer's
treatment, a setting) is keyed on its start date and only ever added, so any past date's answer is reproducible.
schema.sql is written from the same column model; tests/test_schema.py keeps the two equal.
"""
import dataclasses
from datetime import date, datetime
from decimal import Decimal
from typing import Iterable, Optional
from erp_platform.repository import Aggregate, Part, upsert
from sqlalchemy import TIMESTAMP, BigInteger, Boolean, Column, Date, Integer, MetaData, Numeric, String, Table, select
from brick_tax.domain import settings_book
from brick_tax.domain.document_tax import CodeFacts, DocLine, DocumentTax, Suggestion
from brick_tax.domain.registration import Registration
from brick_tax.domain.rounding import Computed, LineResult, SummaryRow
from brick_tax.domain.rules import Rule
from brick_tax.domain.tax_code import Rate, TaxCode
from brick_tax.domain.values import ChangeState, DocState, encode_value
md = MetaData()
RATE = Numeric(19, 4)
MONEY = Numeric(18, 2)
FX = Numeric(19, 8)
tax_registration = Table("tax_registration", md,
Column("jurisdiction", String, primary_key=True),
Column("effective_from", Date, primary_key=True),
Column("status", String),
Column("vat_number", String),
Column("scheme", String),
Column("effective_to", Date),
Column("change_state", String),
Column("created_by", String),
Column("approved_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("created_sub", String), # 00002: the author's sub beside their email
Column("accounting_basis", String), Column("flat_rate_sector", String), Column("flat_rate_pct", RATE), # 00003: v4 schemes
Column("first_year_until", Date), Column("ni_goods", Boolean), Column("partial_exemption_year_end", String),
Column("expected_turnover", MONEY))
TAX_REGISTRATION_COLUMNS = ("jurisdiction", "effective_from", "status", "vat_number", "scheme", "effective_to",
"change_state", "created_by", "approved_by", "created_sub", "accounting_basis", "flat_rate_sector",
"flat_rate_pct", "first_year_until", "ni_goods", "partial_exemption_year_end", "expected_turnover")
tax_code = Table("tax_code", md,
Column("tax_code", String, primary_key=True),
Column("jurisdiction", String),
Column("treatment", String),
Column("description", String),
Column("document_wording", String),
Column("reporting_box", String),
Column("lifecycle", String),
Column("used", Boolean),
Column("version", Integer),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("direction", String), # 00003 (v3/v4): sale | purchase
Column("input_reporting_box", String),
Column("value_box", String),
Column("reclaimable_pct", RATE),
Column("attribution", String), Column("eu_goods", Boolean)) # 00003: partial exemption, NI goods
TAX_CODE_COLUMNS = ("tax_code", "jurisdiction", "treatment", "description", "document_wording", "reporting_box", "lifecycle", "used",
"direction", "input_reporting_box", "value_box", "reclaimable_pct", "attribution", "eu_goods")
tax_rate = Table("tax_rate", md,
Column("tax_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("rate_pct", RATE),
Column("charged_pct", RATE),
Column("change_state", String),
Column("created_by", String),
Column("approved_by", String),
Column("withdrawn_by", String),
Column("withdrawn_at", TIMESTAMP),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("created_sub", String)) # 00002
TAX_RATE_COLUMNS = ("tax_code", "valid_from", "rate_pct", "charged_pct", "change_state", "created_by", "approved_by",
"withdrawn_by", "withdrawn_at", "created_sub")
TAX_CODE = Aggregate(Part(tax_code, ("tax_code",)), Part(tax_rate, ("tax_code", "valid_from")), version="version")
tax_category = Table("tax_category", md,
Column("category_code", String, primary_key=True),
Column("description", String),
Column("created_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
tax_treatment = Table("tax_treatment", md,
Column("treatment_code", String, primary_key=True),
Column("description", String),
Column("needs_vat_number", Boolean),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
# One customer's treatment and VAT number FROM A DATE. The first row holds from the start (FOREVER): no document could
# issue without one; a change is a new row from tomorrow.
tax_party = Table("tax_party", md,
Column("bp_id", BigInteger, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("treatment_code", String),
Column("vat_number", String),
Column("checked_on", Date),
Column("check_result", String),
Column("recorded_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("check_name", String), Column("check_address", String), Column("consultation_number", String), # 00003: the HMRC check
Column("check_flag", String), Column("confirmed_by", String), Column("confirm_reason", String),
Column("notice_kind", String), Column("notice_ref", String), Column("notice_date", Date))
def _rule_table(name: str) -> Table:
return Table(name, md,
Column("category_code", String, primary_key=True),
Column("treatment_code", String, primary_key=True),
Column("supply", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("tax_code", String),
Column("priority", Integer),
Column("valid_to", Date),
Column("change_state", String),
Column("created_by", String),
Column("approved_by", String),
Column("withdrawn_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("created_sub", String)) # 00002
tax_rule = _rule_table("tax_rule") # the sale-direction rules
# The purchase-direction rules (v3): the same columns in a table of their own, because tax_rule's primary key cannot gain
# `direction` additively (K3 changes no key in place) — a purchase pair and a sale pair with one scope never collide.
tax_rule_purchase = _rule_table("tax_rule_purchase")
# v6 (TAX-51): a rule that tells eat_in from takeaway, in a table of its own for the same reason: consumption is part of the key and
# no key changes in place. Rules of consumption `any` (every earlier row) stay in the two tables above.
tax_rule_consumption = Table("tax_rule_consumption", md,
Column("category_code", String, primary_key=True),
Column("treatment_code", String, primary_key=True),
Column("supply", String, primary_key=True),
Column("direction", String, primary_key=True),
Column("consumption", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("tax_code", String),
Column("priority", Integer),
Column("valid_to", Date),
Column("change_state", String),
Column("created_by", String),
Column("approved_by", String),
Column("withdrawn_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("created_sub", String))
TAX_RULE_COLUMNS = ("category_code", "treatment_code", "supply", "valid_from", "tax_code", "priority", "valid_to",
"change_state", "created_by", "approved_by", "withdrawn_by", "created_sub")
tax_settings = Table("tax_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))
tax_document = Table("tax_document", md,
Column("document_ref", String, primary_key=True),
Column("bp_id", BigInteger),
Column("tax_point", Date),
Column("currency", String),
Column("treatment_code", String),
Column("status", String),
Column("rounding_method", String),
Column("prices_include_tax", Boolean),
Column("registered", Boolean),
Column("vat_number", String),
Column("title", String),
Column("fx_rate", FX),
Column("vat_total_gbp", MONEY),
Column("frozen_by", String),
Column("frozen_at", TIMESTAMP),
Column("version", Integer),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("direction", String), # 00003: sale | purchase
Column("ship_to_country", String), Column("self_billed", Boolean), Column("due_date", Date))
TAX_DOCUMENT_COLUMNS = ("document_ref", "bp_id", "tax_point", "currency", "treatment_code", "status", "rounding_method",
"prices_include_tax", "registered", "vat_number", "title", "fx_rate", "vat_total_gbp", "frozen_by",
"frozen_at", "direction", "ship_to_country", "self_billed", "due_date")
tax_document_line = Table("tax_document_line", md,
Column("document_ref", String, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("description", String),
Column("category_code", String),
Column("supply", String),
Column("amount", MONEY),
Column("quantity", RATE),
Column("unit_price", RATE),
Column("tax_code", String),
Column("treatment", String),
Column("rate_pct", RATE),
Column("charged_pct", RATE),
Column("wording", String),
Column("reporting_box", String),
Column("source", String),
Column("rule_ref", String),
Column("rule_code", String),
Column("refusal", String),
Column("override_reason", String),
Column("override_by", String),
Column("override_state", String),
Column("override_approved_by", String),
Column("override_approval_reason", String),
Column("lowers_tax", Boolean),
Column("suggestion_state", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("override_by_sub", String), # 00002: the overrider's sub beside their email
Column("calc_net", MONEY), Column("calc_vat", MONEY), # 00002: the line's tax as issued, stored at freeze
Column("calc_vat_due", MONEY), Column("calc_gross", MONEY),
Column("reclaimable_pct", RATE), # 00003: a purchase line's code, as it stood
Column("input_reporting_box", String), Column("value_box", String),
Column("pinned_from", String), Column("attribution", String), Column("eu_goods", Boolean), # 00003: v4
Column("consumption", String)) # 00004: v6 (TAX-51)
TAX_LINE_COLUMNS = ("line_no", "description", "category_code", "supply", "amount", "quantity", "unit_price", "tax_code",
"treatment", "rate_pct", "charged_pct", "wording", "reporting_box", "source", "rule_ref", "rule_code",
"refusal", "override_reason", "override_by", "override_state", "override_approved_by",
"override_approval_reason", "lowers_tax", "suggestion_state", "override_by_sub", "calc_net", "calc_vat",
"calc_vat_due", "calc_gross", "reclaimable_pct", "input_reporting_box", "value_box", "pinned_from", "attribution",
"eu_goods", "consumption")
# The per-rate summary of an ISSUED document, stored at freeze: what the customer received is never recomputed (00002).
tax_document_summary = Table("tax_document_summary", md,
Column("document_ref", String, primary_key=True),
Column("tax_code", String, primary_key=True),
Column("rate_pct", RATE),
Column("charged_pct", RATE),
Column("net", MONEY), Column("vat", MONEY), Column("vat_due", MONEY), Column("gross", MONEY),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
TAX_SUMMARY_COLUMNS = ("tax_code", "rate_pct", "charged_pct", "net", "vat", "vat_due", "gross")
# Every AI suggestion, kept for the accountant: its inputs, the model, the output, the reason, who decided and when.
tax_document_suggestion = Table("tax_document_suggestion", md,
Column("document_ref", String, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("tax_code", String),
Column("reason", String),
Column("confidence", Integer),
Column("model", String),
Column("inputs", String),
Column("state", String),
Column("created_by", String),
Column("decided_by", String),
Column("decided_at", TIMESTAMP),
Column("decision_reason", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
TAX_SUGGESTION_COLUMNS = ("line_no", "seq", "tax_code", "reason", "confidence", "model", "inputs", "state", "created_by",
"decided_by", "decided_at", "decision_reason")
DOCUMENT = Aggregate(Part(tax_document, ("document_ref",)), Part(tax_document_line, ("document_ref", "line_no")),
Part(tax_document_summary, ("document_ref", "tax_code")),
Part(tax_document_suggestion, ("document_ref", "line_no", "seq")), version="version")
# ── v4 (00003): HMRC, the VAT return, bad debt relief ─────────────────────────────────────────────────────────────
# The figures the law sets, from a date (cash accounting limits, the error-correction thresholds ...): read on the date they apply to.
tax_statutory = Table("tax_statutory", md,
Column("stat_key", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", String),
Column("unit", String),
Column("source", String),
Column("set_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
# The connection to HMRC for one VRN. The tokens are NOT here: secret_ref names the platform secret that holds them.
tax_hmrc_connection = Table("tax_hmrc_connection", md,
Column("vrn", String, primary_key=True),
Column("scopes", String),
Column("environment", String),
Column("authorised_by", String),
Column("authorised_at", TIMESTAMP),
Column("expires_at", TIMESTAMP),
Column("secret_ref", String),
Column("status", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("sync_error", String), Column("sync_error_at", TIMESTAMP)) # HMRC's error code from the last sync, retried next day
# A copy of HMRC's obligation (dates are HMRC's, never computed here).
tax_vat_obligation = Table("tax_vat_obligation", md,
Column("vrn", String, primary_key=True),
Column("period_key", String, primary_key=True),
Column("period_start", Date),
Column("period_end", Date),
Column("due_date", Date),
Column("status", String),
Column("received_on", Date),
Column("synced_at", TIMESTAMP),
Column("hmrc_error", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
tax_vat_return = Table("tax_vat_return", md,
Column("vrn", String, primary_key=True),
Column("period_key", String, primary_key=True),
Column("lifecycle", String),
Column("accounting_basis", String),
Column("flat_rate_pct", RATE),
Column("box1_vat_due_sales", MONEY), Column("box2_vat_due_acquisitions", MONEY), Column("box3_total_vat_due", MONEY),
Column("box4_vat_reclaimed", MONEY), Column("box5_net_vat_due", MONEY), Column("box6_total_sales_ex_vat", MONEY),
Column("box7_total_purchases_ex_vat", MONEY), Column("box8_goods_supplied_ex_vat", MONEY),
Column("box9_acquisitions_ex_vat", MONEY),
Column("prepared_by", String), Column("locked_by", String), Column("locked_by_sub", String), Column("locked_at", TIMESTAMP),
Column("submitted_by", String), Column("submitted_at", TIMESTAMP), Column("declaration_text", String),
Column("processing_date", String), Column("form_bundle_number", String), Column("payment_indicator", String),
Column("charge_ref_number", String), Column("fraud_headers_json", String), Column("settle_journal_id", String),
Column("hmrc_error", String),
Column("version", Integer),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
VAT_RETURN = Aggregate(Part(tax_vat_return, ("vrn", "period_key")), version="version")
# One row per (document, code, kind) CLAIMED by a return: a plain INSERT, so a document's tax enters one return only.
tax_vat_return_entry = Table("tax_vat_return_entry", md,
Column("document_ref", String, primary_key=True),
Column("tax_code", String, primary_key=True),
Column("entry_kind", String, primary_key=True),
Column("vrn", String),
Column("period_key", String),
Column("direction", String),
Column("tax_point", Date),
Column("net_amount", MONEY),
Column("vat_amount", MONEY),
Column("vat_reclaimable", MONEY),
Column("box1_amount", MONEY),
Column("box4_amount", MONEY),
Column("box_value", String),
Column("source", String),
Column("detail", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("gross_amount", MONEY), Column("attribution", String), Column("box2_amount", MONEY))
tax_vat_adjustment = Table("tax_vat_adjustment", md,
Column("vrn", String, primary_key=True),
Column("period_key", String, primary_key=True),
Column("adjustment_no", Integer, primary_key=True),
Column("corrects_period_key", String),
Column("corrects_period_end", Date),
Column("box", Integer),
Column("amount", MONEY),
Column("reason", String),
Column("recorded_by", String),
Column("recorded_at", TIMESTAMP),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
tax_vat_error_notice = Table("tax_vat_error_notice", md,
Column("vrn", String, primary_key=True),
Column("period_key", String, primary_key=True),
Column("notice_no", Integer, primary_key=True),
Column("reference", String),
Column("sent_on", Date),
Column("periods", String),
Column("amount", MONEY),
Column("reason", String),
Column("recorded_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
tax_vat_liability = Table("tax_vat_liability", md,
Column("vrn", String, primary_key=True),
Column("period_key", String, primary_key=True),
Column("charge_type", String, primary_key=True),
Column("original_amount", MONEY),
Column("outstanding_amount", MONEY),
Column("due_date", Date),
Column("paid_amount", MONEY),
Column("paid_on", Date),
Column("synced_at", TIMESTAMP),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
tax_bad_debt_claim = Table("tax_bad_debt_claim", md,
Column("invoice_id", String, primary_key=True),
Column("lifecycle", String),
Column("bp_id", BigInteger),
Column("writeoff_id", String),
Column("supply_date", Date),
Column("due_date", Date),
Column("eligible_from", Date),
Column("written_off_on", Date),
Column("gross", MONEY),
Column("vat", MONEY),
Column("unpaid_at_claim", MONEY),
Column("vat_fraction", FX),
Column("relief_amount", MONEY),
Column("claimed_in_period_key", String),
Column("claimed_by", String),
Column("claimed_on", Date),
Column("dismissed_reason", String),
Column("version", Integer),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
BAD_DEBT = Aggregate(Part(tax_bad_debt_claim, ("invoice_id",)), version="version")
# (odoo db, odoo model, odoo id) -> our natural key, written by tax:history:import (shared.yaml conventions.odoo_migration).
tax_migration_map = Table("tax_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("our_table", String),
Column("our_key", String),
Column("created_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
def _dec(v) -> Optional[Decimal]:
return None if v is None else Decimal(str(v))
def _d(v) -> Optional[date]:
return v.date() if isinstance(v, datetime) else v
class RegistrationRepository:
def __init__(self, conn):
self.conn = conn
def rows(self) -> list:
"""Every registration, read once per request (put() forgets them)."""
if self.__dict__.get("_rows") is None:
self.__dict__["_rows"] = self._read()
return [dataclasses.replace(r) for r in self.__dict__["_rows"]]
def _read(self) -> list:
stmt = select(tax_registration).order_by(tax_registration.c.effective_from)
return [Registration(r["jurisdiction"], _d(r["effective_from"]), r["status"], r["vat_number"] or "", r["scheme"] or "standard",
_d(r["effective_to"]), r["change_state"] or ChangeState.APPROVED.value, r["created_by"] or "",
r["approved_by"] or "", r["created_sub"] or "", r["accounting_basis"] or "invoice",
r["flat_rate_sector"] or "", _dec(r["flat_rate_pct"]), _d(r["first_year_until"]), bool(r["ni_goods"]),
r["partial_exemption_year_end"] or "", _dec(r["expected_turnover"])) for r in (x._mapping for x in self.conn.execute(stmt))]
def put(self, row: Registration) -> None:
self.__dict__["_rows"] = None
upsert(self.conn, tax_registration, [{c: getattr(row, c) for c in TAX_REGISTRATION_COLUMNS}],
("jurisdiction", "effective_from"))
class CodeRepository:
"""TaxCode loads and saves whole, with its rates."""
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
@staticmethod
def _build(root: dict, rates: Iterable[dict]) -> TaxCode:
c = TaxCode(root["tax_code"], root["jurisdiction"] or "GB", root["treatment"], root["description"] or "",
root["document_wording"] or "", root["reporting_box"] or "", root["lifecycle"], bool(root["used"]))
c.direction, c.input_reporting_box = root.get("direction") or "sale", root.get("input_reporting_box") or ""
c.value_box, c.reclaimable_pct = root.get("value_box") or "", _dec(root.get("reclaimable_pct"))
c.attribution, c.eu_goods = root.get("attribution") or "", bool(root.get("eu_goods"))
c.rates = [Rate(r["tax_code"], _d(r["valid_from"]), _dec(r["rate_pct"]), _dec(r["charged_pct"]), r["change_state"],
r["created_by"] or "", r["approved_by"] or "", r["withdrawn_by"] or "", r["withdrawn_at"],
r["created_sub"] or "")
for r in sorted(rates, key=lambda r: r["valid_from"])]
return c
def get(self, code: str) -> Optional[tuple]:
rows = TAX_CODE.load(self.conn, {"tax_code": code})
if rows is None:
return None
self._snapshots[code] = rows.snapshot
return self._build(rows["tax_code"][0], rows["tax_rate"]), TAX_CODE.version_of(rows)
def save(self, agg: TaxCode, expected_version: int) -> int:
snapshot = self._snapshots.pop(agg.tax_code, None) # spent: after this save the rows are no longer what it read
return TAX_CODE.save(self.conn, {
"tax_code": [{**{c: getattr(agg, c) for c in TAX_CODE_COLUMNS}}],
"tax_rate": [{c: getattr(r, c) for c in TAX_RATE_COLUMNS} for r in agg.rates]}, expected_version=expected_version, snapshot=snapshot)
def all(self) -> list:
roots = [dict(r._mapping) for r in self.conn.execute(select(tax_code).order_by(tax_code.c.tax_code))]
rates: dict = {}
for r in self.conn.execute(select(tax_rate)):
rates.setdefault(r._mapping["tax_code"], []).append(dict(r._mapping))
return [self._build(root, rates.get(root["tax_code"], [])) for root in roots]
def facts_on(self, on: date) -> dict:
"""code -> CodeFacts on a tax point: its meaning and the rate valid then."""
out = {}
for c in self.all():
r = c.rate_on(on)
out[c.tax_code] = CodeFacts(c.tax_code, c.treatment, c.lifecycle, c.description, c.document_wording, c.reporting_box,
r.rate_pct if r else None, r.charged_pct if r else None, c.direction, c.input_reporting_box,
c.value_box, c.reclaimable_pct, c.attribution, c.eu_goods)
return out
class CategoryRepository:
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
return [{"category_code": r["category_code"], "description": r["description"] or ""}
for r in (x._mapping for x in self.conn.execute(select(tax_category).order_by(tax_category.c.category_code)))]
def get(self, code: str) -> Optional[dict]:
return next((c for c in self.all() if c["category_code"] == code), None)
def put(self, code: str, description: str, by: str) -> None:
upsert(self.conn, tax_category, [{"category_code": code, "description": description, "created_by": by}], ("category_code",))
class TreatmentRepository:
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
stmt = select(tax_treatment).order_by(tax_treatment.c.treatment_code)
return [{"treatment_code": r["treatment_code"], "description": r["description"] or "",
"needs_vat_number": bool(r["needs_vat_number"])} for r in (x._mapping for x in self.conn.execute(stmt))]
def get(self, code: str) -> Optional[dict]:
return next((t for t in self.all() if t["treatment_code"] == code), None)
def put(self, code: str, description: str, needs_vat_number: bool) -> None:
upsert(self.conn, tax_treatment, [{"treatment_code": code, "description": description,
"needs_vat_number": needs_vat_number}], ("treatment_code",))
class PartyRepository:
def __init__(self, conn):
self.conn = conn
def rows_of(self, bp_id: int) -> list:
stmt = select(tax_party).where(tax_party.c.bp_id == bp_id).order_by(tax_party.c.valid_from)
return [{**dict(r._mapping), "valid_from": _d(r._mapping["valid_from"])} for r in self.conn.execute(stmt)]
def effective(self, bp_id: int, on: date) -> Optional[dict]:
found = [r for r in self.rows_of(bp_id) if r["valid_from"] <= on]
return found[-1] if found else None
def put(self, bp_id: int, valid_from: date, treatment_code: str, vat_number: str, by: str, **more) -> None:
upsert(self.conn, tax_party, [{"bp_id": bp_id, "valid_from": valid_from, "treatment_code": treatment_code,
"vat_number": vat_number, "recorded_by": by, **more}], ("bp_id", "valid_from"))
def update(self, bp_id: int, valid_from: date, **fields) -> None:
self.conn.execute(tax_party.update().where(tax_party.c.bp_id == bp_id, tax_party.c.valid_from == valid_from).values(**fields))
def count_with_vat(self) -> int:
return sum(1 for r in self.conn.execute(select(tax_party.c.bp_id, tax_party.c.vat_number))
if r._mapping["vat_number"])
def uses_treatment(self, treatment_code: str) -> bool:
return any(True for _ in self.conn.execute(select(tax_party.c.bp_id).where(tax_party.c.treatment_code == treatment_code)))
def all_current(self, on: date) -> dict:
latest: dict = {}
for r in self.conn.execute(select(tax_party)):
m = dict(r._mapping)
m["valid_from"] = _d(m["valid_from"])
if m["valid_from"] <= on and (m["bp_id"] not in latest or m["valid_from"] > latest[m["bp_id"]]["valid_from"]):
latest[m["bp_id"]] = m
return latest
class RuleRepository:
"""Sale rules live in tax_rule, purchase rules in tax_rule_purchase; a Rule carries its direction."""
def __init__(self, conn):
self.conn = conn
def all(self, direction: Optional[str] = "sale") -> list:
"""The rules of one direction (default sale: what v2 callers meant); None = both."""
out = []
for d, t in (("sale", tax_rule), ("purchase", tax_rule_purchase)):
if direction not in (None, d):
continue
stmt = select(t).order_by(t.c.category_code, t.c.treatment_code, t.c.supply, t.c.valid_from)
out += [Rule(r["category_code"], r["treatment_code"], r["supply"], _d(r["valid_from"]), r["tax_code"] or "",
r["priority"] or 0, _d(r["valid_to"]), r["change_state"] or ChangeState.APPROVED.value,
r["created_by"] or "", r["approved_by"] or "", r["withdrawn_by"] or "", r["created_sub"] or "", d)
for r in (x._mapping for x in self.conn.execute(stmt))]
stmt = select(tax_rule_consumption).order_by(tax_rule_consumption.c.category_code, tax_rule_consumption.c.treatment_code,
tax_rule_consumption.c.supply, tax_rule_consumption.c.valid_from)
out += [Rule(r["category_code"], r["treatment_code"], r["supply"], _d(r["valid_from"]), r["tax_code"] or "",
r["priority"] or 0, _d(r["valid_to"]), r["change_state"] or ChangeState.APPROVED.value,
r["created_by"] or "", r["approved_by"] or "", r["withdrawn_by"] or "", r["created_sub"] or "", r["direction"], r["consumption"])
for r in (x._mapping for x in self.conn.execute(stmt)) if direction in (None, r["direction"])]
return out
def put(self, rule: Rule) -> None:
if rule.consumption != "any":
upsert(self.conn, tax_rule_consumption, [{**{c: getattr(rule, c) for c in TAX_RULE_COLUMNS}, "direction": rule.direction,
"consumption": rule.consumption}],
("category_code", "treatment_code", "supply", "direction", "consumption", "valid_from"))
return
upsert(self.conn, tax_rule_purchase if rule.direction == "purchase" else tax_rule,
[{c: getattr(rule, c) for c in TAX_RULE_COLUMNS}], ("category_code", "treatment_code", "supply", "valid_from"))
class SettingsRepository:
def __init__(self, conn):
self.conn = conn
def rows(self) -> list:
"""Every setting row, in one read once per request (put() forgets them)."""
if self.__dict__.get("_rows") is None:
stmt = select(tax_settings).order_by(tax_settings.c.setting_key, tax_settings.c.valid_from)
self.__dict__["_rows"] = [settings_book.from_stored(r["setting_key"], _d(r["valid_from"]), r["value"], r["set_by"] or "")
for r in (x._mapping for x in self.conn.execute(stmt))]
return [dataclasses.replace(r) for r in self.__dict__["_rows"]]
def put(self, key: str, valid_from: date, value, by: str) -> None:
self.__dict__["_rows"] = None
upsert(self.conn, tax_settings, [{"setting_key": key, "valid_from": valid_from, "value": encode_value(value),
"set_by": by}], ("setting_key", "valid_from"))
class DocumentRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, document_ref: str) -> Optional[tuple]:
rows = DOCUMENT.load(self.conn, {"document_ref": document_ref})
if rows is None: # nothing under the key: a save of it writes without reading the keys again (its version claim refuses a racer)
self._snapshots[document_ref] = DOCUMENT.nothing_stored({"document_ref": document_ref})
return None
self._snapshots[document_ref] = rows.snapshot
return self._build(rows)
def get_many(self, refs) -> dict:
"""{document_ref: (DocumentTax, version)} for a set of documents: one query per table per 100 documents, in one round trip."""
out = {}
for (ref,), rows in DOCUMENT.load_many(self.conn, [{"document_ref": r} for r in dict.fromkeys(refs)]).items():
if rows is None:
continue
self._snapshots[ref] = rows.snapshot
out[ref] = self._build(rows)
return out
def _build(self, rows) -> tuple:
h = rows["tax_document"][0]
doc = DocumentTax(h["document_ref"], h["bp_id"], _d(h["tax_point"]), h["currency"], h["treatment_code"] or "", h["status"],
h["rounding_method"], bool(h["prices_include_tax"]), bool(h["registered"]), h["vat_number"] or "",
h["title"] or "", _dec(h["fx_rate"]), _dec(h["vat_total_gbp"]), h["frozen_by"] or "", h["frozen_at"],
h["direction"] or "sale")
doc.ship_to_country, doc.self_billed, doc.due_date = h["ship_to_country"] or "", bool(h["self_billed"]), _d(h["due_date"])
for r in sorted(rows["tax_document_line"], key=lambda r: r["line_no"]):
doc.lines.append(DocLine(
r["line_no"], r["description"] or "", r["category_code"] or "", r["supply"] or "any", _dec(r["amount"]) or Decimal(0),
_dec(r["quantity"]), _dec(r["unit_price"]), r["tax_code"] or "", r["treatment"] or "", _dec(r["rate_pct"]),
_dec(r["charged_pct"]), r["wording"] or "", r["reporting_box"] or "", r["source"] or "", r["rule_ref"] or "",
r["rule_code"] or "", r["refusal"] or "", r["override_reason"] or "", r["override_by"] or "",
r["override_state"] or "none", r["override_approved_by"] or "", r["override_approval_reason"] or "",
bool(r["lowers_tax"]), r["suggestion_state"] or "", r["override_by_sub"] or "", _dec(r["reclaimable_pct"]),
r["input_reporting_box"] or "", r["value_box"] or "", r["pinned_from"] or "", r["attribution"] or "",
bool(r["eu_goods"]), r["consumption"] or ""))
doc.sealed = self._sealed(doc, rows)
for r in sorted(rows["tax_document_suggestion"], key=lambda r: (r["line_no"], r["seq"])):
doc.suggestions.append(Suggestion(r["line_no"], r["seq"], r["tax_code"], r["reason"] or "", r["confidence"], r["model"] or "",
r["inputs"] or "", r["state"], r["created_by"] or "", r["decided_by"] or "",
r["decided_at"], r["decision_reason"] or ""))
return doc, DOCUMENT.version_of(rows)
@staticmethod
def _sealed(doc: DocumentTax, rows: dict) -> Optional[Computed]:
"""The tax as issued, read back as stored: None for a document not frozen, or frozen before it was stored."""
stored = rows.get("tax_document_summary") or []
if doc.status != DocState.FROZEN.value or not stored or any(r["calc_vat"] is None for r in rows["tax_document_line"]):
return None
lines = {r["line_no"]: LineResult(r["line_no"], _dec(r["calc_net"]), _dec(r["calc_vat"]), _dec(r["calc_vat_due"]),
_dec(r["calc_gross"])) for r in rows["tax_document_line"]}
summary = [SummaryRow(r["tax_code"], _dec(r["rate_pct"]), _dec(r["charged_pct"]), _dec(r["net"]), _dec(r["vat"]),
_dec(r["vat_due"]), _dec(r["gross"])) for r in sorted(stored, key=lambda r: r["tax_code"])]
net, vat = sum((x.net for x in summary), Decimal(0)), sum((x.vat for x in summary), Decimal(0))
return Computed(lines, summary, net, vat, sum((x.vat_due for x in summary), Decimal(0)), net + vat, doc.rounding_method)
def save(self, doc: DocumentTax, expected_version: int) -> int:
snapshot = self._snapshots.pop(doc.document_ref, None) # spent: after this save the rows are no longer what it read
ref = doc.document_ref
sealed = doc.sealed if doc.status == DocState.FROZEN.value else None
def line_row(ln):
got = sealed.lines.get(ln.line_no) if sealed else None
calc = {"calc_net": got and got.net, "calc_vat": got and got.vat, "calc_vat_due": got and got.vat_due,
"calc_gross": got and got.gross}
return {"document_ref": ref, **{c: getattr(ln, c) for c in TAX_LINE_COLUMNS if c not in calc}, **calc}
return DOCUMENT.save(self.conn, {
"tax_document": [{c: getattr(doc, c) for c in TAX_DOCUMENT_COLUMNS}],
"tax_document_line": [line_row(ln) for ln in doc.lines],
"tax_document_summary": [{"document_ref": ref, **{c: getattr(x, "code" if c == "tax_code" else c) for c in TAX_SUMMARY_COLUMNS}}
for x in (sealed.summary if sealed else [])],
"tax_document_suggestion": [{"document_ref": ref, **{c: getattr(s, c) for c in TAX_SUGGESTION_COLUMNS}}
for s in doc.suggestions]}, expected_version=expected_version, snapshot=snapshot)
def refs(self) -> list:
return [r[0] for r in self.conn.execute(select(tax_document.c.document_ref))]
class MigrationMapRepository:
def __init__(self, conn):
self.conn = conn
def get(self, odoo_db: str, model: str, odoo_id) -> Optional[dict]:
stmt = select(tax_migration_map).where(tax_migration_map.c.odoo_db == odoo_db, tax_migration_map.c.odoo_model == model,
tax_migration_map.c.odoo_id == str(odoo_id))
row = self.conn.execute(stmt).first()
return dict(row._mapping) if row else None
def put(self, odoo_db: str, model: str, odoo_id, our_table: str, our_key: str, by: str) -> None:
upsert(self.conn, tax_migration_map, [{"odoo_db": odoo_db, "odoo_model": model, "odoo_id": str(odoo_id),
"our_table": our_table, "our_key": our_key, "created_by": by}],
("odoo_db", "odoo_model", "odoo_id"))
def all(self, odoo_db: str) -> list:
stmt = select(tax_migration_map).where(tax_migration_map.c.odoo_db == odoo_db)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
# ── v4 slice B (00003): CIS, trade declarations, import VAT, partial exemption, capital goods, AI findings ───────────────
def _t(name, *cols):
return Table(name, md, *cols, Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
def _s(n, pk=False):
return Column(n, String, primary_key=pk)
# A subcontractor's identity (the row at FOREVER, status 'unverified') and each verification from a date (a new row, never an edit).
tax_cis_subcontractor = _t("tax_cis_subcontractor", Column("bp_id", BigInteger, primary_key=True), Column("valid_from", Date, primary_key=True),
_s("name"), _s("utr"), _s("company_number"), _s("ni_number"), _s("status"), Column("rate_pct", RATE), _s("verification_number"),
_s("verified_source"), _s("recorded_by"))
# One deduction per [payment_ref, bp_id]: a retry returns the same row; a reversal marks it (refused once its month is submitted).
tax_cis_deduction = _t("tax_cis_deduction", _s("payment_ref", True), Column("bp_id", BigInteger, primary_key=True), _s("tax_month"),
Column("paid_on", Date), _s("bill_ids"), Column("gross_ex_vat", MONEY), Column("materials", MONEY), Column("labour", MONEY),
_s("status"), Column("rate_pct", RATE), Column("deduction", MONEY), _s("state"), Column("reversed_on", Date), _s("reversed_reason"))
tax_cis_return = _t("tax_cis_return", _s("tax_month", True), _s("lifecycle"), Column("nil_return", Boolean), Column("total_deduction", MONEY),
_s("lines_json"), _s("prepared_by"), _s("submitted_by"), Column("submitted_at", TIMESTAMP), _s("receipt_json"),
Column("statements_sent_at", TIMESTAMP), _s("statements_json"), _s("declaration_text"))
tax_cis_suffered = _t("tax_cis_suffered", Column("contractor_bp_id", BigInteger, primary_key=True), _s("statement_ref", True), _s("tax_month"),
_s("invoice_id"), _s("payment_id"), Column("gross", MONEY), Column("materials", MONEY), Column("deduction", MONEY))
tax_trade_declaration = _t("tax_trade_declaration", _s("kind", True), _s("period", True), Column("version", Integer, primary_key=True),
_s("lifecycle"), Column("total", MONEY), _s("file_text"), _s("prepared_by"), _s("submission_ref"), Column("filed_on", Date),
_s("filed_by"), _s("gates_json"))
tax_trade_declaration_line = _t("tax_trade_declaration_line", _s("kind", True), _s("period", True), Column("version", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True), _s("partner_vat_number"), _s("partner_country"), _s("commodity_code"), _s("nature"),
_s("origin_country"), Column("value", MONEY), Column("net_mass_kg", RATE), Column("supplementary_units", RATE), _s("indicator"))
tax_import_entry = _t("tax_import_entry", _s("mrn", True), Column("entry_date", Date), _s("bill_ref"), Column("bp_id", BigInteger),
Column("customs_value", MONEY), Column("import_vat", MONEY), _s("method"), _s("tax_code"), Column("box_vat_output", MONEY),
Column("box_vat_input", MONEY), Column("box_value", MONEY), Column("certificate_listed", Boolean), _s("recorded_by"))
tax_pva_statement = _t("tax_pva_statement", _s("month", True), _s("lifecycle"), _s("file_text"), _s("imported_by"), _s("reconciled_by"))
tax_pva_statement_line = _t("tax_pva_statement_line", _s("month", True), _s("mrn", True), Column("statement_vat", MONEY),
Column("our_vat", MONEY), _s("status"), Column("difference", MONEY), _s("explanation"), Column("use_hmrc_figure", Boolean), _s("explained_by"))
tax_partial_exemption = _t("tax_partial_exemption", _s("vrn", True), Column("year_end", Date, primary_key=True), Column("provisional_pct", RATE),
Column("final_pct", RATE), Column("taxable_supplies", MONEY), Column("exempt_supplies", MONEY), Column("exempt_input_vat", MONEY),
Column("residual_input_vat", MONEY), Column("de_minimis_met", Boolean), Column("adjustment_amount", MONEY), _s("adjusted_in_period_key"))
tax_capital_goods = _t("tax_capital_goods", _s("item_ref", True), _s("kind"), Column("bought_on", Date), Column("net_value", MONEY),
Column("input_vat", MONEY), Column("intervals", Integer), Column("first_pct", RATE), _s("adjustments_json"), Column("disposed_on", Date),
_s("recorded_by"))
# Every finding of the AI review of a draft return, kept with its evidence, the model, a digest of the inputs and the decision.
tax_ai_finding = _t("tax_ai_finding", _s("vrn", True), _s("period_key", True), Column("finding_no", Integer, primary_key=True), _s("evidence"),
Column("box", Integer), _s("model"), _s("inputs_digest"), _s("decision"), _s("decided_by"), Column("decided_at", TIMESTAMP), _s("note"))
class _Rows:
"""Keyed rows of one table as dicts: get, list, put (an idempotent keyed upsert)."""
def __init__(self, conn, table: Table):
self.conn, self.table = conn, table
self.keys = tuple(c.name for c in table.primary_key.columns)
def _row(self, r) -> dict:
m = dict(r._mapping)
return {k: (_d(v) if isinstance(self.table.c[k].type, Date) else v) for k, v in m.items()}
def get(self, **key) -> Optional[dict]:
rows = self.where(**key)
return rows[0] if rows else None
def where(self, **eq) -> list:
stmt = select(self.table)
for k, v in eq.items():
stmt = stmt.where(self.table.c[k] == v)
return [self._row(r) for r in self.conn.execute(stmt.order_by(*(self.table.c[k] for k in self.keys)))]
def put(self, row: dict) -> None:
upsert(self.conn, self.table, [row], self.keys)
def remove(self, **eq) -> None:
stmt = self.table.delete()
for k, v in eq.items():
stmt = stmt.where(self.table.c[k] == v)
self.conn.execute(stmt)
class EntryRepository(_Rows):
"""The entries a return CLAIMS: a plain INSERT keyed [document_ref, tax_code, entry_kind], so a document's tax enters one
return only — a second claim (a concurrent preparation) fails the key and is told so."""
def claim(self, row: dict) -> bool:
"""Check, then a plain INSERT in the SAME transaction (no savepoint, no delete-then-insert: on the DataK3 pg wire a failed
INSERT aborts the transaction). True: the row is this return's now (inserted, or this return's own refreshed in place).
False: ANOTHER return already holds the key — the document is in a return. A racer that passed the same check collides on
the primary key at commit and surfaces as a Conflict (409 -> the retry sees the other return's row and skips)."""
from erp_platform.repository import Conflict
from sqlalchemy.exc import IntegrityError
key = {k: row[k] for k in self.keys}
have = self.get(**key)
if have is not None:
if (have["vrn"], have["period_key"]) != (row["vrn"], row["period_key"]):
return False
stmt = self.table.update()
for k, v in key.items():
stmt = stmt.where(self.table.c[k] == v)
self.conn.execute(stmt.values(**{k: v for k, v in row.items() if k not in self.keys}))
return True
try:
self.conn.execute(self.table.insert().values(**row))
except IntegrityError:
raise Conflict(self.table.name, key) from None
return True
def claim_many(self, rows: list) -> list:
"""claim() for a whole batch in a constant number of statements: ONE read of the existing keys (document_ref IN ...), then one
bulk INSERT for the new keys and one keyed upsert for this return's own (refreshed in place). Returns one bool per row, as claim()."""
from erp_platform.repository import Conflict, upsert
from sqlalchemy.exc import IntegrityError
if not rows:
return []
refs = sorted({r["document_ref"] for r in rows})
have = {tuple(r[k] for k in self.keys): r for r in
(self._row(x) for x in self.conn.execute(select(self.table).where(self.table.c.document_ref.in_(refs))))}
out, fresh, own = [], [], []
for row in rows:
h = have.get(tuple(row[k] for k in self.keys))
if h is None:
fresh.append(row)
out.append(True)
elif (h["vrn"], h["period_key"]) != (row["vrn"], row["period_key"]):
out.append(False)
else:
own.append(row)
out.append(True)
if own:
upsert(self.conn, self.table, own, self.keys, insert_only=())
groups: dict = {}
for row in fresh:
groups.setdefault(tuple(sorted(row)), []).append(row)
try:
for g in groups.values():
self.conn.execute(self.table.insert(), g)
except IntegrityError:
raise Conflict(self.table.name, {"document_ref": refs[0]}) from None
return out
def sweep(self, vrn: str, period_key: str, keep: set, protect: tuple = ("bad_debt_claim",)) -> int:
"""Drop this return's entries a refresh no longer produces (keys NOT re-inserted in the same transaction)."""
n = 0
for r in self.where(vrn=vrn, period_key=period_key):
if r["entry_kind"] in protect or (r["document_ref"], r["tax_code"], r["entry_kind"]) in keep:
continue
self.remove(document_ref=r["document_ref"], tax_code=r["tax_code"], entry_kind=r["entry_kind"])
n += 1
return n
class ReturnRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, vrn: str, period_key: str) -> Optional[tuple]:
from brick_tax.domain.vat_return import VatReturn
rows = VAT_RETURN.load(self.conn, {"vrn": vrn, "period_key": period_key})
if rows is None:
return None
self._snapshots[(vrn, period_key)] = rows.snapshot
r = rows["tax_vat_return"][0]
return VatReturn.from_row(r), VAT_RETURN.version_of(rows)
def save(self, ret, expected_version: int) -> int:
snapshot = self._snapshots.pop((ret.vrn, ret.period_key), None) # spent: after this save the rows are no longer what it read
return VAT_RETURN.save(self.conn, {"tax_vat_return": [ret.to_row()]}, expected_version=expected_version, snapshot=snapshot)
def all(self, vrn: Optional[str] = None) -> list:
from brick_tax.domain.vat_return import VatReturn
stmt = select(tax_vat_return).order_by(tax_vat_return.c.period_key)
if vrn:
stmt = stmt.where(tax_vat_return.c.vrn == vrn)
return [VatReturn.from_row(dict(r._mapping)) for r in self.conn.execute(stmt)]
class BadDebtRepository:
def __init__(self, conn):
self.conn = conn
self._snapshots: dict = {} # root key -> what its load read: save() hands it back so only changed rows are written
def get(self, invoice_id: str) -> Optional[tuple]:
from brick_tax.domain.bad_debt import Claim
rows = BAD_DEBT.load(self.conn, {"invoice_id": invoice_id})
if rows is None:
return None
self._snapshots[invoice_id] = rows.snapshot
return Claim.from_row(rows["tax_bad_debt_claim"][0]), BAD_DEBT.version_of(rows)
def save(self, claim, expected_version: int) -> int:
snapshot = self._snapshots.pop(claim.invoice_id, None) # spent: after this save the rows are no longer what it read
return BAD_DEBT.save(self.conn, {"tax_bad_debt_claim": [claim.to_row()]}, expected_version=expected_version, snapshot=snapshot)
def all(self) -> list:
from brick_tax.domain.bad_debt import Claim
return [Claim.from_row(dict(r._mapping)) for r in self.conn.execute(select(tax_bad_debt_claim).order_by(tax_bad_debt_claim.c.invoice_id))]
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.registrations = RegistrationRepository(conn)
self.codes = CodeRepository(conn)
self.categories = CategoryRepository(conn)
self.treatments = TreatmentRepository(conn)
self.parties = PartyRepository(conn)
self.rules = RuleRepository(conn)
self.settings = SettingsRepository(conn)
self.documents = DocumentRepository(conn)
self.migration_map = MigrationMapRepository(conn)
self.statutory = _Rows(conn, tax_statutory)
self.connections = _Rows(conn, tax_hmrc_connection)
self.obligations = _Rows(conn, tax_vat_obligation)
self.returns = ReturnRepository(conn)
self.entries = EntryRepository(conn, tax_vat_return_entry)
self.adjustments = _Rows(conn, tax_vat_adjustment)
self.notices = _Rows(conn, tax_vat_error_notice)
self.liabilities = _Rows(conn, tax_vat_liability)
self.claims = BadDebtRepository(conn)
self.cis_subs = _Rows(conn, tax_cis_subcontractor)
self.cis_deductions = _Rows(conn, tax_cis_deduction)
self.cis_returns = _Rows(conn, tax_cis_return)
self.cis_suffered = _Rows(conn, tax_cis_suffered)
self.trade_declarations = _Rows(conn, tax_trade_declaration)
self.trade_lines = _Rows(conn, tax_trade_declaration_line)
self.imports = _Rows(conn, tax_import_entry)
self.pva_statements = _Rows(conn, tax_pva_statement)
self.pva_lines = _Rows(conn, tax_pva_statement_line)
self.pe_years = _Rows(conn, tax_partial_exemption)
self.capital_items = _Rows(conn, tax_capital_goods)
self.findings = _Rows(conn, tax_ai_finding)
def seed(self) -> None:
"""A new bucket starts with no rows: the UK pack is loaded by tax:code:seed (load_pack), the settings stand at
their defaults until a dated row says otherwise."""