dodil/erp-brick-payroll

DODIL ERP brick: payroll

src/brick_payroll/infrastructure/readers.py

"""The year-to-date reader port (PRL-24, workflow prl_ytd_run): the previous provider's last FPS of the tax year and its year-to-date report per
employee, behind one small interface, so the load never knows how they are reached and tests hand it fake figures. `ports.reader` is what the use
cases use; the default is None, which refuses with a plain message until a composition wires one (a reader of the files the owner attached, or of
Odoo's payroll). Nothing here calls anything: reading is the reader's job, and this brick keeps no credential for it.

What a reader answers for a source reference:
  {"tax_year": "2026-27", "paye_ref": "123/AB45678", "tax_month": 7,
   "fps": {"employees": [{"payroll_id", "taxable", "tax", "ee_ni", "er_ni"}], "totals": {"taxable", "tax", "ee_ni", "er_ni"} (optional)},
   "report": {"employees": [{"payroll_id", "tax_code", "basis", "gross" (optional), "taxable", "tax", "er_ni", "pension",
                             "ni": {"A": {"earnings_at_lel", "lel_to_pt", "pt_to_uel", "above_uel", "employee", "employer"}},
                             "ssp", "smp", "spp", "student_loan" (all optional)}]},
   "source_file_key": "raw/payroll/UK01/migration/..."}
Amounts are strings with two places."""
from typing import Optional, Protocol


class YtdUnavailable(Exception):
    """No reader (409) or the reader failed (502): a reply, never a half-loaded run."""

    def __init__(self, http: int, reason: str):
        super().__init__(reason)
        self.http, self.reason = http, reason


class YtdReader(Protocol):
    def read(self, source_ref: str) -> dict: ...


class Ports:
    def __init__(self):
        self.reader: Optional[YtdReader] = None

    def get(self) -> YtdReader:
        if self.reader is not None:
            return self.reader
        raise YtdUnavailable(409, "no year-to-date reader is configured: the composition wires one that reads the previous provider's last FPS and its "
                                  "year-to-date report")


ports = Ports()