dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/domain/registration.py
"""VAT registration: whether, and from when, this business is registered (TAX-01). Dated rows, never edited: the status
on a tax point is the latest APPROVED row that started on or before it. "Not registered" is a real state: such a business
charges no VAT and its document may not call itself a VAT invoice.
Workflow (module.yaml): record_not_registered / record_registered (the first row, any date: the HMRC effective date);
register (not_registered -> registered, a VAT number, after today, a second person where the control asks); deregister
(registered -> deregistered, after today, ALWAYS a second person).
"""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_tax.domain import schedule
from brick_tax.domain.errors import RuleBroken, unmet
from brick_tax.domain.values import ChangeState, RegStatus, Scheme, clean_vat_number
@dataclass
class Registration:
jurisdiction: str
effective_from: date
status: str
vat_number: str = ""
scheme: str = Scheme.STANDARD.value
effective_to: Optional[date] = None
change_state: str = ChangeState.APPROVED.value
created_by: str = ""
approved_by: str = ""
created_sub: str = ""
accounting_basis: str = "invoice" # v4: invoice | cash
flat_rate_sector: str = ""
flat_rate_pct: Optional[Decimal] = None
first_year_until: Optional[date] = None
ni_goods: bool = False # an XI registration trading goods with the EU
partial_exemption_year_end: str = "" # MM-DD
expected_turnover: Optional[Decimal] = None
def live(rows) -> list:
return sorted((r for r in rows if r.change_state != ChangeState.WITHDRAWN.value), key=lambda r: r.effective_from)
def status_on(rows, on: date) -> Optional[Registration]:
"""The registration row in force on a tax point; None when the business has recorded nothing by then."""
found = [r for r in rows if r.change_state == ChangeState.APPROVED.value and r.effective_from <= on]
return max(found, key=lambda r: r.effective_from) if found else None
def is_registered(rows, on: date) -> Optional[bool]:
"""True/False on a tax point, or None when nothing is recorded (the business must say first)."""
row = status_on(rows, on)
if row is None:
return None
return row.status == RegStatus.REGISTERED.value
def _number(raw: Optional[str]) -> str:
n = clean_vat_number(raw or "")
if not n:
raise unmet("vat_number", "a VAT registration needs its number (GB and 9 or 12 digits)", "Tax settings > Registration")
return n
def _scheme(scheme: Optional[str]) -> str:
s = scheme or Scheme.STANDARD.value
if s not in {x.value for x in Scheme}:
raise RuleBroken(f"scheme is one of {', '.join(x.value for x in Scheme)}, not '{s}'")
return s
def _free(rows, start: date) -> None:
if any(r.effective_from == start for r in live(rows)):
raise RuleBroken(f"there is already a registration row from {start}: a row is never edited — "
"record the correction from a later date")
def _scheme_fields(scheme: str, flat_rate_pct, flat_rate_sector) -> dict:
"""The basis a scheme implies, and the rate a flat rate registration must carry (never a flat rate with no rate)."""
if scheme == Scheme.FLAT_RATE.value and flat_rate_pct is None:
raise unmet("flat_rate_pct", "the flat rate scheme needs its percentage", "give the flat rate percentage for the sector")
flat = scheme == Scheme.FLAT_RATE.value
return {"accounting_basis": "cash" if scheme == Scheme.CASH_ACCOUNTING.value else "invoice",
"flat_rate_sector": (flat_rate_sector or "") if flat else "", "flat_rate_pct": flat_rate_pct if flat else None}
def record_first(rows, status: str, vat_number: Optional[str], effective_from: Optional[date], *, today: date, scheme: Optional[str],
by: str, jurisdiction: str, flat_rate_pct: Optional[Decimal] = None, flat_rate_sector: str = "") -> Registration:
"""record_not_registered or record_registered: the first row. It may carry any date (nothing has used it)."""
if live(rows):
raise RuleBroken("the first registration row is already recorded: change it with register or deregister, "
"from tomorrow at the earliest")
number = _number(vat_number) if status == RegStatus.REGISTERED.value else ""
start = schedule.start_date(effective_from, today=today, has_rows=False, what="the registration", first_default=today)
sch = _scheme(scheme)
extra = _scheme_fields(sch, flat_rate_pct, flat_rate_sector) if status == RegStatus.REGISTERED.value else {}
return Registration(jurisdiction, start, status, number, sch, None, ChangeState.APPROVED.value, by, **extra)
def register(rows, vat_number: Optional[str], effective_from: Optional[date], *, today: date, second_person: bool,
scheme: Optional[str], by: str, jurisdiction: str, flat_rate_pct: Optional[Decimal] = None,
flat_rate_sector: str = "") -> Registration:
rows_ = live(rows)
if not rows_:
raise RuleBroken("record how things stand first (record_not_registered or record_registered)")
current = rows_[-1]
if current.status != RegStatus.NOT_REGISTERED.value:
raise RuleBroken(f"the business is already {current.status.replace('_', ' ')}: only a business that is not registered registers")
number = _number(vat_number)
start = schedule.start_date(effective_from, today=today, has_rows=True, what="a registration",
fix="Tax settings > Registration")
_free(rows, start)
wait = schedule.needs_second_person(control_on=second_person, first_row=False)
sch = _scheme(scheme)
return Registration(jurisdiction, start, RegStatus.REGISTERED.value, number, sch, None,
ChangeState.PENDING.value if wait else ChangeState.APPROVED.value, by,
**_scheme_fields(sch, flat_rate_pct, flat_rate_sector))
def deregister(rows, effective_from: Optional[date], *, today: date, by: str, jurisdiction: str) -> Registration:
rows_ = live(rows)
if not rows_ or rows_[-1].status != RegStatus.REGISTERED.value:
raise RuleBroken("only a registered business deregisters")
start = schedule.start_date(effective_from, today=today, has_rows=True, what="a deregistration")
_free(rows, start)
last = rows_[-1]
return Registration(jurisdiction, start, RegStatus.DEREGISTERED.value, last.vat_number, last.scheme, None,
ChangeState.PENDING.value, by)
def change_scheme(rows, *, scheme: str, effective_from: Optional[date], today: date, second_person: bool, by: str, jurisdiction: str,
expected_turnover: Optional[Decimal], flat_rate_sector: str, flat_rate_pct: Optional[Decimal],
first_year_until: Optional[date], join_limit: Optional[Decimal], ni_goods: Optional[bool],
partial_exemption_year_end: Optional[str], period_starts: set) -> Registration:
"""change_vat_scheme (TAX-36, TAX-37): a new registration row from the FIRST DAY OF A VAT PERIOD after today; the figures the law
sets decide who may join (cash: expected taxable turnover within the joining limit; flat rate: within its limit and a sector)."""
rows_ = live(rows)
if not rows_ or rows_[-1].status != RegStatus.REGISTERED.value:
raise RuleBroken("only a VAT registered business changes its scheme")
scheme = _scheme(scheme)
if effective_from is None or effective_from <= today:
raise unmet("after_today", "a scheme change starts after today (business time)", "pick the first day of a later VAT period")
if effective_from not in period_starts:
raise unmet("vat_period_start", f"{effective_from} is not the first day of a VAT period HMRC shows",
"VAT returns > sync with HMRC, then pick a period's first day")
_free(rows, effective_from)
if scheme != Scheme.STANDARD.value:
if expected_turnover is None:
raise unmet("expected_turnover", "joining a scheme needs the expected taxable turnover", "give the expected turnover")
if join_limit is not None and expected_turnover > join_limit:
raise unmet("scheme_eligible", f"expected taxable turnover {expected_turnover:,.2f} is above the joining limit {join_limit:,.2f}",
"stay on the standard scheme")
if scheme == Scheme.FLAT_RATE.value and (not flat_rate_sector or flat_rate_pct is None):
raise unmet("flat_rate_sector", "the flat rate scheme needs a sector and its rate", "pick a sector")
last = rows_[-1]
wait = schedule.needs_second_person(control_on=second_person, first_row=False)
return Registration(jurisdiction, effective_from, RegStatus.REGISTERED.value, last.vat_number, scheme, None,
ChangeState.PENDING.value if wait else ChangeState.APPROVED.value, by, "", "",
"cash" if scheme == Scheme.CASH_ACCOUNTING.value else "invoice",
flat_rate_sector if scheme == Scheme.FLAT_RATE.value else "",
flat_rate_pct if scheme == Scheme.FLAT_RATE.value else None,
first_year_until if scheme == Scheme.FLAT_RATE.value else None,
last.ni_goods if ni_goods is None else ni_goods,
last.partial_exemption_year_end if partial_exemption_year_end is None else partial_exemption_year_end,
expected_turnover)
def approve(row: Registration, *, today: date, approver: str, approver_ids: tuple = ()) -> Registration:
if row.change_state != ChangeState.PENDING.value:
return row
schedule.approvable(row.effective_from, today=today, author=row.created_by, author_sub=row.created_sub, approver=approver,
approver_ids=approver_ids, what="registration change")
row.change_state, row.approved_by = ChangeState.APPROVED.value, approver
return row
def withdraw(row: Registration, *, today: date) -> Registration:
schedule.withdrawable(row.effective_from, today=today, what="registration")
row.change_state = ChangeState.WITHDRAWN.value
return row