dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/domain/tax_code.py
"""The TaxCode aggregate: a jurisdiction's code for one tax treatment (S standard, Z zero, RC reverse charge), with its
dated rates. This module is the sole writer (shared.yaml keys.tax_code); the ledger only maps a code to an account.
Lifecycle draft -> active -> retired (terminal). A code is never deleted. Its treatment and reporting_box are fixed
once a document has used it (a different meaning is a new code); its description and wording may change, because each
issued document keeps the wording it froze. A rate row in effect is never edited; a later one is scheduled; a row not yet
in effect may be withdrawn.
"""
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from brick_tax.domain import schedule
from brick_tax.domain.errors import Gate, GateUnmet, RuleBroken, unmet
from brick_tax.domain.events import (
CodeActivated,
CodeAdded,
CodeRetired,
PackLoaded,
RateApproved,
RateScheduled,
RateWithdrawn,
)
from brick_tax.domain.values import CODE, ChangeState, CodeState, Treatment
ZERO_TREATMENTS = (Treatment.ZERO.value, Treatment.EXEMPT.value, Treatment.OUTSIDE_SCOPE.value, Treatment.NOT_REGISTERED.value)
HUNDRED = Decimal(100)
@dataclass
class Rate:
"""tax_rate: a code's percentage from a date. rate_pct is what is due, charged_pct what the supplier charges."""
tax_code: str
valid_from: date
rate_pct: Decimal
charged_pct: Decimal
change_state: str = ChangeState.APPROVED.value
created_by: str = ""
approved_by: str = ""
withdrawn_by: str = ""
withdrawn_at: Optional[datetime] = None
created_sub: str = ""
@property
def in_force(self) -> bool:
return self.change_state == ChangeState.APPROVED.value
def check_rates(treatment: str, rate_pct: Decimal, charged_pct: Decimal) -> None:
if not (Decimal(0) <= rate_pct <= HUNDRED and Decimal(0) <= charged_pct <= HUNDRED):
raise RuleBroken("a rate is between 0 and 100 percent")
if treatment in ZERO_TREATMENTS and (rate_pct != 0 or charged_pct != 0):
raise RuleBroken(f"a {treatment.replace('_', ' ')} code carries 0%: use a standard or reduced code for a taxed supply")
if treatment == Treatment.REVERSE_CHARGE.value:
if charged_pct != 0:
raise RuleBroken("a reverse-charge code charges 0%: the customer accounts for the VAT (rate_pct is what is due)")
if rate_pct <= 0:
raise RuleBroken("a reverse-charge code records the VAT due: give the rate that is due (20, say)")
if treatment in (Treatment.STANDARD.value, Treatment.REDUCED.value):
if rate_pct != charged_pct:
raise RuleBroken("a standard or reduced code charges the rate that is due")
if rate_pct <= 0:
raise RuleBroken("a standard or reduced code carries a rate above 0: use a zero-rated code for 0%")
@dataclass
class TaxCode:
tax_code: str
jurisdiction: str = "GB"
treatment: str = Treatment.STANDARD.value
description: str = ""
document_wording: str = ""
reporting_box: str = ""
lifecycle: str = CodeState.DRAFT.value
used: bool = False # a document has used it: its meaning is fixed
direction: str = "sale" # sale | purchase (v3)
input_reporting_box: str = "" # a purchase reverse charge: the box of the input side
value_box: str = "" # 6 | 7 | 8 | 9: where the net value is reported (v4)
reclaimable_pct: Optional[Decimal] = None # purchase codes only: 100 reclaimable, 0 non-reclaimable (business entertainment)
attribution: str = "" # purchase codes: taxable | exempt | residual (partial exemption, v4)
eu_goods: bool = False # NI goods to or from the EU: feeds boxes 2, 8, 9 and the trade declarations (v4)
rates: list = field(default_factory=list)
events: list = field(default_factory=list, repr=False, compare=False)
# ── adding ────────────────────────────────────────────────────────────────────────────────────────────────
@classmethod
def add(cls, tax_code: str, treatment: str, *, description: str = "", document_wording: str = "",
reporting_box: str = "", jurisdiction: str = "GB", by: str = "", direction: str = "sale",
input_reporting_box: str = "", value_box: str = "", reclaimable_pct: Optional[Decimal] = None,
attribution: str = "", eu_goods: bool = False) -> "TaxCode":
code = (tax_code or "").strip()
if not CODE.match(code):
raise RuleBroken("a tax code is up to 16 letters, digits and . _ - characters, starting with a letter or digit (S, RC, Z)")
if treatment not in {t.value for t in Treatment}:
raise RuleBroken(f"treatment is one of {', '.join(t.value for t in Treatment)}, not '{treatment}'")
if len(jurisdiction) != 2 or not jurisdiction.isalpha():
raise RuleBroken("a jurisdiction is an ISO 3166-1 alpha-2 country code (GB)")
if direction not in ("sale", "purchase"):
raise RuleBroken("a code's direction is sale or purchase")
if value_box and value_box not in ("6", "7", "8", "9"):
raise RuleBroken("the value box is 6, 7, 8 or 9")
if direction == "sale" and (reclaimable_pct is not None or input_reporting_box):
raise RuleBroken("reclaimable_pct and the input box belong to purchase codes")
if direction == "purchase":
reclaimable_pct = Decimal(100) if reclaimable_pct is None else reclaimable_pct
if not (Decimal(0) <= reclaimable_pct <= HUNDRED):
raise RuleBroken("reclaimable_pct is between 0 and 100")
if attribution and (direction != "purchase" or attribution not in ("taxable", "exempt", "residual")):
raise RuleBroken("attribution (taxable, exempt or residual) belongs to purchase codes")
c = cls(code, jurisdiction.upper(), treatment, description.strip(), document_wording.strip(), reporting_box.strip())
c.attribution, c.eu_goods = attribution, bool(eu_goods)
c.direction, c.input_reporting_box, c.value_box, c.reclaimable_pct = direction, input_reporting_box.strip(), value_box, reclaimable_pct
c.events.append(CodeAdded(code, treatment, by))
return c
def describe(self, *, description=None, document_wording=None, treatment=None, reporting_box=None, by: str = "") -> None:
"""Rename and reword freely; the treatment and reporting_box are fixed once a document used the code."""
if self.lifecycle == CodeState.RETIRED.value:
raise RuleBroken(f"{self.tax_code} is retired")
if self.used and ((treatment is not None and treatment != self.treatment)
or (reporting_box is not None and reporting_box != self.reporting_box)):
raise RuleBroken(f"{self.tax_code} has been used by a document: its treatment and reporting box are fixed — "
"a different meaning is a new code")
if treatment is not None and treatment != self.treatment:
if treatment not in {t.value for t in Treatment}:
raise RuleBroken(f"treatment is one of {', '.join(t.value for t in Treatment)}, not '{treatment}'")
self.treatment = treatment
if description is not None:
self.description = description.strip()
if document_wording is not None:
self.document_wording = document_wording.strip()
if reporting_box is not None:
self.reporting_box = reporting_box.strip()
# ── rates ─────────────────────────────────────────────────────────────────────────────────────────────────
def live_rates(self) -> list:
return sorted((r for r in self.rates if r.change_state != ChangeState.WITHDRAWN.value), key=lambda r: r.valid_from)
def rate_on(self, on: date) -> Optional[Rate]:
"""The rate charged on a tax point: the latest approved row that started on or before it."""
found = [r for r in self.rates if r.in_force and r.valid_from <= on]
return max(found, key=lambda r: r.valid_from) if found else None
def schedule_rate(self, rate_pct: Decimal, charged_pct: Optional[Decimal], valid_from: Optional[date], *, today: date,
second_person: bool, by: str) -> Rate:
"""The first rate of a code carries any date (no document has used it); every later one starts after today, and
waits for a second holder of tax:code:approve where the owner's control asks."""
if self.lifecycle == CodeState.RETIRED.value:
raise RuleBroken(f"{self.tax_code} is retired: it takes no new rate")
charged = rate_pct if charged_pct is None and self.treatment != Treatment.REVERSE_CHARGE.value else (charged_pct or Decimal(0))
check_rates(self.treatment, rate_pct, charged)
first = not self.live_rates()
start = schedule.start_date(valid_from, today=today, has_rows=not first, what=f"a new rate for {self.tax_code}", first_default=today)
if any(r.valid_from == start for r in self.live_rates()):
raise RuleBroken(f"{self.tax_code} already has a rate from {start}: a rate row is never edited — "
"schedule the correction from a later date")
wait = schedule.needs_second_person(control_on=second_person, first_row=first)
rate = Rate(self.tax_code, start, rate_pct, charged, ChangeState.PENDING.value if wait else ChangeState.APPROVED.value, by)
self.rates.append(rate)
self.events.append(RateScheduled(self.tax_code, start, str(rate_pct), by))
return rate
def _row(self, valid_from: date) -> Rate:
found = next((r for r in self.rates if r.valid_from == valid_from and r.change_state != ChangeState.WITHDRAWN.value), None)
if found is None:
raise RuleBroken(f"{self.tax_code} has no rate from {valid_from}")
return found
def approve_rate(self, valid_from: date, *, today: date, approver: str, approver_ids: tuple = ()) -> Rate:
row = self._row(valid_from)
if row.change_state != ChangeState.PENDING.value:
return row
schedule.approvable(row.valid_from, today=today, author=row.created_by, author_sub=row.created_sub, approver=approver,
approver_ids=approver_ids, what="rate change")
row.change_state, row.approved_by = ChangeState.APPROVED.value, approver
self.events.append(RateApproved(self.tax_code, valid_from, approver))
return row
def withdraw_rate(self, valid_from: date, *, today: date, by: str, now: datetime) -> Rate:
"""A rate not yet in effect never takes effect. A draft code's rate may also be withdrawn: nothing used it."""
row = self._row(valid_from)
if not (self.lifecycle == CodeState.DRAFT.value and not self.used):
schedule.withdrawable(row.valid_from, today=today, what="rate")
row.change_state, row.withdrawn_by, row.withdrawn_at = ChangeState.WITHDRAWN.value, by, now
self.events.append(RateWithdrawn(self.tax_code, valid_from, by))
return row
# ── lifecycle ─────────────────────────────────────────────────────────────────────────────────────────────
def activation_gates(self) -> list:
has_rate = any(r.in_force for r in self.rates)
needs_wording = self.treatment == Treatment.REVERSE_CHARGE.value
return [Gate("first_rate", has_rate, "a code needs a rate before it can be used" if not has_rate else "the code has its rate",
"Tax codes > the code > Rates"),
Gate("wording", bool(self.document_wording) or not needs_wording,
"a reverse-charge code prints its legend on the invoice" if not (self.document_wording or not needs_wording)
else "the wording is set", "Tax codes > the code > Wording")]
def activate(self, *, by: str = "") -> None:
if self.lifecycle != CodeState.DRAFT.value:
raise RuleBroken(f"{self.tax_code} is {self.lifecycle}: only a draft code is activated")
bad = [g for g in self.activation_gates() if not g.met]
if bad:
raise GateUnmet(bad)
self.lifecycle = CodeState.ACTIVE.value
self.events.append(CodeActivated(self.tax_code, by))
def retire(self, *, rules_pointing_here: int, by: str = "") -> None:
if self.lifecycle != CodeState.ACTIVE.value:
raise RuleBroken(f"{self.tax_code} is {self.lifecycle}: only an active code is retired")
if rules_pointing_here:
raise unmet("no_rule_points_here", f"{rules_pointing_here} tax rule(s) in effect today or later still choose "
f"{self.tax_code}", "Tax rules: end that rule first")
self.lifecycle = CodeState.RETIRED.value
self.events.append(CodeRetired(self.tax_code, by))
def mark_used(self) -> None:
self.used = True
@classmethod
def from_history(cls, tax_code: str, treatment: str, *, description: str, document_wording: str, reporting_box: str,
rate_pct: Decimal, charged_pct: Decimal, valid_from: date, lifecycle: str, by: str) -> "TaxCode":
"""A code the Odoo migration creates (tax:history:import): with its one rate, straight to ACTIVE, or — an archived Odoo tax
a posted invoice used — straight to RETIRED, so migrated history names a real code. First rows only."""
c = cls.add(tax_code, treatment, description=description, document_wording=document_wording,
reporting_box=reporting_box, by=by)
c.rates.append(Rate(tax_code, valid_from, rate_pct, charged_pct, ChangeState.APPROVED.value, by))
c.lifecycle = lifecycle
c.used = lifecycle == CodeState.RETIRED.value
c.events.append(PackLoaded(tax_code, by))
return c