dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/infrastructure/seams.py
"""The neighbours this brick asks, each behind ONE small port, so the domain never knows how it is reached and a test
swaps in an in-memory fake. Declared HERE and nowhere else.
partners business-partner GET /partners/{bp_id} -> {bp_id, legal_name, ... address?}: the customer's name and address
for the invoice-content check (TAX-12). business-partner is a RECOMMENDED dependency: absent, the content check
asks the caller's own copy of the name and address and says which gate is unmet.
ledger gl the sterling rate for a foreign-currency VAT total (TAX-22): GET /fx-rates/{currency}?on=date
-> {rate}. NOT built yet: the route name follows gl's spec (entity fx_rate, table fx_rates); without the ledger
the rate is entered on the document.
clock business time zone: ONE value per bucket (shared.yaml conventions.business_timezone). Until the platform holds
it, it is read from this brick's tax_settings; the read is the only line to change when it moves.
The OTHER direction (sales-order, ar and quote call compute; the ledger maps tax_code to an account) is not here: those
bricks call THIS brick's routes; this brick never reaches into them and never writes the ledger.
"""
from datetime import date
from decimal import Decimal
from typing import Optional, Protocol
from zoneinfo import ZoneInfo
from erp_platform.seam import Seam
from brick_tax.infrastructure.hmrc import FileSecrets, HmrcPort, HttpHmrc, SecretsPort
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=[])
LEDGER = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:fx:read"]) # GET /fx-rates/{currency}?on= (gl spec)
LEDGER_POST = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post"]) # POST /journal-entry/journals (GL-30)
RECEIVABLES = Seam("ar", "AR_BASE_URL", roles=["service"], permissions=["ar:invoice:read"]) # GET /writeoffs (AR-09)
PAYMENTS = Seam("payments", "PAYMENTS_BASE_URL", roles=["service"], permissions=["pay:receipt:read"]) # GET /settlement, /allocations, /cash-basis
PURCHASING = Seam("purchasing", "PURCHASING_BASE_URL", roles=["service"], permissions=["purchasing:payment:read"]) # GET /supplier-payments (PUR-66)
PRODUCT = Seam("product", "PRODUCT_BASE_URL", roles=["service"], permissions=["product:product:read"]) # GET /products/{id} (PRD-28)
class PartnerPort(Protocol):
def customer(self, bp_id: int, caller: dict) -> Optional[dict]:
"""{legal_name, address} as the register has them, or None when it cannot say (absent, refused, down)."""
def resolve_odoo(self, odoo_db: str, odoo_partner_id: int, caller: dict) -> Optional[int]:
"""The bp_id business-partner's Odoo migration gave this Odoo partner (its <prefix>_migration_map), or None."""
class LedgerPort(Protocol):
def fx_rate(self, currency: str, on: date) -> Optional[Decimal]:
"""Sterling per unit of `currency` on a date, or None when the ledger cannot say."""
class LedgerPostPort(Protocol):
def post_journal(self, body: dict) -> Optional[dict]:
"""POST /journal-entry/journals -> {journal_id, ...} (201 or the 200 replay), or None when the ledger cannot say."""
class ReceivablesPort(Protocol):
def writeoffs(self, approved_from: date) -> Optional[list]:
"""ar's approved write-offs from a date: [{writeoff_id, invoice_id, bp_id, amount, currency, approved_at,
invoice_issue_date, tax_point_date, due_date, total}], or None when ar cannot say."""
def invoice(self, invoice_id: str) -> Optional[dict]:
"""ar's GET /invoices/{invoice_id}: {invoice_id, bp_id, ..., lines: [{line_no, product_id, description, quantity, unit_code,
net_amount, tax_code}]}, or None when ar cannot say."""
class PurchasingPort(Protocol):
def supplier_payments(self, frm: date, to: date) -> Optional[list]:
"""purchasing's confirmed supplier payments paid between two dates: [{payment_id, bp_id, amount, paid_on, bills: [{bill_id,
amount}]}], or None when purchasing cannot say."""
class ProductPort(Protocol):
def product(self, product_id: str, on: date) -> Optional[dict]:
"""product's GET /products/{product_id}?on=: commodity_code, country_of_origin, units[].measures.net_weight, or None."""
class PaymentsPort(Protocol):
def cash_basis(self, frm: date, to: date) -> Optional[list]:
"""payments' GET /cash-basis: receipt allocations dated by received_on in the VAT period:
[{payment_id, invoice_id, seq, amount, currency, received_on, allocated_on, source_kind}], or None when payments cannot say."""
def settlement(self, invoice_ids: list, as_at: date) -> Optional[dict]:
"""payments' settlement: {as_at, invoices: [{invoice_id, currency, total, outstanding, state}], unknown: [ids]}."""
def allocations(self, invoice_ids: list, frm: Optional[date]) -> Optional[list]:
"""payments' receipt allocations to invoices from a date: [{source_kind, source_id, invoice_id, seq, amount, currency,
allocated_on, received_on, reverses_seq}], or None when payments cannot say."""
class HttpPartners:
def customer(self, bp_id: int, caller: dict) -> Optional[dict]:
if not PARTNERS.installed:
return None
r = PARTNERS.get(f"/partners/{int(bp_id)}", actor=caller.get("email", ""), caller=caller)
if not r.ok:
return None
return {"legal_name": r.data.get("legal_name") or "", "address": r.data.get("address") or ""}
def resolve_odoo(self, odoo_db: str, odoo_partner_id: int, caller: dict) -> Optional[int]:
"""business-partner's migration map: GET /migration-map/{db}/res.partner?ids= -> [{odoo_model, odoo_id, ours}]
(ours = the bp_id; an unmapped id is left out). Later bricks resolve Odoo ids through the earlier brick's map, never
by matching names (shared.yaml)."""
if not PARTNERS.installed:
return None
r = PARTNERS.get(f"/migration-map/{odoo_db}/res.partner", params={"ids": str(int(odoo_partner_id))},
actor=caller.get("email", ""), caller=caller)
rows = r.data.get("body") if r.ok else None
try:
return next((int(x["ours"]) for x in rows or [] if str(x.get("odoo_id")) == str(int(odoo_partner_id))), None)
except (TypeError, ValueError, KeyError):
return None
class HttpLedger:
def fx_rate(self, currency: str, on: date) -> Optional[Decimal]:
if not LEDGER.installed:
return None
r = LEDGER.get(f"/fx-rates/{currency}", params={"on": on.isoformat()})
if not r.ok or r.data.get("rate") in (None, ""):
return None
try:
return Decimal(str(r.data["rate"]))
except Exception:
return None
class HttpLedgerPost:
def post_journal(self, body: dict) -> Optional[dict]:
if not LEDGER_POST.installed:
return None
r = LEDGER_POST.post("/journal-entry/journals", body, actor="tax")
return r.data if r.ok else None
class HttpReceivables:
def writeoffs(self, approved_from: date) -> Optional[list]:
if not RECEIVABLES.installed:
return None
r = RECEIVABLES.get("/writeoffs", params={"status": "approved", "approved_from": approved_from.isoformat()}, actor="tax")
return (r.data.get("writeoffs") or []) if r.ok else None
def invoice(self, invoice_id: str) -> Optional[dict]:
if not RECEIVABLES.installed:
return None
r = RECEIVABLES.get(f"/invoices/{invoice_id}", actor="tax")
return r.data if r.ok else None
class HttpPurchasing:
def supplier_payments(self, frm: date, to: date) -> Optional[list]:
if not PURCHASING.installed:
return None
r = PURCHASING.get("/supplier-payments", params={"status": "confirmed", "from": frm.isoformat(), "to": to.isoformat()}, actor="tax")
return (r.data.get("payments") or []) if r.ok else None
class HttpProduct:
def product(self, product_id: str, on: date) -> Optional[dict]:
if not PRODUCT.installed:
return None
r = PRODUCT.get(f"/products/{product_id}", params={"on": on.isoformat()}, actor="tax")
return r.data if r.ok else None
class HttpPayments:
def cash_basis(self, frm: date, to: date) -> Optional[list]:
if not PAYMENTS.installed:
return None
r = PAYMENTS.get("/cash-basis", params={"from": frm.isoformat(), "to": to.isoformat()}, actor="tax")
return (r.data.get("allocations") or []) if r.ok else None
def settlement(self, invoice_ids: list, as_at: date) -> Optional[dict]:
if not PAYMENTS.installed:
return None
r = PAYMENTS.get("/settlement", params={"invoice_ids": ",".join(invoice_ids), "as_at": as_at.isoformat()}, actor="tax")
return r.data if r.ok else None
def allocations(self, invoice_ids: list, frm: Optional[date]) -> Optional[list]:
if not PAYMENTS.installed:
return None
params = {"invoice_ids": ",".join(invoice_ids), **({"from": frm.isoformat()} if frm else {})}
r = PAYMENTS.get("/allocations", params=params, actor="tax")
return (r.data.get("allocations") or []) if r.ok else None
# The adapters a use case calls. Tests and a composition replace these attributes.
partners: PartnerPort = HttpPartners()
ledger: LedgerPort = HttpLedger()
ledger_post: LedgerPostPort = HttpLedgerPost()
receivables: ReceivablesPort = HttpReceivables()
payments: PaymentsPort = HttpPayments()
purchasing: PurchasingPort = HttpPurchasing()
products: ProductPort = HttpProduct()
hmrc: HmrcPort = HttpHmrc() # HMRC's VAT API (not a neighbouring brick: no seam contract)
secrets: SecretsPort = FileSecrets() # where the HMRC tokens live
def _zone_row_in_force(settings_rows: list):
"""(zone name, the row that put it in force or None). Which row is in force is decided in BUSINESS time, not UTC: walking
the dated rows in order, a row takes over once the business date in the zone BEFORE it has reached its valid_from. (By
the UTC date a row dated tomorrow could be in force at once, at 18:00 in Pago_Pago.)"""
from erp_platform import work
from brick_tax.domain.values import FOREVER, SETTINGS
rows = sorted((r for r in settings_rows if r.setting_key == "business_timezone"), key=lambda r: r.valid_from)
default = SETTINGS["business_timezone"].default
name, row = next((((r.value or default), r) for r in reversed(rows) if r.valid_from <= FOREVER), (default, None))
for r in rows:
if r.valid_from <= FOREVER:
continue
if r.valid_from <= work.now().astimezone(ZoneInfo(name)).date():
name, row = r.value or default, r
else:
break
return name, row
def business_timezone(settings_rows: list) -> str:
"""The one business time zone. Read from this brick's settings today; the platform's single value tomorrow."""
return _zone_row_in_force(settings_rows)[0]
def business_today(settings_rows: list) -> date:
"""The business date NOW in the zone in force — and never earlier than the date that zone took over, so a switch to a zone
behind the old one cannot turn the clock back (what was refused as 'today' stays refused)."""
from erp_platform import work
name, row = _zone_row_in_force(settings_rows)
now = work.now().astimezone(ZoneInfo(name)).date()
return max(now, row.valid_from) if row is not None else now