dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/cis.py
"""The Construction Industry Scheme (v4: TAX-38 .. TAX-41). A contractor verifies each subcontractor with HMRC, deducts the HMRC
rate from the LABOUR only (materials excluded) when purchasing pays, files the CIS300 for each tax month (6th to 5th) and sends
each subcontractor a statement. The other direction — a contractor customer deducted CIS from what they paid us — is kept per tax
month so the deduction suffered is claimed back.
tax_cis_subcontractor identity at FOREVER ('unverified'), then each verification from a date (a new row, never an edit)
tax_cis_deduction one row per [payment_ref, bp_id]: a retry returns the same deduction; a reversal marks it
tax_cis_return the CIS300 of one tax month: open -> prepared -> submitted
tax_cis_suffered what a contractor deducted from us, per [contractor bp_id, statement_ref]
"""
import json
from datetime import date, timedelta
from decimal import ROUND_HALF_UP, Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.repository import Conflict
from erp_platform.types import BigId, IsoDate, text
from sqlalchemy.exc import IntegrityError
from brick_tax.application import returns as R
from brick_tax.application import use_cases as uc
from brick_tax.application.inputs import Money, Reason
from brick_tax.application.use_cases import (
CIS_DEDUCT,
CIS_READ,
CIS_SUBMIT,
CIS_SUFFERED,
CIS_WRITE,
TAX,
)
from brick_tax.domain.errors import Gate, unmet
from brick_tax.domain.values import FOREVER
from brick_tax.infrastructure import seams
from brick_tax.infrastructure.hmrc import HmrcError
P2 = Decimal("0.01")
STATUSES = ("gross", "net", "unmatched")
RATE_KEY = {"gross": "cis_rate_gross", "net": "cis_rate_net", "unmatched": "cis_rate_unmatched"}
def tax_month_end(d: date) -> date:
"""A tax month runs from the 6th to the 5th: the month of a date ends on the 5th on or after it."""
if d.day <= 5:
return date(d.year, d.month, 5)
return date(d.year + (d.month == 12), d.month % 12 + 1, 5)
def tax_year(d: date) -> int:
return d.year if (d.month, d.day) >= (4, 6) else d.year - 1
def _month(label: str) -> date:
try:
return date.fromisoformat(label)
except ValueError:
raise Refused(422, "a tax month is named by the date it ends, such as 2026-11-05") from None
def _verified_on(repo, bp_id: int, on: date) -> Optional[dict]:
rows = [r for r in repo.cis_subs.where(bp_id=bp_id) if r["valid_from"] <= on and r["status"] in STATUSES]
return rows[-1] if rows else None
def _paid_recently(repo, bp_id: int, on: date) -> bool:
"""Paid in this or the previous two tax years: no fresh verification is needed."""
return any(tax_year(on) - 2 <= tax_year(r["paid_on"]) <= tax_year(on) for r in repo.cis_deductions.where(bp_id=bp_id) if r["state"] != "reversed")
def _readiness(repo, sub: dict, on: date) -> list:
ok = _verified_on(repo, sub["bp_id"], on) is not None or _paid_recently(repo, sub["bp_id"], on)
return [Gate("cis_verified", ok, "" if ok else f"{sub['name'] or 'the subcontractor'} has not been verified with HMRC",
"" if ok else "CIS > Subcontractors > Verify").as_dict()]
def _identity(repo, bp_id: int) -> Optional[dict]:
return repo.cis_subs.get(bp_id=bp_id, valid_from=FOREVER)
def _sub_view(repo, bp_id: int, on: date) -> dict:
ident = _identity(repo, bp_id)
if ident is None:
raise Refused(404, f"{bp_id} is not a CIS subcontractor")
v = _verified_on(repo, bp_id, on)
return {"bp_id": bp_id, "name": ident["name"], "utr": ident["utr"], "verification_number": v["verification_number"] if v else None,
"status": v["status"] if v else None, "rate_pct": v["rate_pct"] if v else None, "verified_on": v["valid_from"] if v else None,
"readiness": _readiness(repo, ident, on)}
# ═══ the subcontractor (TAX-38) ═════════════════════════════════════════════════════════════════
@TAX.use_case(kind="command", route="POST /cis/subcontractors", permission=CIS_WRITE)
def add_cis_subcontractor(caller: dict, bp_id: BigId, name: text(200, min_length=1), utr: text(20, min_length=10),
company_number: Optional[text(20)] = None, ni_number: Optional[text(20)] = None) -> dict:
"""Make a business partner a CIS subcontractor: it needs a UTR and a company number or a National Insurance number. It starts
unverified; verify it with HMRC before the first payment."""
def unit(repo):
if not (company_number or ni_number):
raise unmet("cis_identity", "a subcontractor is matched by UTR and a company number or a National Insurance number",
"CIS > Subcontractors > the company or NI number")
if _identity(repo, int(bp_id)) is None:
repo.cis_subs.put({"bp_id": int(bp_id), "valid_from": FOREVER, "name": name, "utr": utr.strip(),
"company_number": company_number or "", "ni_number": ni_number or "", "status": "unverified",
"recorded_by": uc._who(caller)})
return _sub_view(repo, int(bp_id), uc._today(repo))
return uc._atomic(unit)
def _verify(caller: dict, bp_id: int, given_status: Optional[str], given_number: Optional[str], again: bool) -> dict:
def unit(repo):
status, verification_number = given_status, given_number
ident = _identity(repo, bp_id)
if ident is None:
raise Refused(404, f"{bp_id} is not a CIS subcontractor: add it first")
today = uc._today(repo)
if again and _verified_on(repo, bp_id, today) is None:
raise Refused(409, f"{bp_id} has no verification to renew: verify it first")
if repo.cis_subs.get(bp_id=bp_id, valid_from=today) is not None:
raise Refused(409, f"{bp_id} was already verified today: a row is never edited — re-verify from tomorrow")
source = "person"
if status is not None: # a person read the result from HMRC's CIS service
if status not in STATUSES or not (verification_number or "").strip():
raise unmet("cis_verification_number", "a result read from HMRC is entered with its status (gross, net or unmatched) and "
"the verification number", "CIS > Subcontractors > Verify")
else:
vrn = R._vrn(repo)
conn = R._connection(repo, vrn)
try:
got = seams.hmrc.verify_subcontractor(uc._settings_now(repo).hmrc_environment, {
"contractor_utr": uc._settings_now(repo).cis_contractor_utr, "utr": ident["utr"], "company_number": ident["company_number"],
"ni_number": ident["ni_number"], "name": ident["name"]}, R._tokens(conn), {})
except HmrcError as exc:
raise unmet("hmrc_verification", f"HMRC could not verify the subcontractor: {exc.code} {exc.message}".strip(),
"read the result from HMRC's CIS service and enter it with its verification number") from None
status, verification_number, source = got["status"], got["verification_number"], "hmrc"
rate = uc_stat(repo, RATE_KEY[status], today)
row = {"bp_id": bp_id, "valid_from": today, "name": ident["name"], "utr": ident["utr"],
"company_number": ident["company_number"], "ni_number": ident["ni_number"], "status": status, "rate_pct": rate,
"verification_number": verification_number, "verified_source": source, "recorded_by": uc._who(caller)}
try: # check (above), then a plain INSERT: a racer that passed the same check collides
repo.conn.execute(repo.cis_subs.table.insert().values(**row))
except IntegrityError:
raise Conflict("tax_cis_subcontractor", {"bp_id": bp_id, "valid_from": today}) from None
return _sub_view(repo, bp_id, today)
return uc._atomic(unit)
def uc_stat(repo, key: str, on: date) -> Decimal:
return R._stat(repo, key, on)
@TAX.use_case(kind="command", route="POST /cis/subcontractors/{bp_id}/verify", permission=CIS_WRITE)
def verify_cis_subcontractor(caller: dict, bp_id: BigId, status: Optional[text(10)] = None,
verification_number: Optional[text(40)] = None) -> dict:
"""Verify a subcontractor with HMRC (through the HMRC connection), or enter the result a person read from HMRC's CIS service:
`status` gross | net | unmatched with its `verification_number`. Stores the status, the rate 0 | 20 | 30 (dated statutory
rows), the verification number and today's date."""
return _verify(caller, int(bp_id), status, verification_number, False)
@TAX.use_case(kind="command", route="POST /cis/subcontractors/{bp_id}/reverify", permission=CIS_WRITE)
def reverify_cis_subcontractor(caller: dict, bp_id: BigId, status: Optional[text(10)] = None,
verification_number: Optional[text(40)] = None) -> dict:
"""A new dated verification row for an already verified subcontractor: deductions from then on use it (never an edit)."""
return _verify(caller, int(bp_id), status, verification_number, True)
@TAX.use_case(kind="query", route="GET /cis/subcontractors/{bp_id}", permission=CIS_READ)
def get_cis_subcontractor(caller: dict, bp_id: BigId, on: Optional[IsoDate] = None) -> dict:
"""A subcontractor's CIS status on the pay date, for purchasing's cis_verified gate and the rate it proposes: {bp_id, utr,
verification_number, status, rate_pct, verified_on, readiness}. 404 when the partner is not a CIS subcontractor."""
with uc.repository() as repo:
return _sub_view(repo, int(bp_id), uc._date(on, "on") or uc._today(repo))
# ═══ the deduction on a payment (TAX-39) ════════════════════════════════════════════════════════
def _deduction_view(r: dict, gates: Optional[list] = None, replay: bool = False) -> dict:
return {"payment_ref": r["payment_ref"], "bp_id": r["bp_id"], "tax_month": r["tax_month"], "status": r["status"],
"rate_pct": r["rate_pct"], "labour": r["labour"], "deduction": r["deduction"], "materials": r["materials"],
"gross_ex_vat": r["gross_ex_vat"], "state": r["state"], "gates": gates or [], "replay": replay}
@TAX.use_case(kind="command", route="POST /cis/deductions", permission=CIS_DEDUCT, idempotent=True)
def compute_cis_deduction(caller: dict, payment_ref: text(200, min_length=1), bp_id: BigId, paid_on: IsoDate,
bill_ids: list[text(200, min_length=1)], gross_ex_vat: Money, materials: Money) -> dict:
"""Purchasing asks for the CIS deduction on a payment to a subcontractor: HMRC's rate x (gross excluding VAT less the
materials), booked to the keyed row [payment_ref, bp_id] in the tax month (6th to 5th) of the payment date, so a retry adds
nothing. Unverified (and not paid in this or the last two tax years) answers {gate: cis_verified}; materials above the bill's own
figure are refused; an unmatched subcontractor is deducted at 30%, a gross one at 0% (the row is still kept for the return)."""
def unit(repo):
key = {"payment_ref": payment_ref, "bp_id": int(bp_id)}
have = repo.cis_deductions.get(**key)
if have is not None:
return _deduction_view(have, replay=True)
ident = _identity(repo, int(bp_id))
if ident is None:
raise Refused(404, f"{bp_id} is not a CIS subcontractor")
when = uc._date(paid_on, "paid_on")
if materials < 0 or gross_ex_vat < 0 or materials > gross_ex_vat:
raise Refused(422, f"materials {materials} are above the bill's own figure {gross_ex_vat}: the deduction is on the labour only")
month = tax_month_end(when).isoformat()
ret = repo.cis_returns.get(tax_month=month)
if ret is not None and ret["lifecycle"] == "submitted":
g = Gate("cis_return_open", False, f"the CIS return for the tax month ending {month} is submitted",
"pay in the next tax month, or correct it in the next month's return")
raise Refused(409, f"{g.reason} — unmet gates: {json.dumps([g.as_dict()])}")
gates = _readiness(repo, ident, when)
if not gates[0]["met"]:
raise Refused(422, f"{gates[0]['reason']} — unmet gates: {json.dumps([dict(gates[0])])}")
v = _verified_on(repo, int(bp_id), when)
status = v["status"] if v else "unmatched"
rate = Decimal(str(v["rate_pct"])) if v else uc_stat(repo, RATE_KEY["unmatched"], when)
labour = gross_ex_vat - materials
row = {**key, "tax_month": tax_month_end(when).isoformat(), "paid_on": when, "bill_ids": ",".join(bill_ids),
"gross_ex_vat": gross_ex_vat, "materials": materials, "labour": labour, "status": status, "rate_pct": rate,
"deduction": (labour * rate / Decimal(100)).quantize(P2, ROUND_HALF_UP), "state": "computed"}
try: # check (above), then a plain INSERT: a racer collides -> retry replays
repo.conn.execute(repo.cis_deductions.table.insert().values(**row))
except IntegrityError:
raise Conflict("tax_cis_deduction", key) from None
return _deduction_view(row, gates)
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /cis/deductions/{payment_ref}/reverse", permission=CIS_DEDUCT, idempotent=True)
def reverse_cis_deduction(caller: dict, payment_ref: text(200, min_length=1), bp_id: BigId, reason: Reason, reversed_on: IsoDate) -> dict:
"""Purchasing rejected the payment: the deduction no longer counts in the month's return or the subcontractor's statement.
A repeat answers the same. Refused ({gate: cis_return_open}, 409) once the tax month is submitted: correct it in the next return."""
def unit(repo):
row = repo.cis_deductions.get(payment_ref=payment_ref, bp_id=int(bp_id))
if row is None:
raise Refused(404, f"no CIS deduction for payment {payment_ref} and subcontractor {bp_id}")
if row["state"] == "reversed":
return {"payment_ref": payment_ref, "bp_id": int(bp_id), "status": "reversed", "tax_month": row["tax_month"]}
ret = repo.cis_returns.get(tax_month=row["tax_month"])
if ret is not None and ret["lifecycle"] == "submitted":
g = Gate("cis_return_open", False, f"the CIS return for the tax month ending {row['tax_month']} is submitted",
"correct it in the next month's return")
raise Refused(409, f"{g.reason} — unmet gates: {json.dumps([g.as_dict()])}")
repo.conn.execute(repo.cis_deductions.table.update().where(
repo.cis_deductions.table.c.payment_ref == payment_ref, repo.cis_deductions.table.c.bp_id == int(bp_id)).values(
state="reversed", reversed_on=uc._date(reversed_on, "reversed_on"), reversed_reason=reason))
return {"payment_ref": payment_ref, "bp_id": int(bp_id), "status": "reversed", "tax_month": row["tax_month"]}
return uc._atomic(unit)
# ═══ the monthly return and the statements (TAX-40) ═════════════════════════════════════════════
def _lines(repo, month: str) -> list:
by: dict = {}
for r in repo.cis_deductions.where(tax_month=month):
if r["state"] == "reversed":
continue
s = by.setdefault(r["bp_id"], {"bp_id": r["bp_id"], "gross_ex_vat": Decimal(0), "materials": Decimal(0), "deduction": Decimal(0)})
s["gross_ex_vat"] += Decimal(str(r["gross_ex_vat"]))
s["materials"] += Decimal(str(r["materials"]))
s["deduction"] += Decimal(str(r["deduction"]))
out = []
for bp, s in sorted(by.items()):
ident = _identity(repo, bp)
v = _verified_on(repo, bp, _month(month))
out.append({**s, "name": ident["name"], "utr": ident["utr"], "verification_number": v["verification_number"] if v else "",
"status": v["status"] if v else "unmatched"})
return out
def _return_view(row: dict, today: date) -> dict:
end = _month(row["tax_month"])
due = end + timedelta(days=14)
lines = json.loads(row["lines_json"] or "[]")
gates = [] if row["lifecycle"] == "submitted" or today <= due else [Gate(
"cis300_on_time", False, f"the CIS300 for the tax month ending {end} was due on {due}: filing late brings automatic penalties",
"file it now").as_dict()]
return {"tax_month": row["tax_month"], "lifecycle": row["lifecycle"],
"nil_return": bool(row["nil_return"]), "lines": lines, "total_deduction": row["total_deduction"], "due_date": due,
"pay_hmrc_by": end + timedelta(days=17), "gates": gates, "receipt": json.loads(row["receipt_json"]) if row["receipt_json"] else None,
"statements_sent_at": row["statements_sent_at"]}
def _get_return(repo, month: str) -> dict:
row = repo.cis_returns.get(tax_month=month)
if row is None:
raise Refused(404, f"no CIS return for the tax month ending {month}: prepare it first")
return row
@TAX.use_case(kind="command", route="POST /cis/returns/{tax_month}/prepare", permission=CIS_WRITE)
def prepare_cis_return(caller: dict, tax_month: text(10, min_length=10)) -> dict:
"""Prepare the CIS300 of a finished tax month (named by the date it ends, such as 2026-11-05): one line per subcontractor
with name, UTR, verification number, gross paid excluding VAT, materials and deduction, from the month's deductions; a month
with none is a nil return. Needs our contractor UTR (setting cis_contractor_utr). A submitted return is never changed."""
def unit(repo):
end, today = _month(tax_month), uc._today(repo)
if tax_month_end(end) != end:
raise Refused(422, "a tax month ends on the 5th")
if end >= today:
raise unmet("tax_month_ended", f"the tax month ends on {end}: it can be prepared from {end + timedelta(days=1)}", "wait for the 6th")
if not uc._settings_now(repo).cis_contractor_utr:
raise unmet("cis_contractor_utr", "our UTR as a CIS contractor is not set", "Tax settings > cis_contractor_utr")
have = repo.cis_returns.get(tax_month=tax_month)
if have is not None and have["lifecycle"] == "submitted":
raise Refused(409, f"the CIS return for {tax_month} is submitted and never changes")
lines = _lines(repo, tax_month)
row = {"tax_month": tax_month, "lifecycle": "prepared", "nil_return": not lines, "total_deduction": sum((x["deduction"] for x in lines), Decimal(0)),
"lines_json": json.dumps(lines, default=str), "prepared_by": uc._who(caller)}
repo.cis_returns.put(row)
return _return_view(repo.cis_returns.get(tax_month=tax_month), today)
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /cis/returns/{tax_month}", permission=CIS_WRITE)
def get_cis_return(caller: dict, tax_month: text(10, min_length=10)) -> dict:
"""The CIS300 of a tax month as prepared: its lines, the deductions total (what the business pays HMRC by the 22nd with its
PAYE), the due date and an overdue warning."""
with uc.repository() as repo:
return _return_view(_get_return(repo, tax_month), uc._today(repo))
@TAX.use_case(kind="command", route="POST /cis/returns/{tax_month}/submit", permission=CIS_SUBMIT)
def submit_cis_return(caller: dict, tax_month: text(10, min_length=10), employment_status_declared: bool, verification_declared: bool) -> dict:
"""Submit the CIS300 to HMRC over the HMRC connection once a PERSON has confirmed both declarations (the subcontractors'
employment status; that every subcontractor was verified). HMRC's receipt is kept and the return is immutable. Late filing
answers an overdue gate with the penalty warning but is still allowed."""
if not R._is_person(caller):
raise Refused(403, "a CIS declaration is made by a person, never a service client")
if not (employment_status_declared and verification_declared):
raise Refused(422, "both declarations (employment status, verification) must be confirmed by a person")
def unit(repo):
row = _get_return(repo, tax_month)
if row["lifecycle"] == "submitted":
return _return_view(row, uc._today(repo))
vrn = R._vrn(repo)
conn = R._connection(repo, vrn)
try:
receipt = seams.hmrc.submit_cis300(uc._settings_now(repo).hmrc_environment, {
"contractor_utr": uc._settings_now(repo).cis_contractor_utr, "tax_month_end": tax_month, "nil_return": bool(row["nil_return"]),
"subcontractors": json.loads(row["lines_json"] or "[]"), "employment_status_declared": True, "verification_declared": True},
R._tokens(conn), {})
except HmrcError as exc:
raise unmet("hmrc_cis_filing", f"HMRC did not take the CIS300: {exc.code} {exc.message}".strip(),
"file it in HMRC's CIS online service and record the receipt") from None
repo.cis_returns.put({**{k: v for k, v in row.items() if k not in ("created_at", "updated_at")}, "lifecycle": "submitted",
"submitted_by": uc._who(caller), "submitted_at": R._utcnow(), "receipt_json": json.dumps(receipt, default=str),
"declaration_text": "employment status and verification declarations confirmed by a person"})
return _return_view(repo.cis_returns.get(tax_month=tax_month), uc._today(repo))
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /cis/returns/{tax_month}/send-statements", permission=CIS_SUBMIT, idempotent=True)
def send_cis_statements(caller: dict, tax_month: text(10, min_length=10)) -> dict:
"""After submission: a payment and deduction statement per subcontractor (gross paid, materials, labour, deduction), due to
them by the 19th. The send is recorded; delivery is the composition's mail channel."""
def unit(repo):
row = _get_return(repo, tax_month)
if row["lifecycle"] != "submitted":
raise unmet("cis_return_submitted", "statements follow the submitted CIS300", "submit the CIS return first")
statements = [{"bp_id": x["bp_id"], "name": x["name"], "utr": x["utr"], "tax_month_end": tax_month, "gross_ex_vat": x["gross_ex_vat"],
"materials": x["materials"], "labour": str(Decimal(x["gross_ex_vat"]) - Decimal(x["materials"])), "deducted": x["deduction"]}
for x in json.loads(row["lines_json"] or "[]")]
if row["statements_sent_at"] is None:
repo.cis_returns.put({**{k: v for k, v in row.items() if k not in ("created_at", "updated_at")}, "statements_sent_at": R._utcnow(),
"statements_json": json.dumps(statements)})
return {"tax_month": tax_month, "statements": statements, "sent_at": (repo.cis_returns.get(tax_month=tax_month))["statements_sent_at"]}
return uc._atomic(unit)
# ═══ what a contractor deducted from us (TAX-41) ════════════════════════════════════════════════
@TAX.use_case(kind="command", route="POST /cis/suffered", permission=CIS_SUFFERED, idempotent=True)
def record_cis_suffered(caller: dict, contractor_bp_id: BigId, statement_ref: text(200, min_length=1), tax_month: text(10, min_length=10),
invoice_id: text(200, min_length=1), payment_id: text(200, min_length=1), gross: Money, materials: Money,
deduction: Money) -> dict:
"""Payments tells us what a contractor customer deducted under CIS from our invoice, from their payment and deduction
statement: kept per [contractor, statement_ref] with the tax month, gross, materials and deduction, so the deduction suffered is
claimed back (PAYE offset, or self assessment). A deduction above 30% of the labour named is refused ({gate: within_cis_rate});
the same statement twice adds nothing."""
def unit(repo):
key = {"contractor_bp_id": int(contractor_bp_id), "statement_ref": statement_ref}
if repo.cis_suffered.get(**key) is not None:
return {"recorded": True, "replay": True}
try:
month_end = date.fromisoformat(tax_month)
except ValueError:
raise Refused(422, f"tax_month {tax_month} is not a date (YYYY-MM-DD)") from None
if month_end.day != 5:
raise Refused(422, f"tax_month {tax_month} is not the end of a tax month: a CIS tax month ends on the 5th")
labour = gross - materials
if labour < 0 or deduction < 0 or deduction > (labour * Decimal("0.30")).quantize(P2, ROUND_HALF_UP):
g = Gate("within_cis_rate", False, f"a deduction of {deduction} is above 30% of the labour {labour}: CIS never deducts more",
"check the contractor's statement")
raise Refused(422, f"{g.reason} — unmet gates: {json.dumps([g.as_dict()])}")
try:
repo.conn.execute(repo.cis_suffered.table.insert().values(**key, tax_month=tax_month, invoice_id=invoice_id, payment_id=payment_id,
gross=gross, materials=materials, deduction=deduction))
except IntegrityError:
raise Conflict("tax_cis_suffered", key) from None
return {"recorded": True, "replay": False}
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /cis/suffered", permission=CIS_WRITE)
def cis_suffered_view(caller: dict) -> dict:
"""What contractors deducted from us, summed per tax month and per tax year (for the PAYE offset or self assessment)."""
with uc.repository() as repo:
months: dict = {}
years: dict = {}
for r in repo.cis_suffered.where():
d = Decimal(str(r["deduction"]))
months[r["tax_month"]] = months.get(r["tax_month"], Decimal(0)) + d
y = f"{tax_year(_month(r['tax_month']))}/{tax_year(_month(r['tax_month'])) + 1}"
years[y] = years.get(y, Decimal(0)) + d
return {"per_tax_month": months, "per_tax_year": years}