dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/returns.py
"""HMRC, the VAT return and bad debt relief (v4: TAX-26 .. TAX-35).
The owner connects the business to HMRC (the tokens live in the platform's secret store, never in a row); obligations are
HMRC's, copied here; a return's nine boxes are SUMS of entries CLAIMED from frozen document tax (a plain INSERT keyed
[document_ref, tax_code, entry_kind], so a document enters one return only); the tax admin locks it; a PERSON (not the one who
locked it) makes the legal declaration and submits it, with HMRC's fraud-prevention headers; HMRC's receipt makes it immutable
and the settle journal posts once. ar asks which return holds a tax point; payments reads the liabilities.
"""
import json
import secrets
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional
from erp_platform import work
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from erp_platform.types import IsoDate, text
from brick_tax.application import schemes
from brick_tax.application import use_cases as uc
from brick_tax.application.inputs import Key, Money, Reason, day
from brick_tax.application.use_cases import (
ADJUST,
BAD_DEBT_CLAIM,
BAD_DEBT_SUGGEST,
HMRC_CONNECT,
OBLIGATION_SYNC,
RETURN_PREPARE,
RETURN_READ,
RETURN_SUBMIT,
TAX,
TAX_POINT_READ,
)
from brick_tax.domain import bad_debt, fraud, schedule, vat_return
from brick_tax.domain.errors import Gate, GateUnmet, unmet
from brick_tax.infrastructure import seams
from brick_tax.infrastructure.hmrc import HmrcError, HmrcTimeout
DECLARATION = ("When you submit this VAT information you are making a legal declaration that the information is true and complete. "
"A false declaration can result in prosecution.")
P2 = Decimal("0.01")
EPOCH = date(1900, 1, 1)
CONNECT_FIX = "Tax settings > HMRC connection (the owner)"
def _utcnow() -> datetime:
"""Business-clock now as a naive UTC instant (work.now() lets a demo move the clock)."""
return work.now().astimezone(timezone.utc).replace(tzinfo=None)
def _naive_utc(v: Optional[datetime]) -> Optional[datetime]:
"""A stored timestamp as a naive UTC instant. SQLite returns naive values, the DataK3 pg wire returns AWARE ones: every
comparison against _utcnow() goes through here, so both compare."""
if v is None or v.tzinfo is None:
return v
return v.astimezone(timezone.utc).replace(tzinfo=None)
def _money(v) -> Decimal:
return Decimal(str(v)).quantize(P2)
# ═══ the connection to HMRC (TAX-26) ═════════════════════════════════════════════════════════
def _vrn(repo) -> str:
"""The VAT registration number HMRC knows us by: the digits of our registered VAT number (GB123456789 -> 123456789)."""
from brick_tax.domain import registration
row = registration.status_on(repo.registrations.rows(), uc._today(repo))
if row is None or not row.vat_number:
rows = [r for r in registration.live(repo.registrations.rows()) if r.vat_number]
row = rows[-1] if rows else None
if row is None or not row.vat_number:
raise unmet("registration_recorded", "no VAT registration is recorded: there is no VRN to connect or file for",
"Tax settings > Registration")
return row.vat_number[2:11]
def _conn_gate(why: str = "this business is not connected to HMRC") -> GateUnmet:
return unmet("hmrc_connected", why, CONNECT_FIX)
def _connection(repo, vrn: str) -> dict:
"""The valid connection for a VRN, or the gate {hmrc_connected} every return transition answers."""
row = repo.connections.get(vrn=vrn)
if row is None or row["status"] != "connected":
raise _conn_gate()
if row["expires_at"] is not None and _naive_utc(row["expires_at"]) <= _utcnow():
raise _conn_gate(f"the HMRC connection expired on {row['expires_at'].date()}")
return row
def _tokens(conn: dict) -> dict:
try:
tokens = seams.secrets.get(conn["secret_ref"])
except HmrcError as exc:
raise _conn_gate(exc.message) from None
if not tokens:
raise _conn_gate("the HMRC tokens are not in the secret store: connect again")
return tokens
def _conn_view(row: Optional[dict]) -> dict:
if row is None:
return {"connected": False, "gate": Gate("hmrc_connected", False, "this business is not connected to HMRC", CONNECT_FIX).as_dict()}
now = _utcnow()
expires = _naive_utc(row["expires_at"])
live = row["status"] == "connected" and expires > now
days = (expires - now).days
return {"connected": live, "vrn": row["vrn"], "scopes": (row["scopes"] or "").split(), "environment": row["environment"],
"authorised_by": row["authorised_by"], "authorised_at": _naive_utc(row["authorised_at"]), "expires_at": expires,
"status": row["status"] if live or row["status"] != "connected" else "expired", "days_left": days,
"reconnect_due": days <= 30,
"notice": f"the HMRC connection ends on {row['expires_at'].date()}: the owner reconnects" if days <= 30 else None,
"last_sync_error": row.get("sync_error")}
@TAX.use_case(kind="query", route="GET /hmrc/connection", permission=RETURN_READ)
def get_hmrc_connection(caller: dict) -> dict:
"""Whether this business is connected to HMRC for VAT: the VRN, the scopes granted, who authorised it and when, when HMRC's
18-month grant ends and whether the owner should reconnect (30 days before). Not connected answers the gate every return
transition answers: {gate: hmrc_connected, met: false, reason, fix}. The tokens are never shown."""
with uc.repository() as repo, uc._decide():
return _conn_view(repo.connections.get(vrn=_vrn(repo)))
@TAX.use_case(kind="query", route="GET /hmrc/connection/authorize-url", permission=HMRC_CONNECT)
def hmrc_authorize_url(caller: dict, redirect_uri: text(500, min_length=1)) -> dict:
"""Where the owner signs in to HMRC's Government Gateway as our business, granting read:vat and write:vat. The web screen
sends the browser there and, when HMRC redirects back with a code, calls connect_hmrc. `state` is for the screen to check."""
with uc.repository() as repo, uc._decide():
_vrn(repo)
env = uc._settings_now(repo).hmrc_environment
state = secrets.token_urlsafe(16)
return {"url": seams.hmrc.authorize_url(env, state, redirect_uri), "state": state, "environment": env}
@TAX.use_case(kind="command", route="POST /hmrc/connection/connect", permission=HMRC_CONNECT)
def connect_hmrc(caller: dict, code: text(500, min_length=1), redirect_uri: text(500, min_length=1)) -> dict:
"""Record the connection after the owner signed in at HMRC: the code HMRC gave is exchanged for tokens, which go to the
platform's secret store (a reference is kept here, never a token); the row records the VRN, the scopes, who authorised it
and when, and when it expires (HMRC's grant lasts 18 months). hmrc_environment (sandbox | production, the owner's control)
decides which HMRC is called."""
with uc.repository() as repo, uc._decide():
vrn = _vrn(repo)
env = uc._settings_now(repo).hmrc_environment
try:
tokens = seams.hmrc.exchange_code(env, code, redirect_uri)
ref = f"hmrc/{vrn}"
seams.secrets.put(ref, tokens)
except HmrcError as exc:
raise Refused(409, f"HMRC did not accept the sign-in ({exc.code}): try connecting again") from None
now = _utcnow()
expires = datetime.combine(vat_return.add_months(now.date(), 18), now.time())
repo.connections.put({"vrn": vrn, "scopes": "read:vat write:vat", "environment": env, "authorised_by": uc._who(caller),
"authorised_at": now, "expires_at": expires, "secret_ref": ref, "status": "connected",
"sync_error": None, "sync_error_at": None})
return _conn_view(repo.connections.get(vrn=vrn))
# ═══ HMRC's obligations (TAX-27) ═════════════════════════════════════════════════════════════
def _obligation_view(o: dict, ret=None) -> dict:
return {"vrn": o["vrn"], "period_key": o["period_key"], "period_start": o["period_start"], "period_end": o["period_end"],
"due_date": o["due_date"], "status": o["status"], "received_on": o["received_on"], "hmrc_error": o.get("hmrc_error"),
"return_status": ret.lifecycle if ret is not None else None}
@TAX.use_case(kind="command", route="POST /obligations/sync", permission=OBLIGATION_SYNC, idempotent=True)
def sync_vat_obligations(caller: dict, from_date: Optional[IsoDate] = None, to_date: Optional[IsoDate] = None) -> dict:
"""Copy HMRC's VAT obligations for a date range (default: two years back to a year ahead): one row per periodKey with HMRC's
own dates (never computed here); a period HMRC reports as filed — even if filed elsewhere — is marked filed with its received
date. A second run changes nothing. An HMRC error is recorded with its code on the connection and retried next day
(the daily clock-worker job, and when the VAT returns screen opens)."""
def unit(repo):
vrn = _vrn(repo)
conn = _connection(repo, vrn)
today = uc._today(repo)
start = uc._date(from_date, "from_date") or vat_return.add_months(today, -24)
end = uc._date(to_date, "to_date") or vat_return.add_months(today, 12)
env, tokens, now = uc._settings_now(repo).hmrc_environment, _tokens(conn), _utcnow()
got: dict = {}
try:
lo = start
while lo <= end: # HMRC answers at most 366 days at a time
hi = min(end, lo + timedelta(days=365))
for o in seams.hmrc.obligations(env, vrn, lo, hi, tokens, {}):
got[o["periodKey"]] = o
lo = hi + timedelta(days=1)
except HmrcError as exc:
repo.connections.put({"vrn": vrn, "sync_error": exc.code, "sync_error_at": now})
return {"vrn": vrn, "synced": 0, "obligations": [], "error": {"code": exc.code, "message": exc.message}}
changed = 0
for key in sorted(got):
o = got[key]
new = {"vrn": vrn, "period_key": key, "period_start": day(o["start"]), "period_end": day(o["end"]),
"due_date": day(o["due"]), "status": o["status"], "received_on": day(o.get("received"))}
old = repo.obligations.get(vrn=vrn, period_key=key)
if old and all(old[k] == v for k, v in new.items()):
continue
repo.obligations.put({**new, "synced_at": now})
changed += 1
if conn.get("sync_error"):
repo.connections.put({"vrn": vrn, "sync_error": None, "sync_error_at": None})
rows = [_obligation_view(o) for o in repo.obligations.where(vrn=vrn)]
return {"vrn": vrn, "synced": changed, "obligations": rows, "error": None}
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /vat-returns", permission=RETURN_READ)
def list_vat_returns(caller: dict) -> dict:
"""The VAT periods HMRC has told us about, each with the state of our return for it (none yet, draft, locked, submitting,
submitted), newest first."""
with uc.repository() as repo, uc._decide():
vrn = _vrn(repo)
rets = {r.period_key: r for r in repo.returns.all(vrn)}
obs = sorted(repo.obligations.where(vrn=vrn), key=lambda o: o["period_start"], reverse=True)
return {"vrn": vrn, "periods": [_obligation_view(o, rets.get(o["period_key"])) for o in obs]}
# ═══ preparing a return (TAX-28) ═════════════════════════════════════════════════════════════
def _split(box: str) -> list:
return [b.strip() for b in (box or "").split(",") if b.strip()]
def _value_boxes(ln, direction: str) -> str:
"""Where a line's net value is reported: box 6 / 7, and goods to or from the EU also in box 8 / 9 (which are part of 6 / 7)."""
vb, sale = ln.value_box or "", direction == "sale"
if vb in ("8", "9") or (not vb and ln.eu_goods):
return "6,8" if sale else "7,9"
return vb or ("6" if sale else "7")
def _doc_entries(doc, vrn: str, pk: str) -> list:
"""The entries one FROZEN document contributes: per tax code, its net, VAT and the boxes they land in (sterling)."""
factor = Decimal(1) if doc.currency == "GBP" else doc.fx_rate
if factor is None:
raise unmet("fx_rate", f"{doc.document_ref} is in {doc.currency} and has no sterling rate", "enter the rate on the document")
computed, split = doc.computed(), doc.purchase_split()
first: dict = {}
for ln in doc.lines:
first.setdefault(ln.tax_code, ln)
out = []
for s in computed.summary:
ln = first[s.code]
net, vat = _money(s.net * factor), _money(s.vat * factor)
box2 = Decimal(0)
if doc.direction == "sale":
box1, box4, rec = (vat if "1" in _split(ln.reporting_box) else Decimal(0)), Decimal(0), None
else:
rec = _money(split[1][s.code][0] * factor)
reverse = s.charged_pct == 0 and s.rate_pct > 0
due = _money(s.vat_due * factor) if reverse else Decimal(0)
box1 = due if "1" in _split(ln.reporting_box) else Decimal(0)
box2 = due if "2" in _split(ln.reporting_box) else Decimal(0) # NI acquisitions from the EU
box4 = rec
out.append({"document_ref": doc.document_ref, "tax_code": s.code, "entry_kind": "document", "vrn": vrn, "period_key": pk,
"direction": doc.direction, "tax_point": doc.tax_point, "net_amount": net, "vat_amount": vat,
"vat_reclaimable": rec, "box1_amount": box1, "box2_amount": box2, "box4_amount": box4,
"box_value": _value_boxes(ln, doc.direction), "source": "document", "detail": None,
"attribution": ln.attribution or None, "_gross": _money(s.gross * factor)})
return out
def _claim_repayments(repo, vrn: str, pk: str, claim) -> int:
"""TAX-35: every receipt allocated to a claimed invoice AFTER its claim repays the VAT fraction, once per allocation
(an entry keyed [invoice, allocation]); a reversing row (an undone allocation) repays negatively on the next return."""
rows = []
for c in (x for x in repo.claims.all() if x.lifecycle == bad_debt.CLAIMED):
for a in seams.payments.allocations([c.invoice_id], c.claimed_on) or []:
if day(a["allocated_on"]) <= c.claimed_on:
continue
amt = bad_debt.fraction_of(_money(a["amount"]), c.vat, c.gross) * (-1 if a.get("reverses_seq") is not None else 1)
key = f"{a['source_kind']}:{a['source_id']}:{a['seq']}"
rows.append({
"document_ref": c.invoice_id, "tax_code": key, "entry_kind": "bad_debt_repayment", "vrn": vrn, "period_key": pk,
"direction": "sale", "tax_point": day(a["allocated_on"]), "net_amount": Decimal(0), "vat_amount": amt,
"vat_reclaimable": None, "box1_amount": amt, "box4_amount": Decimal(0), "box_value": None,
"source": "bad_debt_repayment", "detail": f"repays claim on {c.invoice_id}, allocation {key}"})
return claim(rows)
def _recompute(repo, ret) -> None:
entries = repo.entries.where(vrn=ret.vrn, period_key=ret.period_key)
ret.set_boxes(entries, repo.adjustments.where(vrn=ret.vrn, period_key=ret.period_key))
def _return_view(repo, ret, version: int, extra: Optional[dict] = None) -> dict:
ob = repo.obligations.get(vrn=ret.vrn, period_key=ret.period_key) or {}
out = {"vrn": ret.vrn, "period_key": ret.period_key, "lifecycle": ret.lifecycle, "period_start": ob.get("period_start"),
"period_end": ob.get("period_end"), "due_date": ob.get("due_date"), "accounting_basis": ret.accounting_basis,
"boxes": {str(i): v for i, v in ret.boxes.items()}, "prepared_by": ret.prepared_by or None,
"locked_by": ret.locked_by or None, "locked_at": ret.locked_at, "submitted_by": ret.submitted_by or None,
"submitted_at": ret.submitted_at, "receipt": None, "settle_journal_id": ret.settle_journal_id or None,
"hmrc_error": ret.hmrc_error or None, "version": version}
if ret.fraud_headers_json:
out["fraud_headers"] = json.loads(ret.fraud_headers_json) # the headers sent, kept with the submission for audit
if ret.processing_date:
out["receipt"] = {"processingDate": ret.processing_date, "formBundleNumber": ret.form_bundle_number,
"paymentIndicator": ret.payment_indicator, "chargeRefNumber": ret.charge_ref_number}
return out | (extra or {})
def _load_return(repo, vrn: str, period_key: str) -> tuple:
found = repo.returns.get(vrn, period_key)
if found is None:
raise Refused(404, f"no VAT return {period_key}: prepare it first")
return found
def _prepare(repo, caller: dict, period_key: str, must_exist: bool) -> dict:
vrn = _vrn(repo)
ob = repo.obligations.get(vrn=vrn, period_key=period_key)
if ob is None or ob["status"] != "O":
raise unmet("obligation_open", f"HMRC shows no open obligation {period_key}" if ob is None
else f"HMRC shows {period_key} as filed", "VAT returns > sync with HMRC")
found = repo.returns.get(vrn, period_key)
if found is None and must_exist:
raise Refused(404, f"no VAT return {period_key} to refresh: prepare it first")
ret, version = found or (vat_return.VatReturn(vrn, period_key), 0)
ret.editable()
# this draft's entries are refreshed IN PLACE (a plain INSERT for a new key, an UPDATE for its own, a DELETE only for a key no
# longer produced): never delete-then-insert one key in a transaction. Claims of bad debt relief and other returns' entries stay.
p = schemes.plan(repo, vrn, ob)
touched: set = set()
counts = {"claimed": 0, "already_in_a_return": 0}
def claim_all(rows: list) -> int:
tags = []
for row in rows:
row.pop("_gross", None)
tags.append(row.pop("_mark", None))
won = 0
for row, tag, ok in zip(rows, tags, repo.entries.claim_many(rows)):
if ok:
won += 1
touched.add((row["document_ref"], row["tax_code"], row["entry_kind"]))
if tag:
schemes.mark(repo, tag, period_key)
return won
won = claim_all(p["rows"])
counts["claimed"], counts["already_in_a_return"] = won, len(p["rows"]) - won
counts["repayments"] = _claim_repayments(repo, vrn, period_key, claim_all)
repo.entries.sweep(vrn, period_key, touched)
suggested = _suggest_bad_debt(repo)
ret.prepared_by = uc._who(caller)
ret.accounting_basis, ret.flat_rate_pct = p["basis"], p["flat_rate_pct"]
_recompute(repo, ret)
version = repo.returns.save(ret, version)
return _return_view(repo, ret, version, {"entries": counts, "bad_debt_suggested": suggested, "scheme_gates": p["gates"], **p["extras"]})
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/prepare", permission=RETURN_PREPARE)
def prepare_vat_return(caller: dict, period_key: Key) -> dict:
"""Prepare the VAT return for an open HMRC obligation: boxes 1-9 from the frozen document tax of the period that is in no
return yet (each document's tax enters ONE return; a late supplier bill enters the first return prepared after it), plus
bad debt relief claims, repayments and adjustments. Box 3 = 1 + 2, box 5 = |3 - 4|, boxes 6-9 whole pounds (pence
dropped). Preparing again replaces only this draft's entries. The return stays a draft until locked."""
return uc._atomic(lambda repo: _prepare(repo, caller, period_key, False))
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/refresh", permission=RETURN_PREPARE)
def refresh_vat_return(caller: dict, period_key: Key) -> dict:
"""Refresh a DRAFT return: its entries are replaced from the documents now frozen; entries of other returns are untouched."""
return uc._atomic(lambda repo: _prepare(repo, caller, period_key, True))
@TAX.use_case(kind="query", route="GET /vat-returns/period/{period_key}", permission=RETURN_READ)
def get_vat_return(caller: dict, period_key: Key) -> dict:
"""One VAT return: its state, the nine boxes, HMRC's receipt once submitted."""
with uc.repository() as repo, uc._decide():
ret, version = _load_return(repo, _vrn(repo), period_key)
return _return_view(repo, ret, version)
@TAX.use_case(kind="query", route="GET /vat-returns/{period_key}/boxes/{box}", permission=RETURN_READ)
def drill_vat_return_box(caller: dict, period_key: Key, box: int) -> dict:
"""Drill a box to the entries that make it: each entry names its document, tax code, net, VAT and the box amount, plus the
adjustments in boxes 1 and 4. The box equals the sum of what this lists."""
if box not in range(1, 10):
raise Refused(422, "a VAT return has boxes 1 to 9")
with uc.repository() as repo, uc._decide():
vrn = _vrn(repo)
ret, _ = _load_return(repo, vrn, period_key)
rows = []
for e in repo.entries.where(vrn=vrn, period_key=period_key):
amount = (e["box1_amount"] if box == 1 else e["box4_amount"] if box == 4 else e["net_amount"] if str(box) == e["box_value"] else None)
if amount:
rows.append({"document_ref": e["document_ref"], "tax_code": e["tax_code"], "entry_kind": e["entry_kind"],
"tax_point": e["tax_point"], "net_amount": e["net_amount"], "vat_amount": e["vat_amount"], "box_amount": amount,
"detail": e["detail"]})
adj = [{"adjustment_no": a["adjustment_no"], "corrects_period_key": a["corrects_period_key"], "amount": a["amount"], "reason": a["reason"]}
for a in repo.adjustments.where(vrn=vrn, period_key=period_key) if a["box"] == box]
return {"period_key": period_key, "box": box, "value": ret.boxes[box], "entries": rows, "adjustments": adj}
# ═══ adjustments and error notices (TAX-29) ══════════════════════════════════════════════════
def _stat(repo, key: str, on: date) -> Decimal:
rows = [r for r in repo.statutory.where(stat_key=key) if r["valid_from"] <= on]
if not rows:
raise unmet("statutory_value", f"the figure {key} is not set: the law's figures are dated tax_statutory rows", "Tax settings > Statutory figures")
return Decimal(rows[-1]["value"])
def _thresholds(repo, on: date) -> tuple:
return (_stat(repo, "error_correction_small", on), _stat(repo, "error_correction_large", on), _stat(repo, "error_correction_pct", on))
def _net_errors(adjustments: list) -> Decimal:
return abs(sum((Decimal(str(a["amount"])) * (1 if a["box"] == 1 else -1) for a in adjustments), Decimal(0)))
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/record-adjustment", permission=ADJUST)
def record_vat_adjustment(caller: dict, period_key: Key, box: int, amount: Money, reason: Reason, corrects_period_key: Key,
corrects_period_end: Optional[IsoDate] = None) -> dict:
"""Correct an error found in an earlier period on this DRAFT return: `box` 1 (output VAT under-declared) or 4 (input VAT),
the amount, the reason and the period it corrects. Net errors within the threshold (10,000, or 50,000 and 1% of box 6; dated
tax_statutory rows) are corrected on the next return; larger ones are refused {gate: within_error_threshold} — notify HMRC on
form VAT652 and record the notice (record_vat_error_notice). An error older than four years is refused."""
if box not in (1, 4):
raise Refused(422, "an adjustment goes in box 1 or box 4")
if amount <= 0:
raise Refused(422, "an adjustment is an amount above zero (box 1: more VAT due; box 4: more reclaimed)")
def unit(repo):
vrn, today = _vrn(repo), uc._today(repo)
ret, version = _load_return(repo, vrn, period_key)
ret.editable()
ob = repo.obligations.get(vrn=vrn, period_key=corrects_period_key)
end = ob["period_end"] if ob else uc._date(corrects_period_end, "corrects_period_end")
if end is None:
raise unmet("corrected_period_known", f"period {corrects_period_key} is not one HMRC told us about",
"give the date the corrected period ended")
vat_return.check_four_years(end, today)
small, large, pct = _thresholds(repo, today)
mine = repo.adjustments.where(vrn=vrn, period_key=period_key)
new = {"amount": amount, "box": box}
vat_return.check_error_threshold(amount, ret.box6_total_sales_ex_vat, small, large, pct) # this error on its own
vat_return.check_error_threshold(_net_errors([*mine, new]), ret.box6_total_sales_ex_vat, small, large, pct) # and the return's total
no = 1 + max((a["adjustment_no"] for a in mine), default=0)
repo.adjustments.put({"vrn": vrn, "period_key": period_key, "adjustment_no": no, "corrects_period_key": corrects_period_key,
"corrects_period_end": end, "box": box, "amount": amount, "reason": reason,
"recorded_by": uc._who(caller), "recorded_at": _utcnow()})
_recompute(repo, ret)
return _return_view(repo, ret, repo.returns.save(ret, version), {"adjustment_no": no})
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/record-error-notice", permission=ADJUST)
def record_vat_error_notice(caller: dict, period_key: Key, reference: text(100, min_length=1), sent_on: IsoDate, periods: text(200, min_length=1),
amount: Money, reason: Optional[Reason] = None) -> dict:
"""Record that an error above the threshold was notified to HMRC on form VAT652: the reference, the date sent, the periods and
the amount. The notice is kept with the return; it is NOT on the return."""
def unit(repo):
vrn = _vrn(repo)
ret, _ = _load_return(repo, vrn, period_key)
ret.editable()
no = 1 + max((n["notice_no"] for n in repo.notices.where(vrn=vrn, period_key=period_key)), default=0)
repo.notices.put({"vrn": vrn, "period_key": period_key, "notice_no": no, "reference": reference,
"sent_on": uc._date(sent_on, "sent_on"), "periods": periods, "amount": amount, "reason": reason,
"recorded_by": uc._who(caller)})
return {"vrn": vrn, "period_key": period_key, "notice_no": no, "reference": reference, "sent_on": sent_on, "periods": periods,
"amount": amount}
return uc._atomic(unit)
# ═══ lock, reopen (TAX-28) ═══════════════════════════════════════════════════════════════════
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/lock", permission=RETURN_PREPARE, idempotent=True)
def lock_vat_return(caller: dict, period_key: Key) -> dict:
"""Lock a draft return for submission: the period has ended (its end is before today in the business time zone) and no
adjustment is above the threshold. Boxes and entries are frozen, and a new document with a tax point in the period is
refused (tax_point_open). Who locked it is remembered: with vat_return_needs_second_person they may not submit it."""
def unit(repo):
vrn, today = _vrn(repo), uc._today(repo)
ret, version = _load_return(repo, vrn, period_key)
if ret.lifecycle == vat_return.LOCKED:
return _return_view(repo, ret, version)
ret.editable()
ob = repo.obligations.get(vrn=vrn, period_key=period_key)
if ob["period_end"] >= today:
raise unmet("period_ended", f"the period ends on {ob['period_end']}: it cannot be locked before it has ended",
f"lock it from {ob['period_end'] + timedelta(days=1)}")
schemes.lock_gates(repo, vrn, ob)
late = schemes.unclaimed_rows(repo, vrn, ob)
if late:
refs = ", ".join(sorted({str(r["document_ref"]) for r in late})[:5])
raise unmet("return_stale", f"{len(late)} entries ({refs}) belong to this period but were not in the last prepare: locking "
"now would leave their VAT in no return", f"prepare {period_key} again, then lock")
small, large, pct = _thresholds(repo, today)
vat_return.check_error_threshold(_net_errors(repo.adjustments.where(vrn=vrn, period_key=period_key)),
ret.box6_total_sales_ex_vat, small, large, pct)
ret.lock(by=uc._who(caller), by_sub=uc._sub(caller), now=_utcnow())
return _return_view(repo, ret, repo.returns.save(ret, version))
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/reopen", permission=RETURN_PREPARE)
def reopen_vat_return(caller: dict, period_key: Key, reason: Reason) -> dict:
"""Reopen a LOCKED return to a draft, with a reason. A submitted return is never reopened: correct it on a later return."""
if not (reason or "").strip():
raise Refused(422, "reopening a return needs a reason")
def unit(repo):
ret, version = _load_return(repo, _vrn(repo), period_key)
ret.reopen()
return _return_view(repo, ret, repo.returns.save(ret, version), {"reopened_because": reason})
return uc._atomic(unit)
# ═══ submitting to HMRC (TAX-30, TAX-31) ═════════════════════════════════════════════════════
def _is_person(caller: dict) -> bool:
return bool(caller.get("email") or caller.get("sub")) and caller.get("auth_source") not in ("platform", "service")
def _finish(ret, receipt: dict, caller: dict) -> None:
ret.lifecycle, ret.submitted_by, ret.submitted_at = vat_return.SUBMITTED, uc._who(caller), _utcnow()
ret.declaration_text, ret.hmrc_error = DECLARATION, ""
ret.processing_date = str(receipt.get("processingDate") or "")
ret.form_bundle_number, ret.payment_indicator = str(receipt.get("formBundleNumber") or ""), str(receipt.get("paymentIndicator") or "")
ret.charge_ref_number = str(receipt.get("chargeRefNumber") or "")
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/submit", permission=RETURN_SUBMIT)
def submit_vat_return(caller: dict, period_key: Key, declaration: bool, client: Optional[dict[text(64, min_length=1), text(500)]] = None) -> dict:
"""Submit a LOCKED return to HMRC, as a PERSON: read the boxes, tick the legal declaration ("When you submit this VAT
information you are making a legal declaration that the information is true and complete. A false declaration can result in
prosecution.") and submit from the web screen, which passes `client` (the browser's own fraud-prevention headers). Refused
unless the HMRC connection is valid, the declaration is ticked, a second person submits (not the identity that locked it, by
email or sub) where vat_return_needs_second_person, every fraud-prevention header is present and HMRC still shows the
obligation open. HMRC's receipt is stored and the return becomes immutable; the settle journal posts to the ledger once. A
timeout leaves the return `submitting`: calling again first reads HMRC for the period and settles it, never filing twice."""
if not _is_person(caller):
raise Refused(403, "a VAT declaration is a person's: a service client cannot submit a return")
def prepare(repo):
vrn = _vrn(repo)
ret, version = _load_return(repo, vrn, period_key)
if ret.lifecycle == vat_return.SUBMITTED:
raise Refused(409, f"return {period_key} is already submitted and never changes")
if ret.lifecycle not in (vat_return.LOCKED, vat_return.SUBMITTING):
raise unmet("return_locked", f"return {period_key} is a {ret.lifecycle}: the tax admin locks it first", "VAT returns > lock")
conn = _connection(repo, vrn)
if declaration is not True:
raise unmet("declaration_ticked", "the legal declaration is not ticked", "read the boxes and tick the declaration")
settings = uc._settings_now(repo)
if settings.vat_return_needs_second_person and schedule.same_person(
uc._ids(caller), (ret.locked_by, ret.locked_by_sub)):
raise unmet("second_person", "the person who locked a return may not submit it: a second person submits",
"ask a second holder of tax:vat_return:submit")
headers = fraud.build(client, user_id=uc._sub(caller) or uc._who(caller), version=uc.TAX.version)
env, tokens = settings.hmrc_environment, _tokens(conn)
ob = repo.obligations.get(vrn=vrn, period_key=period_key)
if ret.lifecycle == vat_return.SUBMITTING: # a timeout last time: HMRC may have it
try:
if seams.hmrc.get_return(env, vrn, period_key, tokens, headers) is not None:
_finish(ret, {}, caller)
repo.obligations.put({**ob, "status": "F", "received_on": uc._today(repo)})
return None, _return_view(repo, ret, repo.returns.save(ret, version), {"settled_by_read": True})
except HmrcError as exc:
raise Refused(409, f"could not ask HMRC whether it has {period_key} ({exc.code}): not filing again until it can") from None
try:
held = [o for o in seams.hmrc.obligations(env, vrn, ob["period_start"], ob["period_end"], tokens, headers)
if o["periodKey"] == period_key]
except HmrcError as exc:
raise Refused(409, f"could not read HMRC's obligation for {period_key} ({exc.code})") from None
if held and held[0]["status"] == "F":
repo.obligations.put({**ob, "status": "F", "received_on": day(held[0].get("received")), "hmrc_error": "DUPLICATE_SUBMISSION"})
ret.hmrc_error = "DUPLICATE_SUBMISSION"
ret.lifecycle = vat_return.LOCKED
repo.returns.save(ret, version)
raise unmet("obligation_open", f"HMRC already shows {period_key} as filed (DUPLICATE_SUBMISSION): it is not sent again",
"the return is filed; record how and reconcile")
ret.lifecycle, ret.submitted_by = vat_return.SUBMITTING, uc._who(caller)
ret.fraud_headers_json = json.dumps(headers, sort_keys=True)
version = repo.returns.save(ret, version)
return (vrn, env, tokens, headers, ret.hmrc_body()), None
plan = uc._atomic(prepare)
if plan[0] is None:
return plan[1]
(vrn, env, tokens, headers, body), _ = plan
try:
receipt = seams.hmrc.submit_return(env, vrn, body, tokens, headers)
except HmrcTimeout:
raise Refused(504, f"HMRC did not answer: {period_key} is left submitting — call again, and HMRC is asked whether it has "
"the return before anything is sent") from None
except HmrcError as exc:
code, message = exc.code, exc.message
def failed(repo):
ret, version = _load_return(repo, vrn, period_key)
ret.lifecycle, ret.hmrc_error = vat_return.LOCKED, code
repo.returns.save(ret, version)
if code == "DUPLICATE_SUBMISSION":
ob = repo.obligations.get(vrn=vrn, period_key=period_key)
repo.obligations.put({**ob, "hmrc_error": code})
uc._atomic(failed)
raise Refused(409, f"HMRC refused the return ({code}): {message}" + ("; it is not retried" if code == "DUPLICATE_SUBMISSION" else "")) from None
def done(repo):
ret, version = _load_return(repo, vrn, period_key)
_finish(ret, receipt, caller)
ob = repo.obligations.get(vrn=vrn, period_key=period_key)
repo.obligations.put({**ob, "status": "F", "received_on": uc._today(repo)})
version = repo.returns.save(ret, version)
return _return_view(repo, ret, version)
out = uc._atomic(done)
try:
settled = uc._atomic(lambda repo: _settle(repo, vrn, period_key))
out["settle_journal_id"] = settled.get("settle_journal_id")
except Refused:
out["settle_journal_id"] = None # the ledger could not say: settle_vat_return retries
return out
# ═══ the settle journal and the liability view (TAX-32) ═══════════════════════════════════════
def _settle(repo, vrn: str, period_key: str) -> dict:
ret, version = _load_return(repo, vrn, period_key)
if ret.lifecycle != vat_return.SUBMITTED:
raise Refused(409, f"return {period_key} is {ret.lifecycle}: only a submitted return is settled")
if ret.settle_journal_id:
return _return_view(repo, ret, version)
per_code: dict = {}
for e in repo.entries.where(vrn=vrn, period_key=period_key, entry_kind="document"):
per_code[e["tax_code"]] = per_code.get(e["tax_code"], Decimal(0)) + Decimal(str(e["box1_amount"] or 0)) - Decimal(str(e["box4_amount"] or 0))
lines, total = [], Decimal(0)
for code in sorted(per_code):
net = per_code[code]
if net:
lines.append({"purpose": "tax", "tax_code": code, ("debit" if net > 0 else "credit"): str(abs(net))})
total += net
if not lines:
return _return_view(repo, ret, version) # a nil return: nothing to post
lines.append({"purpose": "vat_control", ("credit" if total > 0 else "debit"): str(abs(total))})
source = f"vat_return:{vrn}:{period_key}"
answer = seams.ledger_post.post_journal({"journal_id": str(derived_id(source, "settle")), "source_module": "tax", "source_document": source,
"event": "settle", "journal_date": ret.submitted_at.date().isoformat(), "currency": "GBP",
"lines": lines})
if not answer or not answer.get("journal_id"):
raise Refused(409, "the ledger could not post the settle journal: try settle_vat_return again")
ret.settle_journal_id = str(answer["journal_id"])
return _return_view(repo, ret, repo.returns.save(ret, version))
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/settle", permission=RETURN_PREPARE, idempotent=True)
def settle_vat_return(caller: dict, period_key: Key) -> dict:
"""Post a submitted return's settle journal to the ledger: each tax code's VAT against vat_control, the journal id derived
from the return, so it posts once (a repeat is the ledger's replay). Runs on submission; call it again if the ledger was down."""
return uc._atomic(lambda repo: _settle(repo, _vrn(repo), period_key))
def _due(period_end: date) -> date:
return vat_return.add_months(period_end, 1) + timedelta(days=7)
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/refresh-liability", permission=OBLIGATION_SYNC, idempotent=True)
def refresh_vat_liability(caller: dict, period_key: Key) -> dict:
"""Copy what HMRC holds for a submitted return's period — its liabilities and payments — keyed [vrn, period_key, charge_type]:
the amount, what is outstanding, what was paid and when. The daily clock-worker job while a liability is outstanding."""
def unit(repo):
vrn = _vrn(repo)
ret, _ = _load_return(repo, vrn, period_key)
if ret.lifecycle != vat_return.SUBMITTED:
raise Refused(409, f"return {period_key} is {ret.lifecycle}: HMRC holds a liability once it is submitted")
ob = repo.obligations.get(vrn=vrn, period_key=period_key)
conn = _connection(repo, vrn)
env, tokens = uc._settings_now(repo).hmrc_environment, _tokens(conn)
hi = vat_return.add_months(ob["period_end"], 12)
try:
liabilities = seams.hmrc.liabilities(env, vrn, ob["period_start"], hi, tokens, {})
payments = seams.hmrc.payments(env, vrn, ob["period_start"], hi, tokens, {})
except HmrcError as exc:
repo.connections.put({"vrn": vrn, "sync_error": exc.code, "sync_error_at": _utcnow()})
return {"period_key": period_key, "liabilities": [], "error": {"code": exc.code, "message": exc.message}}
paid_on = max((day(p.get("received")) for p in payments if p.get("received")), default=None)
n = 0
for lb in liabilities:
if day(lb.get("taxPeriod", {}).get("from")) != ob["period_start"]:
continue
original, outstanding = _money(lb["originalAmount"]), _money(lb.get("outstandingAmount", lb["originalAmount"]))
repo.liabilities.put({"vrn": vrn, "period_key": period_key, "charge_type": lb["type"], "original_amount": original,
"outstanding_amount": outstanding, "due_date": day(lb.get("due")) or _due(ob["period_end"]),
"paid_amount": original - outstanding, "paid_on": paid_on if original != outstanding else None,
"synced_at": _utcnow()})
n += 1
return {"period_key": period_key, "liabilities": n, "error": None}
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /vat-returns/{period_key}/liability", permission=RETURN_READ)
def get_vat_liability(caller: dict, period_key: Key) -> dict:
"""The liability view for a period: our return's net VAT, HMRC's liabilities and payments as HMRC holds them, the amount
outstanding and the due date (one month and seven days after the period end). A difference between our box 5 and HMRC's
liability is shown, never hidden."""
with uc.repository() as repo, uc._decide():
vrn = _vrn(repo)
ret, _ = _load_return(repo, vrn, period_key)
ob = repo.obligations.get(vrn=vrn, period_key=period_key)
rows = repo.liabilities.where(vrn=vrn, period_key=period_key)
ours = ret.box3_total_vat_due - ret.box4_vat_reclaimed
theirs = sum((Decimal(str(r["original_amount"])) for r in rows), Decimal(0))
view = [{"charge_type": r["charge_type"], "original_amount": r["original_amount"], "outstanding_amount": r["outstanding_amount"],
"paid_amount": r["paid_amount"], "paid_on": r["paid_on"], "due_date": r["due_date"]} for r in rows]
return {"period_key": period_key, "return_status": ret.lifecycle, "return_net_vat": ours, "hmrc_liabilities": view,
"outstanding": sum((Decimal(str(r["outstanding_amount"])) for r in rows), Decimal(0)),
"due_date": _due(ob["period_end"]), "hmrc_original_total": theirs if rows else None,
"difference": (ours - theirs) if rows else None, "differs": bool(rows) and ours != theirs}
@TAX.use_case(kind="query", route="GET /vat-returns/liabilities", permission=RETURN_READ)
def list_vat_liabilities(caller: dict) -> dict:
"""The VAT payments due: every period with an outstanding liability as HMRC holds it, with its due date (payments matches an
HMRC debit on the bank statement to the liability it pays). CIS payments due are listed beside them."""
with uc.repository() as repo, uc._decide():
vrn = _vrn(repo)
rows = [r for r in repo.liabilities.where(vrn=vrn) if Decimal(str(r["outstanding_amount"] or 0)) > 0]
return {"liabilities": [{"period_key": r["period_key"], "charge_type": r["charge_type"],
"outstanding_amount": r["outstanding_amount"], "due_date": r["due_date"]} for r in rows],
"cis_due": []}
# ═══ which return holds a tax point (TAX-33) ═════════════════════════════════════════════════
@TAX.use_case(kind="query", route="GET /vat-returns/tax-point-status", permission=TAX_POINT_READ)
def tax_point_status(caller: dict, date: IsoDate) -> dict:
"""ar's question before it issues or voids a document: which VAT return holds this tax point, and may a document still be
issued with it? A date in a submitted return answers filed true and the gate tax_point_open met=false (credit it instead, or
issue with today's tax point); a locked return is closed for a new tax point too; a draft or open period accepts it; a date in
no HMRC obligation answers period_key null and met."""
on = uc._date(date, "date")
with uc.repository() as repo, uc._decide():
vrn = _vrn(repo)
ob = next((o for o in repo.obligations.where(vrn=vrn) if o["period_start"] <= on <= o["period_end"]), None)
out = {"date": on, "period_key": None, "status": "none", "filed": False}
if ob is None:
return out | {"gate": {"gate": "tax_point_open", "met": True, "reason": "no VAT period holds this date", "fix": ""}}
found = repo.returns.get(vrn, ob["period_key"])
ret = found[0] if found else None
out["period_key"] = ob["period_key"]
if ret is not None and ret.lifecycle == vat_return.SUBMITTED:
when = ret.submitted_at.date() if ret.submitted_at else ob["received_on"]
out |= {"status": "submitted", "filed": True}
gate = Gate("tax_point_open", False, f"{on} is in VAT return {ob['period_key']}, submitted {when}",
"credit it instead, or issue with today's tax point")
elif ret is None and ob["status"] == "F": # filed elsewhere: HMRC says so
out |= {"status": "submitted", "filed": True}
gate = Gate("tax_point_open", False, f"{on} is in VAT period {ob['period_key']}, which HMRC shows as filed ({ob['received_on']})",
"credit it instead, or issue with today's tax point")
elif ret is not None and ret.lifecycle in (vat_return.LOCKED, vat_return.SUBMITTING):
out |= {"status": "locked"}
gate = Gate("tax_point_open", False, f"return {ob['period_key']} is locked for submission", "issue with today's tax point")
else:
out |= {"status": "draft" if ret is not None else "open"}
gate = Gate("tax_point_open", True, f"{on} is in VAT period {ob['period_key']}, which is open", "")
return out | {"gate": gate.as_dict()}
# ═══ bad debt relief (TAX-34, TAX-35) ════════════════════════════════════════════════════════
def _vat_accounted(repo, invoice_id: str) -> bool:
for e in repo.entries.where(document_ref=invoice_id, entry_kind="document"):
found = repo.returns.get(e["vrn"], e["period_key"])
if found and found[0].lifecycle == vat_return.SUBMITTED:
return True
return False
def _claim_from(repo, w: dict) -> Optional[bad_debt.Claim]:
"""A claim row from ar's approved write-off and the invoice's frozen tax; None when this brick holds no GBP sale tax for it."""
found = repo.documents.get(w["invoice_id"])
if found is None or found[0].direction != "sale" or found[0].currency != "GBP" or found[0].status != "frozen":
return None
doc, c = found[0], found[0].computed()
return bad_debt.Claim(w["invoice_id"], bad_debt.ELIGIBLE, doc.bp_id, str(w.get("writeoff_id") or ""),
day(w.get("tax_point_date")) or doc.tax_point, day(w.get("due_date")), None,
day(str(w.get("approved_at") or "")[:10]), c.gross, c.vat)
def _suggest_bad_debt(repo) -> list:
"""List as eligible every written-off invoice six months overdue whose VAT is in a submitted return, one row per invoice."""
today, listed = uc._today(repo), []
for w in seams.receivables.writeoffs(EPOCH) or []:
if repo.claims.get(w["invoice_id"]) is not None:
continue
claim = _claim_from(repo, w)
if claim is None:
continue
claim.eligible_from = vat_return.add_months(max(d for d in (claim.due_date, claim.supply_date) if d), 6)
gates = claim.gates(today=today, vat_accounted=_vat_accounted(repo, claim.invoice_id), written_off=True)
if all(g.met for g in gates):
repo.claims.save(claim, 0)
listed.append(claim.invoice_id)
return listed
def _claim_view(c: bad_debt.Claim, gates: Optional[list] = None) -> dict:
out = {"invoice_id": c.invoice_id, "lifecycle": c.lifecycle, "bp_id": c.bp_id, "supply_date": c.supply_date, "due_date": c.due_date,
"eligible_from": c.eligible_from, "written_off_on": c.written_off_on, "invoice_total": c.gross, "invoice_vat": c.vat,
"unpaid_at_claim": c.unpaid_at_claim, "relief_amount": c.relief_amount, "claimed_in_period_key": c.claimed_in_period_key or None,
"claimed_by": c.claimed_by or None, "claimed_on": c.claimed_on, "dismissed_reason": c.dismissed_reason or None}
return out | ({"gates": [g.as_dict() for g in gates]} if gates is not None else {})
@TAX.use_case(kind="command", route="POST /bad-debt/suggest", permission=BAD_DEBT_SUGGEST, idempotent=True)
def suggest_bad_debt_relief(caller: dict) -> dict:
"""List the invoices bad debt relief can be claimed on: written off in receivables (an approved ar write-off), six months past
the later of the due and supply dates, with their VAT in a submitted return. One row per invoice, once; also runs when a
return is prepared."""
return uc._atomic(lambda repo: {"eligible": _suggest_bad_debt(repo)})
@TAX.use_case(kind="query", route="GET /bad-debt", permission=RETURN_READ)
def list_bad_debt_claims(caller: dict) -> dict:
"""The invoices listed for bad debt relief (eligible, claimed, dismissed) with each eligibility gate {gate, met, reason, fix}."""
with uc.repository() as repo, uc._decide():
today = uc._today(repo)
return {"claims": [_claim_view(c, c.gates(today=today, vat_accounted=_vat_accounted(repo, c.invoice_id), written_off=bool(c.writeoff_id)))
for c in repo.claims.all()]}
@TAX.use_case(kind="command", route="POST /bad-debt/{invoice_id}/claim", permission=BAD_DEBT_CLAIM)
def claim_bad_debt_relief(caller: dict, invoice_id: text(200, min_length=1), period_key: Key) -> dict:
"""Claim bad debt relief on a written-off invoice in a DRAFT return: the VAT fraction of the unpaid consideration —
(total - received) x VAT / total, the unpaid amount read from payments as at today — goes into box 4 naming the invoice.
Every gate must be met: six_months_passed (from the later of the due and supply dates), written_off, vat_accounted (the
invoice's VAT is in a submitted return), within_time_limit (four years six months), not_claimed. One claim per invoice."""
def unit(repo):
vrn, today = _vrn(repo), uc._today(repo)
ret, rversion = _load_return(repo, vrn, period_key)
ret.editable()
found = repo.claims.get(invoice_id)
if found is None:
w = next((x for x in seams.receivables.writeoffs(EPOCH) or [] if x["invoice_id"] == invoice_id), None)
claim = _claim_from(repo, w or {"invoice_id": invoice_id})
if claim is None:
raise unmet("invoice_tax_known", f"this brick holds no frozen sale tax for {invoice_id}", "the invoice must be issued with its tax frozen here")
if w is None:
claim.writeoff_id = ""
found = (claim, 0)
claim, version = found
if claim.eligible_from is None:
claim.eligible_from = (vat_return.add_months(max(d for d in (claim.due_date, claim.supply_date) if d), 6)
if (claim.due_date or claim.supply_date) else None)
accounted, written = _vat_accounted(repo, invoice_id), bool(claim.writeoff_id)
bad = [g for g in claim.gates(today=today, vat_accounted=accounted, written_off=written) if not g.met]
if bad:
raise GateUnmet(bad)
settled = seams.payments.settlement([invoice_id], today)
row = next((i for i in (settled or {}).get("invoices", []) if i["invoice_id"] == invoice_id), None)
if row is None:
raise unmet("unpaid_known", f"payments cannot say what is unpaid on {invoice_id}", "payments must know the invoice")
relief = claim.claim(unpaid=_money(row["outstanding"]), period_key=period_key, by=uc._who(caller), today=today,
vat_accounted=accounted, written_off=written)
repo.claims.save(claim, version)
if not repo.entries.claim({"document_ref": invoice_id, "tax_code": "-", "entry_kind": "bad_debt_claim", "vrn": vrn, "period_key": period_key,
"direction": "sale", "tax_point": today, "net_amount": Decimal(0), "vat_amount": relief,
"vat_reclaimable": relief, "box1_amount": Decimal(0), "box4_amount": relief, "box_value": None,
"source": "bad_debt_claim", "detail": f"relief on {invoice_id}"}):
raise Refused(409, f"relief on {invoice_id} is already in a return")
_recompute(repo, ret)
return _claim_view(claim) | {"return": _return_view(repo, ret, repo.returns.save(ret, rversion))}
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /bad-debt/{invoice_id}/dismiss", permission=BAD_DEBT_CLAIM)
def dismiss_bad_debt_relief(caller: dict, invoice_id: text(200, min_length=1), reason: Reason) -> dict:
"""Dismiss an eligible invoice: no relief will be claimed, with a reason."""
def unit(repo):
found = repo.claims.get(invoice_id)
if found is None:
raise Refused(404, f"{invoice_id} is not listed for bad debt relief")
claim, version = found
claim.dismiss(reason)
repo.claims.save(claim, version)
return _claim_view(claim)
return uc._atomic(unit)