dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/use_cases.py
"""The use cases — what a person (or their agent, or a selling brick) can do about VAT.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere else: WHO MAY
(a PERMISSION on the declaration, never a role's name) and WHICH ROWS (`_sees_*`; a row the caller may not see is a 404,
never a 403). Each command is one unit of work: load, let the domain decide, save whole at the version it was loaded.
How the tax is worked out (module.yaml): a product carries a tax CATEGORY and a customer a TREATMENT; a dated RULE turns the
pair into a tax CODE and the code's RATE on the TAX POINT is what is charged. Rates, rules, treatments, registrations and
settings are dated rows: a change starts after today (in business_timezone), a correction is a new row, never an edit; only a
key's first row may carry an earlier date. Tax is computed here and frozen on the document at issue.
"""
import json
from contextlib import contextmanager
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal
from typing import Annotated, Literal, Optional, Union
from erp_platform import auth
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import retry_on_conflict
from erp_platform.types import BigId, IsoDate, Name, Note, text
from pydantic import Field
from sqlalchemy.exc import IntegrityError
from brick_tax.application import odoo_migration
from brick_tax.application.inputs import (
ContentsIn,
Consumption,
Currency,
Fx,
Key,
LineIn,
Money,
OdooDoorIn,
Pct,
Reason,
Supply,
day,
)
from brick_tax.domain import registration, rules, schedule, settings_book, uk_pack, vat_invoice
from brick_tax.domain.document_tax import CodeFacts, Context, DocumentTax
from brick_tax.domain.document_tax import LineIn as DomainLine
from brick_tax.domain.errors import GateUnmet, RuleBroken, unmet
from brick_tax.domain.rules import Rule
from brick_tax.domain.tax_code import Rate, TaxCode
from brick_tax.domain.values import (
CONTROL_KEYS,
FOREVER,
JURISDICTION,
KEY,
ROUND_DOWN_METHODS,
SETTING_DEFS,
SETTINGS,
ChangeState,
CodeState,
DocState,
RegStatus,
Treatment,
check_setting,
)
from brick_tax.infrastructure import ai_steps, odoo, seams
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared. Which people hold which
# is the composition's role catalogue — roles.yaml beside this package is the recommended default.
READ, CODE_WRITE, APPROVE = "tax:code:read", "tax:code:write", "tax:code:approve"
CATEGORY_WRITE, REGISTRATION_WRITE, PARTY_WRITE = "tax:category:write", "tax:registration:write", "tax:party_tax:write"
OVERRIDE, APPROVE_OVERRIDE = "tax:document_tax:override", "tax:document_tax:approve_override"
SETTINGS_WRITE, CONTROLS = "tax:settings:write", "tax:settings:controls"
SEED, COMPUTE, SUGGEST = "tax:code:seed", "tax:document_tax:compute", "tax:document_tax:suggest"
IMPORT = "tax:history:import" # machine: the one-off Odoo migration — first rows only, never an issued document's tax
HMRC_CONNECT, RETURN_READ, RETURN_PREPARE, RETURN_SUBMIT = "tax:hmrc:connect", "tax:vat_return:read", "tax:vat_return:prepare", "tax:vat_return:submit"
ADJUST, BAD_DEBT_CLAIM = "tax:vat_adjustment:write", "tax:bad_debt:claim"
OBLIGATION_SYNC, TAX_POINT_READ, BAD_DEBT_SUGGEST = "tax:obligation:sync", "tax:tax_point:read", "tax:bad_debt:suggest" # machine
CIS_WRITE, CIS_SUBMIT, TRADE_WRITE, IMPORT_WRITE, PE_WRITE = ("tax:cis:write", "tax:cis_return:submit", "tax:trade_declaration:write",
"tax:import_vat:write", "tax:partial_exemption:write")
CIS_READ, CIS_DEDUCT, CIS_SUFFERED, IMPORT_RECORD, VAT_NUMBER_CHECK, MIGRATION_READ = (
"tax:cis:read", "tax:cis:deduct", "tax:cis:record_suffered", "tax:import_vat:record", "tax:vat_number:check", "tax:migration:read") # machine
TAX = Brick("tax", version="0.2.2", ids_as_strings=True, machine_permissions=(SEED, COMPUTE, SUGGEST, IMPORT, OBLIGATION_SYNC, TAX_POINT_READ, BAD_DEBT_SUGGEST, CIS_READ, CIS_DEDUCT,
CIS_SUFFERED, IMPORT_RECORD, VAT_NUMBER_CHECK), instructions=(
"Tax (UK VAT first): the right VAT on every quote, order and invoice without anyone choosing a rate. A product carries a "
"tax CATEGORY and a customer a TREATMENT; a dated rule turns the pair into a tax CODE (S, R, Z, E, OS, RC, NV) and the "
"code's rate on the TAX POINT is charged. Selling documents ask compute_tax for their tax and store the answer; it is "
"frozen at issue (freeze_document_tax) and never recomputed. Every refused step returns its unmet gates as {gate, met, "
"reason, fix}. A pair no rule covers is refused, never taxed at 0%. Rates, rules, registrations and settings are dated "
"rows: a change starts tomorrow at the earliest and a correction is a new row. The AI only ever suggests a code; a "
"person accepts it."))
# The process view (BRICKS.md): the signed workflows as BPMN, GET /process/bpmn.
TAX.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text_ = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text_ or "duplicate key" in text_ or "UNIQUE constraint" in text_
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a write conflict on a claim row can surface at COMMIT (the loser's COMMIT raises 23505), after the use case
has returned - so it is caught here, around the commit, and answered as the same retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_tax.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
repository = _repository_default # tests and a composition swap in their own
@contextmanager
def _decide():
"""A broken rule is a 409 with the rule's own words (and, for a gate, every unmet gate), through both doors."""
try:
yield
except GateUnmet as g:
raise Refused(409, g.text()) from None
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
def _atomic(unit):
"""One unit of work on a fresh transaction, re-run on the latest if another writer saved first."""
def once():
with repository() as repo, _decide():
return unit(repo)
return retry_on_conflict(once)
def _who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _sub(caller: dict) -> str:
return (caller.get("sub") or "").strip().lower()
def _ids(caller: dict) -> tuple:
"""Every identity the caller has (email and sub, lower-case): the same person under either is the same person."""
return tuple(dict.fromkeys(x.strip().lower() for x in (caller.get("email"), caller.get("sub")) if x and x.strip()))
def _today(repo) -> date:
"""Business today: the date NOW (work.now(): a demo may move the clock) in business_timezone — "after today" means after
this date, everywhere a rule says it. It never goes backwards across a zone switch (seams.business_today)."""
return seams.business_today(repo.settings.rows())
def _now() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None)
def _date(v: Optional[str], what: str) -> Optional[date]:
try:
return day(v)
except ValueError:
raise Refused(422, f"{what} is a date like 2026-11-01") from None
def _settings_on(repo, on: date):
return settings_book.at(repo.settings.rows(), on)
def _settings_now(repo):
return _settings_on(repo, _today(repo))
def _pct(d: Optional[Decimal]) -> Optional[str]:
"""A percentage without trailing zeros: 20, 5, 0, 12.5."""
return None if d is None else format(d.normalize(), "f") if d != 0 else "0"
def _sees_drafts(caller: dict) -> bool:
return any(auth.has_permission(caller, p) for p in (CODE_WRITE, SETTINGS_WRITE, SEED, COMPUTE, IMPORT))
# ── how things read ───────────────────────────────────────────────────────────────────────────
def _rate_view(r: Rate) -> dict:
return {"valid_from": r.valid_from, "rate_pct": _pct(r.rate_pct), "charged_pct": _pct(r.charged_pct),
"change_state": r.change_state, "created_by": r.created_by, "approved_by": r.approved_by or None,
"withdrawn_by": r.withdrawn_by or None}
def _code_view(c: TaxCode, version: int, today: date, detail: bool = True) -> dict:
now = c.rate_on(today)
out = {"tax_code": c.tax_code, "jurisdiction": c.jurisdiction, "treatment": c.treatment, "description": c.description,
"document_wording": c.document_wording, "reporting_box": c.reporting_box, "lifecycle": c.lifecycle, "used": c.used,
"rate_pct": _pct(now.rate_pct) if now else None, "charged_pct": _pct(now.charged_pct) if now else None,
"direction": c.direction, "version": version}
if c.direction == "purchase":
out |= {"input_reporting_box": c.input_reporting_box or None, "value_box": c.value_box or None,
"reclaimable_pct": _pct(c.reclaimable_pct), "attribution": c.attribution or None}
if c.eu_goods:
out["eu_goods"] = True
if detail:
out["rates"] = [_rate_view(r) for r in c.live_rates()]
out["readiness"] = {"activate_code": [g.as_dict() for g in c.activation_gates()]} if c.lifecycle == CodeState.DRAFT.value else {}
return out
def _reg_view(r: registration.Registration) -> dict:
return {"jurisdiction": r.jurisdiction, "effective_from": r.effective_from, "status": r.status,
"vat_number": r.vat_number or None, "scheme": r.scheme, "change_state": r.change_state,
"created_by": r.created_by, "approved_by": r.approved_by or None, "accounting_basis": r.accounting_basis,
"flat_rate_sector": r.flat_rate_sector or None, "flat_rate_pct": _pct(r.flat_rate_pct), "first_year_until": r.first_year_until,
"ni_goods": r.ni_goods, "partial_exemption_year_end": r.partial_exemption_year_end or None,
"expected_turnover": r.expected_turnover}
def _rule_view(r: Rule) -> dict:
return {"category_code": r.category_code, "treatment_code": r.treatment_code, "supply": r.supply, "consumption": r.consumption, "valid_from": r.valid_from,
"tax_code": r.tax_code or None, "ended": not r.tax_code, "priority": r.priority, "change_state": r.change_state,
"created_by": r.created_by, "approved_by": r.approved_by or None}
def _line_view(ln, result, split=None) -> dict:
out = {"line_no": ln.line_no, "description": ln.description, "category_code": ln.category_code or None, "supply": ln.supply,
"amount": ln.amount, "tax_code": ln.tax_code or None, "treatment": ln.treatment or None,
"rate_pct": _pct(ln.rate_pct), "charged_pct": _pct(ln.charged_pct), "source": ln.source or None,
"rule": ln.rule_ref or None, "wording": ln.wording or None, "reporting_box": ln.reporting_box or None,
"refusal": ln.refusal or None, "suggestion_state": ln.suggestion_state or None, "pinned_from": ln.pinned_from or None,
"consumption": ln.consumption or None}
if result is not None:
r = result.lines[ln.line_no]
out |= {"net": r.net, "vat": r.vat, "vat_due": r.vat_due, "gross": r.gross}
if split is not None:
out |= {"vat_reclaimable": split[0][ln.line_no][0], "vat_non_reclaimable": split[0][ln.line_no][1]}
if ln.override_state != "none":
out["override"] = {"reason": ln.override_reason, "by": ln.override_by, "lowers_tax": ln.lowers_tax,
"state": ln.override_state, "approved_by": ln.override_approved_by or None,
"approval_reason": ln.override_approval_reason or None, "rule_code": ln.rule_code or None}
return out
def _doc_view(doc: DocumentTax, version: int) -> dict:
result = doc.computed()
gates = doc.gates()
split = doc.purchase_split()
out = {"document_ref": doc.document_ref, "status": doc.status, "direction": doc.direction, "bp_id": doc.bp_id, "tax_point": doc.tax_point,
"currency": doc.currency, "treatment_code": doc.treatment_code or None, "registered": doc.registered,
"title": doc.title, "vat_number": doc.vat_number or None, "rounding_method": doc.rounding_method,
"prices_include_tax": doc.prices_include_tax,
"lines": [_line_view(ln, result, split) for ln in doc.lines], "complete": result is not None,
"gates": [g.as_dict() for g in gates], "version": version}
out["suggestions"] = [{"line_no": s.line_no, "seq": s.seq, "tax_code": s.tax_code, "reason": s.reason, "confidence": s.confidence,
"model": s.model, "state": s.state, "decided_by": s.decided_by or None, "decided_at": s.decided_at}
for s in doc.suggestions]
if result is None:
out |= {"summary": None, "totals": None}
return out
kinds = {ln.tax_code: ln.treatment for ln in doc.lines}
wording = {ln.tax_code: ln.wording for ln in doc.lines}
out["summary"] = [{"tax_code": s.code, "treatment": kinds[s.code], "rate_pct": _pct(s.rate_pct), "charged_pct": _pct(s.charged_pct),
"net": s.net, "vat": s.vat, "vat_due": s.vat_due, "gross": s.gross, "wording": wording[s.code] or None}
for s in result.summary]
sep = {k: sum((s.net for s in result.summary if kinds[s.code] == k), Decimal(0))
for k in (Treatment.ZERO.value, Treatment.EXEMPT.value, Treatment.OUTSIDE_SCOPE.value)}
out["totals"] = {"net": result.net, "vat": result.vat, "vat_due": result.vat_due, "gross": result.gross,
"net_zero_rated": sep["zero"], "net_exempt": sep["exempt"], "net_outside_scope": sep["outside_scope"],
"vat_total_gbp": doc.vat_total_gbp, "fx_rate": doc.fx_rate, "method": result.method}
if split is not None:
out["totals"] |= {"vat_reclaimable": split[2][0], "vat_non_reclaimable": split[2][1]}
if doc.status == DocState.FROZEN.value:
out["frozen_by"], out["frozen_at"] = doc.frozen_by, doc.frozen_at
return out
# ═══ registration (TAX-01) ═══════════════════════════════════════════════════════════════════
@TAX.use_case(kind="query", route="GET /registration", permission=READ)
def get_registration(caller: dict) -> dict:
"""Whether, and from when, this business is VAT registered: every dated registration row (a change is a new row, never an
edit) and the status today. Before the registration date documents charge no VAT and are not titled "VAT invoice"."""
with repository() as repo:
today = _today(repo)
rows = repo.registrations.rows()
now = registration.status_on(rows, today)
return {"today": today, "registrations": [_reg_view(r) for r in rows],
"status_today": now.status if now else None, "vat_number": (now.vat_number or None) if now else None,
"recorded": bool(registration.live(rows))}
def _put_registration(repo, row: registration.Registration, author: Optional[dict] = None) -> dict:
if author is not None: # a NEW row: remember its author's sub beside their email
row.created_sub = _sub(author)
repo.registrations.put(row)
return _reg_view(row)
@TAX.use_case(kind="command", route="POST /registration/record-not-registered", permission=REGISTRATION_WRITE)
def record_not_registered(caller: dict, effective_from: Optional[IsoDate] = None) -> dict:
"""Record that the business is NOT VAT registered — the first registration row of a new installation. It may carry any date
(nothing has used it; default today). Such a business charges no VAT and its documents may not be titled "VAT invoice"."""
with repository() as repo, _decide():
row = registration.record_first(repo.registrations.rows(), RegStatus.NOT_REGISTERED.value, None,
_date(effective_from, "effective_from"), today=_today(repo), scheme=None,
by=_who(caller), jurisdiction=JURISDICTION)
return _put_registration(repo, row, caller)
@TAX.use_case(kind="command", route="POST /registration/record-registered", permission=REGISTRATION_WRITE)
def record_registered(caller: dict, vat_number: text(32), effective_from: IsoDate,
scheme: Optional[Literal["standard", "flat_rate", "cash_accounting"]] = None,
flat_rate_pct: Optional[Pct] = None, flat_rate_sector: Optional[text(64)] = None) -> dict:
"""Record an EXISTING VAT registration — the first registration row of a new installation or the Odoo migration. It carries
the date HMRC registered the business (it may be in the past: nothing has used the row). The number is GB and 9 or 12 digits.
A flat rate registration carries its percentage (flat_rate_pct); cash accounting prepares on the cash basis."""
if scheme == "flat_rate" and flat_rate_pct is None:
raise Refused(422, "the flat rate scheme needs flat_rate_pct, the sector's percentage")
with repository() as repo, _decide():
row = registration.record_first(repo.registrations.rows(), RegStatus.REGISTERED.value, vat_number,
_date(effective_from, "effective_from"), today=_today(repo), scheme=scheme,
by=_who(caller), jurisdiction=JURISDICTION, flat_rate_pct=flat_rate_pct,
flat_rate_sector=flat_rate_sector or "")
return _put_registration(repo, row, caller)
@TAX.use_case(kind="command", route="POST /registration/register", permission=REGISTRATION_WRITE)
def register(caller: dict, vat_number: text(32), effective_from: Optional[IsoDate] = None,
scheme: Optional[Literal["standard", "flat_rate", "cash_accounting"]] = None,
flat_rate_pct: Optional[Pct] = None, flat_rate_sector: Optional[text(64)] = None) -> dict:
"""Record that the business becomes VAT registered from a date AFTER today (default tomorrow). It needs the VAT number
(GB and 9 or 12 digits). Tax points before the date charge no VAT; from it the rules apply and the number prints on every
VAT invoice. Where the owner's control asks for it, the change waits for a second holder of tax:code:approve."""
if scheme == "flat_rate" and flat_rate_pct is None:
raise Refused(422, "the flat rate scheme needs flat_rate_pct, the sector's percentage")
with repository() as repo, _decide():
settings = _settings_now(repo)
row = registration.register(repo.registrations.rows(), vat_number, _date(effective_from, "effective_from"),
today=_today(repo), second_person=settings.tax_changes_need_second_person, scheme=scheme,
by=_who(caller), jurisdiction=JURISDICTION, flat_rate_pct=flat_rate_pct,
flat_rate_sector=flat_rate_sector or "")
return _put_registration(repo, row, caller)
@TAX.use_case(kind="command", route="POST /registration/deregister", permission=REGISTRATION_WRITE)
def deregister(caller: dict, effective_from: Optional[IsoDate] = None) -> dict:
"""Record deregistration from a date AFTER today. It always waits for a second holder of tax:code:approve (never its
author). Tax points from the date charge no VAT."""
with repository() as repo, _decide():
row = registration.deregister(repo.registrations.rows(), _date(effective_from, "effective_from"), today=_today(repo),
by=_who(caller), jurisdiction=JURISDICTION)
return _put_registration(repo, row, caller)
@TAX.use_case(kind="command", route="POST /registration/change-scheme", permission=REGISTRATION_WRITE)
def change_vat_scheme(caller: dict, scheme: Literal["standard", "flat_rate", "cash_accounting"], effective_from: IsoDate,
expected_turnover: Optional[Money] = None, flat_rate_sector: Optional[text(120, min_length=1)] = None,
first_year_until: Optional[IsoDate] = None, ni_goods: Optional[bool] = None,
partial_exemption_year_end: Optional[text(5)] = None) -> dict:
"""Change the VAT scheme from the FIRST DAY OF A VAT PERIOD after today (HMRC's periods, synced first): cash accounting (the
expected taxable turnover within the joining limit), the flat rate scheme (within its limit, a sector from the dated statutory
rows; `first_year_until` carries the first-year 1% discount) or back to standard. A new registration row; returns of later
periods compute under it. Where the owner's control asks, the change waits for a second holder of tax:code:approve.
`ni_goods` marks an XI registration trading goods with the EU; `partial_exemption_year_end` (MM-DD) turns partial exemption on."""
with repository() as repo, _decide():
start = _date(effective_from, "effective_from")
starts = {o["period_start"] for o in repo.obligations.where()}
join, pct = None, None
if scheme == "cash_accounting":
join = Decimal(repo.statutory.get(stat_key="cash_join_limit")["value"])
elif scheme == "flat_rate":
join = Decimal(repo.statutory.get(stat_key="flat_join_limit")["value"])
sector = repo.statutory.get(stat_key=f"flat_sector:{flat_rate_sector or ''}")
if sector is None:
raise unmet("flat_rate_sector", f"'{flat_rate_sector}' is not a flat rate sector in the statutory rows", "Tax settings > Statutory figures")
pct = Decimal(sector["value"])
if partial_exemption_year_end is not None and partial_exemption_year_end:
try:
date.fromisoformat(f"2000-{partial_exemption_year_end}")
except ValueError:
raise Refused(422, "partial_exemption_year_end is MM-DD, such as 03-31") from None
row = registration.change_scheme(
repo.registrations.rows(), scheme=scheme, effective_from=start, today=_today(repo),
second_person=_settings_now(repo).tax_changes_need_second_person, by=_who(caller), jurisdiction=JURISDICTION,
expected_turnover=expected_turnover, flat_rate_sector=flat_rate_sector or "", flat_rate_pct=pct,
first_year_until=_date(first_year_until, "first_year_until"), join_limit=join, ni_goods=ni_goods,
partial_exemption_year_end=partial_exemption_year_end, period_starts=starts)
return _put_registration(repo, row, caller)
def _registration_row(repo, effective_from: str) -> registration.Registration:
on = _date(effective_from, "effective_from")
row = next((r for r in registration.live(repo.registrations.rows()) if r.effective_from == on), None)
if row is None:
raise Refused(404, f"no registration row from {effective_from}")
return row
@TAX.use_case(kind="command", route="POST /registration/approve", permission=APPROVE, idempotent=True)
def approve_registration(caller: dict, effective_from: IsoDate) -> dict:
"""A second holder of tax:code:approve approves a registration change that waits for it. Never its author, and only
before its date."""
with repository() as repo, _decide():
row = registration.approve(_registration_row(repo, effective_from), today=_today(repo), approver=_who(caller),
approver_ids=_ids(caller))
return _put_registration(repo, row)
@TAX.use_case(kind="command", route="POST /registration/withdraw", permission=REGISTRATION_WRITE, idempotent=True)
def withdraw_registration(caller: dict, effective_from: IsoDate) -> dict:
"""Withdraw a registration row that has not taken effect yet: it never will. One in effect is never edited — record a
correction from tomorrow."""
with repository() as repo, _decide():
row = registration.withdraw(_registration_row(repo, effective_from), today=_today(repo))
return _put_registration(repo, row)
# ═══ tax codes and their rates (TAX-02, TAX-03) ══════════════════════════════════════════════
def _code(repo, caller: dict, tax_code: str) -> tuple:
found = repo.codes.get(tax_code)
if found is None or (found[0].lifecycle == CodeState.DRAFT.value and not _sees_drafts(caller)):
raise Refused(404, f"no tax code {tax_code}")
return found
@TAX.use_case(kind="query", route="GET /codes", permission=READ)
def list_codes(caller: dict, lifecycle: Optional[Literal["draft", "active", "retired"]] = None) -> dict:
"""The tax codes with the rate in force today. Draft codes are shown to whoever keeps VAT, not to a biller."""
with repository() as repo:
today = _today(repo)
codes = [c for c in repo.codes.all() if (_sees_drafts(caller) or c.lifecycle != CodeState.DRAFT.value)
and (lifecycle is None or c.lifecycle == lifecycle)]
return {"codes": [_code_view(c, 0, today, detail=False) for c in codes]}
@TAX.use_case(kind="query", route="GET /codes/{tax_code}", permission=READ)
def get_code(caller: dict, tax_code: Key) -> dict:
"""One code with every dated rate (a pending or withdrawn row is shown as such) and, for a draft, what activation still needs."""
with repository() as repo:
c, version = _code(repo, caller, tax_code)
return _code_view(c, version, _today(repo))
@TAX.use_case(kind="command", route="POST /codes/load-pack", permission=SEED, idempotent=True)
def load_pack(caller: dict) -> dict:
"""Load the UK pack on a new installation: S standard 20%, R reduced 5%, Z zero, E exempt, OS outside the scope, RC
domestic reverse charge (VAT due 20%, charged 0%), NV no VAT (not registered) — each with its treatment, its wording and
its rate from the real date — as DRAFT codes for the tax admin to activate, plus the customer treatments and product
categories the rules choose from. Refused when the installation already has other codes; repeated, it changes nothing."""
def unit(repo):
have = {c.tax_code for c in repo.codes.all()}
pack = {p.tax_code for p in uk_pack.UK_CODES}
if have - pack:
raise unmet("new_installation", "the UK pack loads on a new installation with no codes of its own",
"Tax codes: add or change codes by hand")
by = _who(caller)
added = []
for p in uk_pack.UK_CODES:
if p.tax_code in have:
continue
c = TaxCode.add(p.tax_code, p.treatment, description=p.description, document_wording=p.wording,
reporting_box=p.box, by=by)
c.rates.append(Rate(p.tax_code, p.valid_from, p.rate_pct, p.charged_pct, ChangeState.APPROVED.value, by))
repo.codes.save(c, 0)
added.append(p.tax_code)
for code, description, needs in uk_pack.UK_TREATMENTS:
if repo.treatments.get(code) is None:
repo.treatments.put(code, description, needs)
for code, description in uk_pack.UK_CATEGORIES:
if repo.categories.get(code) is None:
repo.categories.put(code, description, by)
for key, value, unit, source in uk_pack.UK_STATUTORY:
if not any(r["stat_key"] == key for r in repo.statutory.where(stat_key=key)):
repo.statutory.put({"stat_key": key, "valid_from": FOREVER, "value": value, "unit": unit, "source": source, "set_by": by})
return {"codes_added": added, "codes_unchanged": sorted(have & pack)}
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /codes", permission=CODE_WRITE)
def add_code(caller: dict, tax_code: Key, treatment: Literal["standard", "reduced", "zero", "exempt", "outside_scope",
"reverse_charge", "not_registered", "postponed_import"],
description: Optional[Name] = None, document_wording: Optional[Note] = None, reporting_box: Optional[Name] = None,
rate_pct: Optional[Pct] = None, charged_pct: Optional[Pct] = None, valid_from: Optional[IsoDate] = None,
direction: Literal["sale", "purchase"] = "sale", input_reporting_box: Optional[Name] = None,
value_box: Optional[Literal["6", "7", "8", "9"]] = None, reclaimable_pct: Optional[Pct] = None,
attribution: Optional[Literal["taxable", "exempt", "residual"]] = None, eu_goods: bool = False) -> dict:
"""Add a tax code as a DRAFT (the code is the one the customer recognises: S, RC, Z). A code in use is refused. Give its
first rate here (`rate_pct`, from `valid_from`: any date, since no document has used the code) or later with
schedule_rate; activate_code then makes it usable. A code is never deleted. A PURCHASE code (`direction: purchase`) is what a
supplier's bill takes: `reclaimable_pct` says how much of its VAT is reclaimed (100, or 0 for business entertainment),
`input_reporting_box` the box of a reverse charge's input side, `value_box` where the net is reported, `attribution` (taxable |
exempt | residual) what partial exemption does with its input VAT. `eu_goods` marks Northern Ireland goods to or from the EU:
their net also lands in box 8 (sales) or 9 (acquisitions) and feeds the EC sales list and Intrastat."""
def unit(repo):
today = _today(repo)
if repo.codes.get(tax_code) is not None:
raise Refused(409, f"the code {tax_code} is already in use")
c = TaxCode.add(tax_code, treatment, description=description or "", document_wording=document_wording or "",
reporting_box=reporting_box or "", by=_who(caller), direction=direction,
input_reporting_box=input_reporting_box or "", value_box=value_box or "", reclaimable_pct=reclaimable_pct,
attribution=attribution or "", eu_goods=eu_goods)
if rate_pct is not None:
c.schedule_rate(rate_pct, charged_pct, _date(valid_from, "valid_from"), today=today, second_person=False, by=_who(caller))
repo.codes.save(c, 0)
return _code_view(c, 1, today)
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /codes/{tax_code}/describe-code", permission=CODE_WRITE, if_match="version")
def describe_code(caller: dict, tax_code: Key, version: Optional[int] = None, description: Optional[Name] = None,
document_wording: Optional[Note] = None, reporting_box: Optional[Name] = None,
treatment: Optional[Literal["standard", "reduced", "zero", "exempt", "outside_scope", "reverse_charge",
"not_registered"]] = None) -> dict:
"""Rename a code or change the wording printed on the document. Its treatment and reporting box are fixed once a document
used the code (a different meaning is a new code); an issued document keeps the wording it froze."""
def unit(repo):
c, current = _code(repo, caller, tax_code)
if version is not None and int(version) != current:
raise Refused(409, f"{tax_code} changed since you opened it (you have version {version}, it is at {current}) — load it again")
c.describe(description=description, document_wording=document_wording, treatment=treatment,
reporting_box=reporting_box, by=_who(caller))
v = repo.codes.save(c, current)
return _code_view(c, v, _today(repo))
return _atomic(unit) if version is None else _once(unit)
def _once(unit):
with repository() as repo, _decide():
return unit(repo)
@TAX.use_case(kind="command", route="POST /codes/{tax_code}/activate-code", permission=CODE_WRITE, idempotent=True)
def activate_code(caller: dict, tax_code: Key) -> dict:
"""Make a draft code usable. It needs its treatment, a first rate (any date) and, for a reverse-charge code, the wording
printed on the invoice; otherwise the unmet gates come back as {gate, met, reason, fix}. An active code is returned as it is."""
def unit(repo):
c, current = _code(repo, caller, tax_code)
if c.lifecycle == CodeState.ACTIVE.value:
return _code_view(c, current, _today(repo))
c.activate(by=_who(caller))
return _code_view(c, repo.codes.save(c, current), _today(repo))
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /codes/{tax_code}/schedule-rate", permission=CODE_WRITE)
def schedule_rate(caller: dict, tax_code: Key, rate_pct: Pct, valid_from: Optional[IsoDate] = None,
charged_pct: Optional[Pct] = None) -> dict:
"""Schedule a rate change from a date AFTER today (default tomorrow): a tax point before it keeps the old rate, one from it
the new, and documents already issued are unchanged. A second rate for the same code and date, or a rate from today or
earlier, is refused (a correction is a new row from tomorrow). Where the owner's control asks, the change waits for a
second holder of tax:code:approve. `charged_pct` differs from `rate_pct` only under reverse charge."""
def unit(repo):
c, current = _code(repo, caller, tax_code)
settings = _settings_now(repo)
rate = c.schedule_rate(rate_pct, charged_pct, _date(valid_from, "valid_from"), today=_today(repo),
second_person=settings.tax_changes_need_second_person, by=_who(caller))
rate.created_sub = _sub(caller)
return _code_view(c, repo.codes.save(c, current), _today(repo))
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /codes/{tax_code}/approve-rate", permission=APPROVE, idempotent=True)
def approve_rate(caller: dict, tax_code: Key, valid_from: IsoDate) -> dict:
"""A second holder of tax:code:approve approves a scheduled rate change that waits for it. Never its author."""
def unit(repo):
c, current = _code(repo, caller, tax_code)
c.approve_rate(_date(valid_from, "valid_from"), today=_today(repo), approver=_who(caller), approver_ids=_ids(caller))
return _code_view(c, repo.codes.save(c, current), _today(repo))
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /codes/{tax_code}/withdraw-rate", permission=CODE_WRITE, idempotent=True)
def withdraw_rate(caller: dict, tax_code: Key, valid_from: IsoDate) -> dict:
"""Withdraw a rate that has not taken effect yet (its date is after today): it is marked withdrawn, with who and when, and
never takes effect. A rate in effect is never withdrawn — schedule a correction."""
def unit(repo):
c, current = _code(repo, caller, tax_code)
c.withdraw_rate(_date(valid_from, "valid_from"), today=_today(repo), by=_who(caller), now=_now())
return _code_view(c, repo.codes.save(c, current), _today(repo))
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /codes/{tax_code}/retire-code", permission=CODE_WRITE, idempotent=True)
def retire_code(caller: dict, tax_code: Key) -> dict:
"""Retire a code: no new determinations, issued documents keep it. Refused while a rule in effect today or later still
chooses it ({gate: no_rule_points_here}: end that rule first)."""
def unit(repo):
c, current = _code(repo, caller, tax_code)
if c.lifecycle == CodeState.RETIRED.value:
return _code_view(c, current, _today(repo))
c.retire(rules_pointing_here=rules.pointing_at(repo.rules.all(None), tax_code, _today(repo)), by=_who(caller))
return _code_view(c, repo.codes.save(c, current), _today(repo))
return _atomic(unit)
# ═══ categories, treatments (the vocabulary the rules join) ══════════════════════════════════
@TAX.use_case(kind="query", route="GET /categories", permission=READ)
def list_categories(caller: dict) -> dict:
"""How a product is taxed, in the abstract (standard, reduced, zero, exempt, construction_service). A product stores one
of these; this brick is the sole writer of the list (shared.yaml keys.tax_category)."""
with repository() as repo:
return {"categories": repo.categories.all()}
@TAX.use_case(kind="command", route="POST /categories/{category_code}", permission=CATEGORY_WRITE, idempotent=True)
def set_category(caller: dict, category_code: Key, description: Optional[Name] = None) -> dict:
"""Add a tax category, or change its description. A category code is lower-case letters, digits and underscores; it is
never deleted (a product may carry it)."""
with repository() as repo:
if not KEY.match(category_code) or category_code == rules.ANY:
raise Refused(422, "a category code is lower-case letters, digits and underscores (standard, construction_service)")
have = repo.categories.get(category_code)
text_ = description if description is not None else (have["description"] if have else "")
repo.categories.put(category_code, text_, _who(caller))
return {"category_code": category_code, "description": text_}
@TAX.use_case(kind="query", route="GET /treatments", permission=READ)
def list_treatments(caller: dict) -> dict:
"""How a customer is taxed (UK business, UK consumer, overseas business, CIS contractor) and which need a VAT number."""
with repository() as repo:
return {"treatments": repo.treatments.all()}
@TAX.use_case(kind="command", route="POST /treatments/{treatment_code}", permission=SETTINGS_WRITE, idempotent=True)
def set_treatment(caller: dict, treatment_code: Key, description: Optional[Name] = None,
needs_vat_number: Optional[bool] = None) -> dict:
"""Add a customer treatment, or change its description. Whether it needs a VAT number is fixed once a customer holds it
(a different meaning is a new treatment)."""
with repository() as repo:
if not KEY.match(treatment_code) or treatment_code == rules.ANY:
raise Refused(422, "a treatment code is lower-case letters, digits and underscores (uk_business)")
have = repo.treatments.get(treatment_code)
if have and needs_vat_number is not None and needs_vat_number != have["needs_vat_number"] \
and repo.parties.uses_treatment(treatment_code):
raise Refused(409, f"customers already hold {treatment_code}: whether it needs a VAT number is fixed — add a new treatment")
row = {"treatment_code": treatment_code,
"description": description if description is not None else (have["description"] if have else treatment_code),
"needs_vat_number": needs_vat_number if needs_vat_number is not None else (have["needs_vat_number"] if have else False)}
repo.treatments.put(row["treatment_code"], row["description"], row["needs_vat_number"])
return row
# ═══ tax rules (TAX-04) ══════════════════════════════════════════════════════════════════════
@TAX.use_case(kind="query", route="GET /rules", permission=READ)
def list_rules(caller: dict, category_code: Optional[Key] = None, treatment_code: Optional[Key] = None,
direction: Literal["sale", "purchase"] = "sale") -> dict:
"""Which code a (category, treatment, supply) combination gets, as dated rows, each marked in force, scheduled, pending
approval, ended or withdrawn. `any` matches every category or treatment."""
with repository() as repo:
today = _today(repo)
rows = [r for r in repo.rules.all(direction) if (category_code in (None, r.category_code)) and (treatment_code in (None, r.treatment_code))]
force = {r.scope: r for r in rules.in_force(rows, today)}
out = []
for r in rows:
v = _rule_view(r)
v["in_force_today"] = force.get(r.scope) is r
out.append(v)
return {"today": today, "rules": out}
def _rule_scope(repo, category_code: str, treatment_code: str) -> None:
if category_code != rules.ANY and repo.categories.get(category_code) is None:
raise Refused(422, f"the tax category '{category_code}' does not exist — add it first (Tax categories)")
if treatment_code != rules.ANY and repo.treatments.get(treatment_code) is None:
raise Refused(422, f"the customer treatment '{treatment_code}' does not exist — add it first (Customer treatments)")
@TAX.use_case(kind="command", route="POST /rules", permission=SETTINGS_WRITE)
def set_rule(caller: dict, category_code: Key, treatment_code: Key, tax_code: Key, supply: Supply = "any",
valid_from: Optional[IsoDate] = None, priority: Annotated[int, Field(ge=0, le=1000)] = 0,
direction: Literal["sale", "purchase"] = "sale", consumption: Consumption = "any") -> dict:
"""Say which tax code a (category, treatment, supply) combination gets, from a date. A rule for a pair nothing covers yet may
start today; a change to a covered pair starts AFTER today (default tomorrow) and is a new row, never an edit. Where the
owner's control asks, the change waits for a second holder of tax:code:approve. The same rule again changes nothing.
`any` for the category or treatment covers all; when several rules match, the higher priority wins, then the more specific.
`direction: purchase` makes it a PURCHASE rule: the treatment is the supplier's and the code must be a purchase code.
v6 (TAX-51): `consumption` eat_in | takeaway makes the rule apply to lines that say so; `any` (the default) covers both."""
def unit(repo):
_rule_scope(repo, category_code, treatment_code)
code = repo.codes.get(tax_code)
if code is None or code[0].lifecycle != CodeState.ACTIVE.value:
raise Refused(422, f"the tax code '{tax_code}' is not an active code — activate it first")
if code[0].direction != direction:
raise Refused(422, f"the tax code '{tax_code}' is a {code[0].direction} code: a {direction} rule chooses {direction} codes")
today, settings = _today(repo), _settings_now(repo)
scope = (category_code, treatment_code, supply, consumption)
every = repo.rules.all(direction)
mine = rules.live(every, scope)
last = max(mine, key=lambda r: r.valid_from) if mine else None
# A first row is one NOTHING covers: no rule already chooses a code for any pair this one touches, on any date from
# its start on (an `any` rule over a covered pair changes tax somebody is getting today).
first = not mine and not rules.covered(every, scope, _date(valid_from, "valid_from") or today)
if last and last.tax_code == tax_code and last.priority == priority and (valid_from is None or _date(valid_from, "valid_from") == last.valid_from):
return _rule_view(last)
start = schedule.start_date(_date(valid_from, "valid_from"), today=today, has_rows=not first, what="a tax rule",
first_default=today, first_floor_today=True)
if any(r.valid_from == start for r in mine):
raise Refused(409, f"there is already a rule for {scope} from {start}: a rule is never edited — change it from a later date")
wait = schedule.needs_second_person(control_on=settings.tax_changes_need_second_person, first_row=first)
row = Rule(category_code, treatment_code, supply, start, tax_code, priority, None,
ChangeState.PENDING.value if wait else ChangeState.APPROVED.value, _who(caller))
row.created_sub, row.direction, row.consumption = _sub(caller), direction, consumption
repo.rules.put(row)
return _rule_view(row)
return _once(unit)
def _find_rule(repo, category_code, treatment_code, supply, valid_from, direction="sale", consumption="any") -> Rule:
on = _date(valid_from, "valid_from")
found = next((r for r in rules.live(repo.rules.all(direction), (category_code, treatment_code, supply, consumption)) if r.valid_from == on), None)
if found is None:
raise Refused(404, f"no rule ({category_code}, {treatment_code}, {supply}) from {valid_from}")
return found
@TAX.use_case(kind="command", route="POST /rules/end", permission=SETTINGS_WRITE)
def end_rule(caller: dict, category_code: Key, treatment_code: Key, supply: Supply = "any", valid_from: Optional[IsoDate] = None,
direction: Literal["sale", "purchase"] = "sale", consumption: Consumption = "any") -> dict:
"""End a rule from a date AFTER today (default tomorrow): from then no rule covers the pair, and a line of that pair is
refused rather than taxed at 0%. It is a new row — the rule is never edited. Do this before retiring the code it chooses."""
def unit(repo):
scope = (category_code, treatment_code, supply, consumption)
mine = rules.live(repo.rules.all(direction), scope)
if not mine:
raise Refused(404, f"no rule {scope} to end")
today, settings = _today(repo), _settings_now(repo)
start = schedule.start_date(_date(valid_from, "valid_from"), today=today, has_rows=True, what="ending a rule")
if any(r.valid_from == start for r in mine):
raise Refused(409, f"there is already a rule row for {scope} from {start}")
wait = schedule.needs_second_person(control_on=settings.tax_changes_need_second_person, first_row=False)
row = Rule(category_code, treatment_code, supply, start, "", 0, None,
ChangeState.PENDING.value if wait else ChangeState.APPROVED.value, _who(caller))
row.created_sub, row.direction, row.consumption = _sub(caller), direction, consumption
repo.rules.put(row)
return _rule_view(row)
return _once(unit)
@TAX.use_case(kind="command", route="POST /rules/approve", permission=APPROVE, idempotent=True)
def approve_rule(caller: dict, category_code: Key, treatment_code: Key, valid_from: IsoDate, supply: Supply = "any",
direction: Literal["sale", "purchase"] = "sale", consumption: Consumption = "any") -> dict:
"""A second holder of tax:code:approve approves a rule change that waits for it. Never its author, and only before its date."""
with repository() as repo, _decide():
row = _find_rule(repo, category_code, treatment_code, supply, valid_from, direction, consumption)
if row.change_state == ChangeState.PENDING.value:
schedule.approvable(row.valid_from, today=_today(repo), author=row.created_by, author_sub=row.created_sub, approver=_who(caller),
approver_ids=_ids(caller), what="rule change")
row.change_state, row.approved_by = ChangeState.APPROVED.value, _who(caller)
repo.rules.put(row)
return _rule_view(row)
@TAX.use_case(kind="command", route="POST /rules/withdraw", permission=SETTINGS_WRITE, idempotent=True)
def withdraw_rule(caller: dict, category_code: Key, treatment_code: Key, valid_from: IsoDate, supply: Supply = "any",
direction: Literal["sale", "purchase"] = "sale", consumption: Consumption = "any") -> dict:
"""Withdraw a rule row that has not taken effect yet: it never will. One in effect is never withdrawn — change it from tomorrow."""
with repository() as repo, _decide():
row = _find_rule(repo, category_code, treatment_code, supply, valid_from, direction, consumption)
schedule.withdrawable(row.valid_from, today=_today(repo), what="rule")
row.change_state, row.withdrawn_by = ChangeState.WITHDRAWN.value, _who(caller)
repo.rules.put(row)
return _rule_view(row)
@TAX.use_case(kind="query", route="GET /rates", permission=READ)
def get_rates(caller: dict, country: text(2, min_length=2), categories: text(4000, min_length=1), on: IsoDate,
consumption: Optional[Literal["eat_in", "takeaway"]] = None, treatment_code: Optional[Key] = None) -> dict:
"""TAX-52: the dated rate table an offline register sells with. One row per (category, supply, consumption) with the tax code, the rate
and the validity in force `on`, for the treatment of an anonymous buyer (walk_in_treatment_code unless one is named), sale direction;
`valid_until` is the day before the next change of any answered row (null when none is scheduled), so the register refreshes before
it; a category with no rule for the treatment is in `missing` and must not be sold offline. A snapshot, never a frozen answer: the
receipt is recomputed at sync by compute_tax."""
point = _date(on, "on")
wanted = [c.strip() for c in categories.split(",") if c.strip()]
if not wanted or len(wanted) > 200:
raise Refused(422, "categories is a comma-separated list of 1 to 200 tax categories")
with repository() as repo:
treatment = treatment_code or _settings_on(repo, point).walk_in_treatment_code
if repo.treatments.get(treatment) is None:
raise Refused(422, f"the treatment '{treatment}' does not exist")
every, facts = repo.rules.all("sale"), _facts(repo, point)
codes = {c.tax_code: c for c in repo.codes.all()}
rows, missing, changes = [], [], []
for cat in wanted:
near = [r for r in every if r.change_state != ChangeState.WITHDRAWN.value and r.category_code in (cat, rules.ANY)
and r.treatment_code in (treatment, rules.ANY)]
changes += [r.valid_from for r in near if r.valid_from > point and r.change_state == ChangeState.APPROVED.value]
matched = [r for r in rules.in_force(every, point) if r.category_code in (cat, rules.ANY) and r.treatment_code in (treatment, rules.ANY)]
found = False
for supply in sorted({r.supply for r in matched}):
kinds = [("eat_in", "takeaway") if any(r.consumption != rules.ANY for r in matched) else ("any",)][0]
for kind in kinds:
if consumption and kind not in (consumption, "any"):
continue
rule = rules.pick(every, cat, treatment, supply, point, "" if kind == "any" else kind)
f = facts.get(rule.tax_code) if rule else None
if rule is None or f is None or not f.usable:
continue
rate = codes[rule.tax_code].rate_on(point)
changes += [r.valid_from for r in codes[rule.tax_code].rates if r.in_force and r.valid_from > point]
rows.append({"category_code": cat, "supply": supply, "consumption": kind, "tax_code": rule.tax_code, "rate_pct": _pct(f.rate_pct),
"rule_valid_from": rule.valid_from, "rate_valid_from": rate.valid_from if rate else None})
found = True
if not found:
missing.append(cat)
until = min(changes) - timedelta(days=1) if changes else None
return {"country": country.upper(), "on": point, "treatment_code": treatment, "valid_until": until, "rates": rows, "missing": missing}
# ═══ a customer's treatment and VAT number (TAX-05) ══════════════════════════════════════════
def _party_view(r: dict) -> dict:
return {"bp_id": r["bp_id"], "valid_from": None if r["valid_from"] == FOREVER else r["valid_from"],
"treatment_code": r["treatment_code"], "vat_number": r["vat_number"] or None,
"checked_on": r.get("checked_on"), "check_result": r.get("check_result") or None, "recorded_by": r["recorded_by"],
"check_name": r.get("check_name") or None, "check_address": r.get("check_address") or None,
"consultation_number": r.get("consultation_number") or None, "check_flag": r.get("check_flag") or None,
"confirmed_by": r.get("confirmed_by") or None, "confirm_reason": r.get("confirm_reason") or None,
"notice": {"kind": r["notice_kind"], "reference": r.get("notice_ref"), "date": r.get("notice_date")} if r.get("notice_kind") else None}
@TAX.use_case(kind="query", route="GET /parties/{bp_id}", permission=READ)
def get_party_tax(caller: dict, bp_id: BigId) -> dict:
"""One customer's treatment and VAT number as dated rows (the first holds from the start) and the row in force today."""
with repository() as repo:
rows = repo.parties.rows_of(bp_id)
if not rows:
raise Refused(404, f"no tax treatment recorded for customer {bp_id}")
now = repo.parties.effective(bp_id, _today(repo))
return {"bp_id": bp_id, "current": _party_view(now) if now else None, "rows": [_party_view(r) for r in rows]}
@TAX.use_case(kind="command", route="POST /parties/{bp_id}/set-party-tax", permission=PARTY_WRITE)
def set_party_tax(caller: dict, bp_id: BigId, treatment_code: Key, vat_number: Optional[text(32)] = None,
valid_from: Optional[IsoDate] = None, notice_kind: Optional[Literal["end_user", "intermediary"]] = None,
notice_reference: Optional[text(100, min_length=1)] = None, notice_date: Optional[IsoDate] = None) -> dict:
"""Record a customer's VAT treatment (and VAT number) once, so every document for them is taxed the same way. A customer's
FIRST treatment holds from the start (no document could issue without one); a change is a new row from tomorrow at the
earliest and applies to documents not yet issued with a tax point from then. A treatment that needs a VAT number
(reverse charge, overseas business) refuses to save without one. The same treatment again changes nothing.
v4 (TAX-46): a construction customer that gives written notice it is an END USER or an INTERMEDIARY is outside the domestic
reverse charge: record the new treatment from tomorrow with `notice_kind`, `notice_reference` and `notice_date` — the notice is
kept on the row, and a notice without its reference and date is refused."""
with repository() as repo, _decide():
if notice_kind and not (notice_reference and notice_date):
raise unmet("notice_evidence", f"an {notice_kind.replace('_', ' ')} notice carries its reference and its date",
"Customer tax > the notice's reference and date")
notice = {"notice_kind": notice_kind, "notice_ref": notice_reference, "notice_date": _date(notice_date, "notice_date")} if notice_kind else {}
t = repo.treatments.get(treatment_code)
if t is None:
raise Refused(422, f"the customer treatment '{treatment_code}' does not exist — add it first (Customer treatments)")
number = (vat_number or "").strip().upper().replace(" ", "")
if t["needs_vat_number"] and not number:
raise unmet("vat_number", f"{treatment_code} needs the customer's VAT number: reverse charge and overseas business "
"treatments refuse to save without one", "Customer tax")
rows = repo.parties.rows_of(bp_id)
today = _today(repo)
latest = rows[-1] if rows else None
if latest and latest["treatment_code"] == treatment_code and (latest["vat_number"] or "") == number and valid_from is None:
return _party_view(latest)
start = schedule.start_date(_date(valid_from, "valid_from"), today=today, has_rows=bool(rows), what="a customer's treatment",
first_default=FOREVER, fix="Customer tax")
if any(r["valid_from"] == start for r in rows):
raise Refused(409, f"customer {bp_id} already has a treatment from {start}: a row is never edited — change it from a later date")
repo.parties.put(bp_id, start, treatment_code, number, _who(caller), **notice)
return _party_view({"bp_id": bp_id, "valid_from": start, "treatment_code": treatment_code, "vat_number": number,
"recorded_by": _who(caller), **notice})
# ═══ settings and the owner's controls (TAX-07, TAX-14) ══════════════════════════════════════
@TAX.use_case(kind="query", route="GET /settings", permission=READ)
def list_tax_settings(caller: dict) -> dict:
"""Every setting with its value today, its default, what it changes, and the dated rows behind it (a change is a new row
from tomorrow). The controls (marked) are set by whoever holds tax:settings:controls, never the tax admin."""
with repository() as repo:
today, rows = _today(repo), repo.settings.rows()
out = []
for d in SETTING_DEFS:
now = settings_book.value_at(rows, d.key, today)
out.append({"setting_key": d.key, "kind": d.kind, "value": d.default if now is None else now, "default": d.default,
"control": d.control, "changes": d.changes,
"rows": [{"valid_from": None if r.valid_from == FOREVER else r.valid_from, "value": r.value, "set_by": r.set_by}
for r in rows if r.setting_key == d.key]})
return {"today": today, "settings": out}
def _set_setting(repo, key: str, value, valid_from: Optional[str], caller: dict) -> dict:
d = SETTINGS.get(key)
if d is None:
raise Refused(404, f"no setting {key}")
why = check_setting(key, value)
if why:
raise Refused(422, f"{key} is {why}")
today, rows = _today(repo), repo.settings.rows()
mine = [r for r in rows if r.setting_key == key]
start = schedule.start_date(_date(valid_from, "valid_from"), today=today, has_rows=bool(mine), what=f"a change to {key}",
first_default=FOREVER, fix="Tax settings")
if mine and max(mine, key=lambda r: r.valid_from).value == value and valid_from is None:
return {"setting_key": key, "value": value, "valid_from": None, "unchanged": True}
if any(r.valid_from == start for r in mine):
raise Refused(409, f"{key} already has a row from {start}: a row is never edited — change it from a later date")
probe = settings_book.at([*rows, settings_book.SettingRow(key, start, value, _who(caller))], max(start, today))
if probe.retailer and probe.rounding_method in ROUND_DOWN_METHODS:
raise unmet("retailer_round_down", "a retailer must not round VAT down (VAT Notice 700 para 17.6): choose per_line_nearest",
"Tax settings > Rounding")
repo.settings.put(key, start, value, _who(caller))
return {"setting_key": key, "value": value, "valid_from": None if start == FOREVER else start, "unchanged": False}
@TAX.use_case(kind="command", route="POST /settings/{setting_key}", permission=SETTINGS_WRITE)
def set_tax_setting(caller: dict, setting_key: Key, value: Optional[Union[bool, int, Name]] = None,
valid_from: Optional[IsoDate] = None) -> dict:
"""Change one setting (rounding_method, prices_include_tax, retailer, simplified_invoice_threshold, ai.suggest_code.mode) as
a dated row: the first value holds from the start, a change starts AFTER today. null puts the default back. A retailer
setting refuses the two round-down methods (VAT Notice 700 para 17.6). The three CONTROLS on whoever keeps VAT —
tax_changes_need_second_person, override_approval, business_timezone — are refused here (403): set_tax_control."""
if setting_key in CONTROL_KEYS:
raise Refused(403, f"missing permission: {CONTROLS} ({setting_key} is a control on whoever keeps VAT: use set_tax_control)")
with repository() as repo, _decide():
return _set_setting(repo, setting_key, value, valid_from, caller)
@TAX.use_case(kind="command", route="POST /tax-controls/{setting_key}", permission=CONTROLS)
def set_tax_control(caller: dict, setting_key: Literal["tax_changes_need_second_person", "override_approval", "business_timezone",
"vat_return_needs_second_person", "hmrc_environment"],
value: Optional[Union[bool, int, Name]] = None, valid_from: Optional[IsoDate] = None) -> dict:
"""Change a control on whoever keeps VAT: tax_changes_need_second_person (true: a rate, rule or registration change waits for a
second holder of tax:code:approve), override_approval (second_person, or self_with_reason for a firm of one) or
business_timezone (the IANA zone in which "today" is counted, Europe/London). Dated rows like every setting. Only
tax:settings:controls — the owner or a finance controller, never the tax admin they control — may; tax:settings:write is refused."""
with repository() as repo, _decide():
return _set_setting(repo, setting_key, value, valid_from, caller)
# ═══ the tax on a document (TAX-10, TAX-20) ══════════════════════════════════════════════════
def _facts(repo, on: date) -> dict:
return repo.codes.facts_on(on)
def _context(repo, tax_point: date, bp_id: Optional[int], treatment_code: Optional[str], prices_include_tax: Optional[bool],
direction: str = "sale") -> Context:
rows = registration.live(repo.registrations.rows())
settings = _settings_on(repo, tax_point)
treatment = None
if bp_id is None and not treatment_code:
raise Refused(422, "name the business partner (bp_id) or the treatment (treatment_code) the document is taxed under", gate="party_or_treatment",
fix="send bp_id, or treatment_code for a party that is not a business partner")
if bp_id is not None:
party = repo.parties.effective(bp_id, tax_point)
treatment = party["treatment_code"] if party else None
if treatment and treatment_code and treatment_code != treatment:
raise Refused(422, f"treatment_code {treatment_code} is not the partner's treatment ({treatment}): the partner's wins", gate="treatment_matches_party",
fix="leave treatment_code out, or send the partner's own")
if treatment is None and treatment_code:
if repo.treatments.get(treatment_code) is None:
raise Refused(422, f"the customer treatment '{treatment_code}' does not exist")
treatment = treatment_code
return Context(tax_point, registration.is_registered(rows, tax_point), treatment, repo.rules.all(direction), _facts(repo, tax_point),
settings, prices_include_tax, direction)
def _vat_number_on(repo, tax_point: date) -> str:
row = registration.status_on(repo.registrations.rows(), tax_point)
return row.vat_number if row else ""
@TAX.use_case(kind="command", route="POST /documents/compute", permission=COMPUTE)
def compute_tax(caller: dict, lines: list[LineIn], tax_point: IsoDate, bp_id: Optional[BigId] = None,
currency: Currency = "GBP", document_ref: Optional[text(200, min_length=1)] = None,
treatment_code: Optional[Key] = None, prices_include_tax: Optional[bool] = None,
fx_rate: Optional[Fx] = None, direction: Literal["sale", "purchase"] = "sale",
ship_to_country: Optional[text(2, min_length=2)] = None, self_billed: bool = False,
due_date: Optional[IsoDate] = None) -> dict:
"""The tax on a quote, order, invoice or credit note (selling bricks call this on every price change and store the answer as
their own copy). Each line's code is chosen by the rule valid on the TAX POINT for (its category, the customer's treatment),
its rate is the code's rate on that date, and the rounding method is the setting in force then. Returns per line the code,
rate, net, VAT and gross, a per-rate summary (zero-rated and exempt as separate totals), the total, the method that rounded
them, the sterling VAT total of a foreign-currency document, and the unmet gates. A line no rule covers is NOT taxed at 0%: it
comes back refused. A business not VAT registered on the tax point charges no VAT (code NV). Pass `document_ref` to keep the
tax on the document (overrides and AI suggestions need it) — recomputing keeps a person's decisions; a frozen document is
never recomputed. The same inputs always give the same answer.
`direction: purchase` (purchasing, payments): a supplier's bill or credit. The rules read are the PURCHASE rules for
(category, the SUPPLIER's treatment, supply) and only purchase codes are chosen; each line answers vat_reclaimable and
vat_non_reclaimable by its code's reclaimable_pct (business entertainment: a cost, not a reclaim), and a purchase reverse
charge charges 0 and records the VAT due as output and as input. A pair no purchase rule covers is refused, never 0%.
Without `direction` it is a sale, exactly as before.
v4: a line may be PINNED (`pinned_from` an earlier frozen document, `tax_code` the code it reverses): it takes that document's
code and rate, never re-determined; it may not exceed the earlier net for that code (less what other issued pinned lines took).
`ship_to_country`, when sent, decides the place of supply of goods lines (outside GB: the overseas treatment); `self_billed`
marks a purchase document WE issued (the content gates read our self-billing series); `due_date` is when payment falls due (a
sale payable more than six months ahead stays on the invoice basis under cash accounting)."""
point = _date(tax_point, "tax_point")
ref = document_ref.strip() if document_ref else None
_bounds(lines, fx_rate)
def unit(repo):
ctx = _context(repo, point, bp_id, treatment_code, prices_include_tax, direction)
ctx.ship_to_country = (ship_to_country or "").upper()
ctx.pins = _pins(repo, lines, ref) if ctx.registered else {}
found = repo.documents.get(ref) if ref else None
doc = DocumentTax.determine(ref or "(not kept)", [DomainLine(ln.line_no, ln.description or "", ln.category_code or "", ln.supply,
ln.amount, ln.quantity, ln.unit_price, ln.pinned_from or "",
ln.tax_code or "", ln.consumption or "") for ln in lines],
ctx, bp_id=bp_id, currency=currency.upper(), fx_rate=fx_rate, vat_number=_vat_number_on(repo, point),
previous=found[0] if found else None, by=_who(caller))
if doc.currency != "GBP" and doc.fx_rate is None and doc.complete():
doc.set_fx(seams.ledger.fx_rate(doc.currency, point))
else:
doc.set_fx(doc.fx_rate)
_fits(doc)
doc.self_billed, doc.due_date = bool(self_billed), _date(due_date, "due_date")
version = 0
if ref:
version = repo.documents.save(doc, found[1] if found else 0)
return _doc_view(doc, version)
return _atomic(unit)
def _pins(repo, lines, this_ref: Optional[str]) -> dict:
"""TAX-47: line_no -> (the earlier frozen line's code facts, "pinned to <id>"). Gates: pinned_document (a frozen document),
pinned_code (it carries that code), pinned_within_original (the pinned amounts per code never exceed its net)."""
pins: dict = {}
pinned = [ln for ln in lines if ln.pinned_from]
if not pinned:
return pins
others = {}
for ref, (d, _) in repo.documents.get_many([r for r in repo.documents.refs() if r != this_ref]).items():
if d.status == DocState.FROZEN.value and any(x.pinned_from for x in d.lines):
others[ref] = d
asked: dict = {}
for ln in pinned:
found = repo.documents.get(ln.pinned_from)
if found is None or found[0].status != DocState.FROZEN.value:
raise unmet("pinned_document", f"{ln.pinned_from} is not a frozen document: a line is pinned only to an issued one",
"pin to the issued deposit invoice")
earlier = found[0]
codes = sorted({x.tax_code for x in earlier.lines if x.resolved})
code = ln.tax_code or (codes[0] if len(codes) == 1 else "")
src = next((x for x in earlier.lines if x.resolved and x.tax_code == code), None)
if src is None:
raise unmet("pinned_code", f"{ln.pinned_from} does not carry the code '{ln.tax_code or ''}' (it carries {', '.join(codes)})",
"pin to a code the earlier document used")
original = sum((r.net for r in earlier.computed().summary if r.code == code), Decimal(0))
took = sum((abs(x.amount) for d in others.values() for x in d.lines if x.pinned_from == ln.pinned_from and x.tax_code == code), Decimal(0))
asked[(ln.pinned_from, code)] = asked.get((ln.pinned_from, code), Decimal(0)) + abs(ln.amount)
if asked[(ln.pinned_from, code)] + took > original:
raise unmet("pinned_within_original", f"pinned amounts of {code} on {ln.pinned_from} would be "
f"{asked[(ln.pinned_from, code)] + took:,.2f}, more than its net {original:,.2f}", "pin no more than the earlier document's net")
pins[ln.line_no] = (CodeFacts(code, src.treatment, CodeState.ACTIVE.value, "", src.wording, src.reporting_box, src.rate_pct,
src.charged_pct, earlier.direction, src.input_reporting_box, src.value_box, src.reclaimable_pct,
src.attribution, src.eu_goods), f"pinned to {ln.pinned_from}")
return pins
LEDGER_LIMIT = Decimal("1e16") # NUMERIC(18,2): sixteen digits before the point
def _too_big(what: str) -> Refused:
return Refused(422, f"{what} is too large to keep: the amounts and the exchange rate together must stay below "
"10,000,000,000,000,000 sterling")
def _bounds(lines, fx_rate) -> None:
"""Refuse BEFORE computing what the books cannot hold: VAT is at most the amount (a rate is at most 100%), so gross is at
most twice it, and a foreign-currency VAT total is that times the exchange rate."""
if sum((abs(ln.amount) for ln in lines), Decimal(0)) * 2 * max(fx_rate or Decimal(1), Decimal(1)) >= LEDGER_LIMIT:
raise _too_big("the document")
def _fits(doc: DocumentTax) -> None:
"""The same check on the answer (the exchange rate may come from the ledger, after the first check)."""
c = doc.computed()
figures = [doc.vat_total_gbp] + ([c.net, c.vat, c.vat_due, c.gross] if c else [])
if any(v is not None and abs(v) >= LEDGER_LIMIT for v in figures):
raise _too_big("the sterling VAT total")
def _document(repo, document_ref: str) -> tuple:
found = repo.documents.get(document_ref)
if found is None:
raise Refused(404, f"no tax kept for document {document_ref}")
return found
@TAX.use_case(kind="query", route="GET /documents/{document_ref}", permission=READ)
def get_document_tax(caller: dict, document_ref: text(200, min_length=1)) -> dict:
"""The tax kept on a document: each line's code, the rule or person that chose it, an open AI suggestion, the per-rate
summary, the totals and the readiness gates {gate, met, reason, fix}. A frozen document reads exactly as it was frozen."""
with repository() as repo:
doc, version = _document(repo, document_ref)
return _doc_view(doc, version)
def _usable(repo, tax_code: str, on: date) -> Optional[CodeFacts]:
return _facts(repo, on).get(tax_code)
def _ctx_for(repo, doc: DocumentTax) -> Context:
return Context(doc.tax_point, doc.registered, doc.treatment_code or None, [], _facts(repo, doc.tax_point), _settings_on(repo, doc.tax_point))
@TAX.use_case(kind="command", route="POST /documents/{document_ref}/lines/{line_no}/override-line", permission=OVERRIDE)
def override_line(caller: dict, document_ref: text(200, min_length=1), line_no: Annotated[int, Field(ge=1, le=99999)],
tax_code: Key, reason: Reason) -> dict:
"""Override a line's tax code with a reason (an unusual supply handled without a new rule). The line shows the new code, the
reason and who changed it. Where the override LOWERS the tax it waits for approval by a holder of
tax:document_tax:approve_override who is not you (unless the owner's control allows self-approval with a reason) before
the document can issue."""
def unit(repo):
doc, version = _document(repo, document_ref)
doc.override(line_no, _usable(repo, tax_code, doc.tax_point), reason, _ctx_for(repo, doc), by=_who(caller), by_sub=_sub(caller))
v = repo.documents.save(doc, version)
return _doc_view(doc, v)
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /documents/{document_ref}/lines/{line_no}/approve-override", permission=APPROVE_OVERRIDE)
def approve_override(caller: dict, document_ref: text(200, min_length=1), line_no: Annotated[int, Field(ge=1, le=99999)],
reason: Optional[Reason] = None) -> dict:
"""Approve an override that lowers tax. The approver is never the person who made it, unless the owner's control
(override_approval = self_with_reason) allows a firm of one to approve its own with a reason."""
def unit(repo):
doc, version = _document(repo, document_ref)
mode = _settings_now(repo).override_approval
doc.approve_override(line_no, approver=_who(caller), mode=mode, reason=reason or "", approver_ids=_ids(caller))
return _doc_view(doc, repo.documents.save(doc, version))
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /documents/{document_ref}/lines/{line_no}/ai-suggest-code", permission=SUGGEST)
def ai_suggest_code(caller: dict, document_ref: text(200, min_length=1), line_no: Annotated[int, Field(ge=1, le=99999)]) -> dict:
"""The system asks the AI to suggest a tax code for a line no rule covers (suggest mode; ai.suggest_code.mode). The proposal
— a code, a confidence and the reason in one sentence — waits on the line; NOTHING is applied until a biller accepts it
(accept_suggestion). The line text and the customer's treatment go to the sovereign model, no other customer data. With no
AI provider connected, or the step off, the line stays refused until a person chooses a code."""
def unit(repo):
doc, version = _document(repo, document_ref)
mode = _settings_now(repo).suggest_code_mode
if mode == "off":
raise Refused(409, "the AI code suggestion is switched off (ai.suggest_code.mode): a person chooses the code")
line = doc.line(line_no)
facts = _facts(repo, doc.tax_point)
context = {"description": line.description, "category_code": line.category_code, "supply": line.supply,
"treatment_code": doc.treatment_code,
"codes": json.dumps([{"tax_code": f.tax_code, "treatment": f.treatment, "description": f.description,
"rate_pct": _pct(f.rate_pct)} for f in facts.values() if f.usable]),
"refusal": line.refusal}
answer = ai_steps.ask("ai_suggest_code", mode, f"tax_document_line:{document_ref}#{line_no}", version, context)
out = {"ai": answer.as_dict(), "suggested": False}
code = (answer.suggestion or {}).get("tax_code") if answer.kind == "done" else None
if code:
f = facts.get(str(code))
if f is None or not f.usable:
out["ai"]["message"] = f"the AI suggested '{code}', which is not an active code: nothing recorded, a person chooses"
else:
doc.suggest(line_no, f, reason=answer.rationale, confidence=answer.confidence, model=answer.model,
inputs=json.dumps(context), by=_who(caller))
out["suggested"] = True
version = repo.documents.save(doc, version)
return {**out, "document": _doc_view(doc, version)}
return _atomic(unit)
@TAX.use_case(kind="command", route="POST /documents/{document_ref}/lines/{line_no}/accept-suggestion", permission=OVERRIDE)
def accept_tax_suggestion(caller: dict, document_ref: text(200, min_length=1), line_no: Annotated[int, Field(ge=1, le=99999)],
decision: Literal["accept", "reject"], note: Optional[Reason] = None) -> dict:
"""Accept or reject the code the AI suggested for a line. Accepted: the line takes the code, recorded as an override with the
reason, the suggestion, its confidence, the model and you as the person who accepted; a code that lowers tax waits for
approval exactly as an override does. Rejected: the line stays refused until you choose a code. Nothing is applied without a person."""
def unit(repo):
doc, version = _document(repo, document_ref)
s = next((x for x in doc.suggestions if x.line_no == line_no and x.state == "open"), None)
facts = _usable(repo, s.tax_code, doc.tax_point) if s else None
doc.decide_suggestion(line_no, decision == "accept", facts, _ctx_for(repo, doc), by=_who(caller), now=_now(), note=note or "", by_sub=_sub(caller))
return _doc_view(doc, repo.documents.save(doc, version))
return _atomic(unit)
def _contents(c: Optional[ContentsIn]) -> vat_invoice.Contents:
c = c or ContentsIn()
return vat_invoice.Contents(title=c.title or "", invoice_number=c.invoice_number or "", issue_date=_date(c.issue_date, "issue_date"),
supplier_name=c.supplier_name or "",
supplier_address=(c.supplier_address.one_line() if hasattr(c.supplier_address, "one_line") else c.supplier_address) or "",
supplier_vat_number=c.supplier_vat_number or "", customer_name=c.customer_name or "",
customer_address=c.customer_address or "", cash_discount_applies=c.cash_discount_applies,
cash_discount_pct=c.cash_discount_pct, simplified=c.simplified)
def _content_gates(repo, doc: DocumentTax, contents: Optional[ContentsIn], caller: dict) -> list:
partner = seams.partners.customer(doc.bp_id, caller) if doc.bp_id is not None and doc.direction == "sale" else None
gates = vat_invoice.check(doc, _contents(contents), threshold=_settings_on(repo, doc.tax_point).simplified_invoice_threshold,
partner=partner)
return gates + _vat_number_gates(repo, doc)
def _vat_number_gates(repo, doc: DocumentTax) -> list:
"""TAX-13 (v4): a reverse-charge sale to a customer whose VAT number was checked and found invalid, or whose last check is older
than vat_number_check_days, waits for a fresh check. A number never checked is not blocked here (the check is the system's job)."""
from brick_tax.domain.errors import Gate
if doc.direction != "sale" or doc.bp_id is None or not any(ln.treatment == Treatment.REVERSE_CHARGE.value for ln in doc.lines):
return []
p = repo.parties.effective(doc.bp_id, doc.tax_point)
if not p or not p.get("check_result"):
return []
days = _settings_on(repo, doc.tax_point).vat_number_check_days
today = _today(repo)
age = (today - p["checked_on"]).days if p.get("checked_on") else None
if p["check_result"] == "invalid":
return [Gate("vat_number_checked", False, f"HMRC found the customer's VAT number {p['vat_number']} invalid: a reverse charge on a "
"customer who is not registered is a VAT debt for us", "Customer tax > correct the number and check it again")]
if p["check_result"] == "could_not_check" and not p.get("confirmed_by"):
return [Gate("vat_number_checked", False, "the customer's VAT number could not be checked and no person has confirmed it",
"Customer tax > confirm the number")]
if age is not None and age > days:
return [Gate("vat_number_checked", False, f"the customer's VAT number was last checked {age} days ago (more than {days})",
"Customer tax > check the number again")]
return [Gate("vat_number_checked", True, "the customer's VAT number check is current", "")]
@TAX.use_case(kind="command", route="POST /documents/{document_ref}/check-contents", permission=COMPUTE)
def check_contents(caller: dict, document_ref: text(200, min_length=1), contents: Optional[ContentsIn] = None,
direction: Optional[Literal["sale", "purchase"]] = None) -> dict:
"""Before a document issues: check it carries everything a VAT invoice must show (VAT Notice 700 para 16.3.1) — number, tax
point, issue date, our name, address and VAT number, the customer's name and address, per line the description, quantity,
unit price, rate and net, the total net and VAT, any cash-discount rate, the sterling VAT total of a foreign-currency
invoice — as one gate each {gate, met, reason, fix}, plus the tax's own gates (every line coded, no open AI suggestion, no
override waiting). A business that is not VAT registered is checked for the opposite: no VAT number, not titled "VAT
invoice". The answer is `ready`; nothing changes.
With `direction: purchase` (purchasing, before a supplier bill is approved) the gates check the SUPPLIER's name, address and
VAT number and invoice number and date, and our name as the customer: a bill with no supplier VAT number answers
supplier_vat_number unmet — input VAT cannot be reclaimed on it."""
with repository() as repo, _decide():
doc, _ = _document(repo, document_ref)
if direction is not None and direction != doc.direction:
raise Refused(409, f"{document_ref} was computed as a {doc.direction}, not a {direction}")
gates = doc.gates() + _content_gates(repo, doc, contents, caller)
less = vat_invoice.is_less_detailed(doc, _settings_on(repo, doc.tax_point).simplified_invoice_threshold)
return {"document_ref": document_ref, "status": doc.status, "ready": all(g.met for g in gates) or doc.status == "frozen",
"title": "less_detailed_invoice" if less else doc.title, "gates": [g.as_dict() for g in gates]}
@TAX.use_case(kind="command", route="POST /documents/{document_ref}/freeze", permission=COMPUTE, idempotent=True)
def freeze_document_tax(caller: dict, document_ref: text(200, min_length=1), contents: Optional[ContentsIn] = None) -> dict:
"""Freeze the tax on a document at issue: it is the tax the customer received, forever. Refused with the unmet gates
{gate, met, reason, fix} unless every gate is met (contents complete, every line coded, no open AI suggestion, no override
waiting for approval). Afterwards recomputing is refused ("issued: credit it instead") and a later rate change changes
nothing. Freezing twice is a no-op. The codes it used can no longer change their meaning."""
def unit(repo):
doc, version = _document(repo, document_ref)
if doc.status == DocState.FROZEN.value:
return _doc_view(doc, version)
if doc.currency != "GBP" and doc.fx_rate is None:
doc.set_fx(seams.ledger.fx_rate(doc.currency, doc.tax_point))
_fits(doc)
doc.freeze(_content_gates(repo, doc, contents, caller), by=_who(caller), now=_now())
for code in sorted({ln.tax_code for ln in doc.lines}):
c, cv = repo.codes.get(code)
if not c.used:
c.mark_used()
repo.codes.save(c, cv)
return _doc_view(doc, repo.documents.save(doc, version))
return _atomic(unit)
# ═══ the one-off Odoo migration (TAX-15) ═════════════════════════════════════════════════════
def _plan_view(p: odoo_migration.Plan, door: OdooDoorIn) -> dict:
return {"registration": p.registration and {**p.registration, "vat_number": p.registration["vat_number"] or None},
"rounding": {"rounding_method": p.rounding[0], "from": p.rounding[1]},
"codes": [{"odoo_id": c.odoo_id, "odoo_name": c.odoo_name, "tax_code": c.tax_code, "treatment": c.treatment,
"rate_pct": _pct(c.rate_pct), "charged_pct": _pct(c.charged_pct), "lifecycle": c.lifecycle,
"reporting_box": c.reporting_box, "creates": c.creates} for c in p.codes],
"treatments": [{"treatment_code": t[0], "description": t[1], "needs_vat_number": t[2]} for t in p.treatments],
"treatment_of_position": {str(k): v for k, v in p.treatment_of_position.items()},
"categories": [c[0] for c in p.categories], "rules": len(p.rules), "customers": len(p.parties),
"handed_to_product": {str(k): v for k, v in p.handed_to_product.items()},
"unresolved_partners": p.unresolved_partners, "skipped_taxes": p.skipped_taxes, "notes": p.notes,
"proposed_code_map": p.proposals}
def _read_for_migration(door: OdooDoorIn):
try:
return odoo_migration.read_odoo(door.company_id, odoo_migration.window(door))
except odoo.OdooUnavailable as e:
raise Refused(503, str(e)) from None
def _migration_plan(caller: dict, door: OdooDoorIn):
rows = _read_for_migration(door)
with repository() as repo:
existing = {t["treatment_code"]: t for t in repo.treatments.all()}
try:
return rows, odoo_migration.plan(rows, door, lambda oid: seams.partners.resolve_odoo(door.odoo_db, oid, caller), existing)
except RuleBroken as e:
raise Refused(409, str(e)) from None
def _reconciled(door: OdooDoorIn, rows, p) -> list:
with repository() as repo: # a second transaction: the first has committed
return odoo_migration.reconcile(repo, rows, p, door)
@TAX.use_case(kind="command", route="POST /migration/odoo", permission=IMPORT, idempotent=True)
def migrate_from_odoo(caller: dict, door: OdooDoorIn) -> dict:
"""The one-off Odoo migration (run once at cutover by the system's service client, after business-partner and product):
the company's Odoo sale taxes, fiscal positions and customer VAT data become this module's FIRST rows — the registration
(from the HMRC date the customer gave at the door), tax codes with their rates and VAT-return boxes (an archived tax a
posted invoice used becomes a RETIRED code), customer treatments, rules, each customer's treatment and VAT number, and the
rounding method from cutover — so the first invoice after cutover is taxed as Odoo would have taxed it. Without
`confirm` it returns the plan (the codes and treatments it proposes) and writes nothing. Safely re-runnable: keyed in
tax_migration_map, a second run writes nothing, and a row that exists with other content is listed, never overwritten.
History is read-only: no issued document's tax is read or written. Returns the plan, what was created or left, and the
reconcile checks {check, odoo, ours, equal, explanation}."""
rows, p = _migration_plan(caller, door)
if not door.confirm:
return {"confirmed": False, "plan": _plan_view(p, door), "written": None, "reconcile": None}
result = _atomic(lambda repo: odoo_migration.apply(repo, p, door, _who(caller)))
return {"confirmed": True, "plan": _plan_view(p, door), "written": result, "reconcile": _reconciled(door, rows, p)}
@TAX.use_case(kind="command", route="POST /migration/odoo/reconcile", permission=IMPORT, idempotent=True)
def reconcile_tax_migration(caller: dict, door: OdooDoorIn) -> dict:
"""Read-only: compare Odoo with what the migration wrote — active sale taxes, rate, reporting box, customers with a VAT number,
treatment per customer, determination parity (a difference is a hand-edited line in Odoo, listed by move and line), VAT by
code and the VAT-return boxes — each {check, odoo, ours, equal, explanation}. Run it again after ar's migration slice has
stored the invoices (the last two checks compare against them). Changes nothing."""
rows, p = _migration_plan(caller, door)
return {"reconcile": _reconciled(door, rows, p)}
# ═══ the Odoo id map other bricks read (shared.yaml conventions.odoo_migration.id_map) ═══════════════
@TAX.use_case(kind="query", route="GET /migration-map/{db}/{model}", permission=READ, tool=False)
def tax_migration_map(caller: dict, db: Name, model: Name, ids: Optional[text(8000)] = None) -> list:
"""What Odoo records became here: (odoo db, model, ids) -> our natural key, as this brick's Odoo migration wrote it in
tax_migration_map (an account.tax becomes a tax_code). Later bricks resolve Odoo ids through this map, never by matching names (shared.yaml
conventions.odoo_migration). `ids` is comma-separated (at most 500); an id with no mapping is left out, never a 404."""
with repository() as repo:
out = []
for oid in [i.strip() for i in (ids or "").split(",") if i.strip()][:500]:
row = repo.migration_map.get(db, model, oid)
if row is not None:
out.append({"odoo_model": model, "odoo_id": oid, "ours": row["our_key"]})
return out
from brick_tax.application import ( # noqa: E402,F401 (register the v4 use cases on TAX)
checks,
cis,
declarations,
exemption,
returns,
)