dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/infrastructure/tables.py
"""The tables, and one repository per concern. Every table is prefixed `prl_`, has a natural primary key (never an
autoincrement), money is NUMERIC(18,2) and a rate NUMERIC(19,4). schema.sql and migrations/ are written from the same
column model; tests/test_schema.py keeps them equal.
FACTS (a tax code row, a claim, a payslip line, a document, an access log row) are plain INSERTs: a second writer of
the same key is rejected (23505), never overwritten - DataK3's ON CONFLICT DO UPDATE rewrites every column in VALUES.
Only rows that move FORWARD (a run's status, a payslip while its run is calculated, an outbox row) use the keyed upsert.
Nothing is deleted except a cancelled run's period claim (released by cancel_pay_run, the spec's one exception).
Statements per request do not grow with the payslips: a run loads in three queries, payslips and their lines are
written in one statement each, and every list is one query (tests/test_statement_budget.py).
"""
import json
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from zoneinfo import ZoneInfo
from erp_platform import work
from erp_platform.repository import upsert
from sqlalchemy import (
TIMESTAMP,
BigInteger,
Boolean,
Column,
Date,
Integer,
MetaData,
Numeric,
String,
Table,
and_,
delete,
func,
insert,
select,
)
from brick_payroll.domain import pay_run as pr
from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import money
from brick_payroll.domain.values import DEFAULT_ELEMENTS, DEFAULT_RATES, PERIOD_RATES, SEED_FROM, SETTING_DEFS, STATUTORY_ELEMENTS, STATUTORY_RATES
md = MetaData()
MONEY = Numeric(18, 2)
RATE = Numeric(19, 4)
def _stamps():
return (Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
prl_paye_scheme = Table("prl_paye_scheme", md,
Column("company_code", String, primary_key=True),
Column("paye_ref", String, primary_key=True),
Column("accounts_office_ref", String),
Column("set_by", String),
*_stamps())
prl_pay_schedule = Table("prl_pay_schedule", md,
Column("company_code", String, primary_key=True),
Column("schedule_code", String, primary_key=True),
Column("frequency", String),
Column("period_rule", String),
Column("pay_day_rule", String),
Column("paye_ref", String),
Column("set_by", String),
*_stamps())
prl_employee = Table("prl_employee", md,
Column("payroll_id", String, primary_key=True),
Column("person_id", BigInteger),
Column("n", Integer),
Column("company_code", String),
Column("schedule_code", String),
Column("status", String),
Column("start_date", Date),
Column("leaving_date", Date),
Column("starter_declaration", String),
Column("p45_pay_to_date", MONEY),
Column("p45_tax_to_date", MONEY),
Column("details_by", String),
Column("details_sub", String),
*_stamps())
prl_tax_code = Table("prl_tax_code", md,
Column("payroll_id", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("tax_code", String),
Column("basis", String),
Column("source", String),
Column("notice_ref", String),
Column("set_by", String),
Column("set_sub", String),
*_stamps())
prl_ni_category = Table("prl_ni_category", md,
Column("payroll_id", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("category", String),
Column("reason", String),
Column("set_by", String),
*_stamps())
prl_statutory_rate = Table("prl_statutory_rate", md,
Column("rate_key", String, primary_key=True),
Column("region", String, primary_key=True),
Column("frequency", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", RATE),
Column("unit", String),
Column("source", String),
Column("to_confirm", Boolean),
Column("set_by", String),
*_stamps())
prl_element = Table("prl_element", md,
Column("company_code", String, primary_key=True),
Column("element_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("name", String),
Column("kind", String),
Column("taxable", Boolean),
Column("niable", Boolean),
Column("pensionable", Boolean),
Column("gl_purpose", String),
Column("maps_from", String),
*_stamps())
prl_pension_scheme = Table("prl_pension_scheme", md,
Column("company_code", String, primary_key=True),
Column("scheme_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("basis", String),
Column("employee_pct", RATE),
Column("employer_pct", RATE),
Column("tax_relief", String),
Column("provider_capability", String),
Column("set_by", String),
*_stamps())
prl_pension_membership = Table("prl_pension_membership", md,
Column("payroll_id", String, primary_key=True),
Column("scheme_code", String, primary_key=True),
Column("enrolled_on", Date, primary_key=True),
Column("status", String),
Column("assessed_as", String),
Column("set_by", String),
*_stamps())
# The P45 of an employment, as data: `in` is what the starter's previous employer gave (its code, its figures, the leaving date), `out` is
# what this payroll gave the leaver (issued when the final payslip is approved). A rehire is a new employment n+1, so one row per direction.
prl_p45 = Table("prl_p45", md,
Column("payroll_id", String, primary_key=True),
Column("direction", String, primary_key=True),
Column("paye_ref", String),
Column("leaving_date", Date),
Column("tax_code", String),
Column("basis", String),
Column("pay_to_date", MONEY),
Column("tax_to_date", MONEY),
Column("student_loan", Boolean),
Column("ni_category", String),
Column("source_ref", String),
Column("document_key", String),
Column("recorded_by", String),
*_stamps())
prl_pay_run = Table("prl_pay_run", md,
Column("pay_run_id", String, primary_key=True),
Column("company_code", String),
Column("schedule_code", String),
Column("frequency", String),
Column("kind", String),
Column("tax_year", String),
Column("tax_period", Integer),
Column("period_code", String),
Column("period_from", Date),
Column("period_to", Date),
Column("pay_date", Date),
Column("paye_ref", String),
Column("status", String),
Column("calc_no", Integer),
Column("opened_by", String),
Column("opened_sub", String),
Column("prepared_by", String),
Column("prepared_sub", String),
Column("calculated_at", TIMESTAMP),
Column("approved_by", String),
Column("approved_sub", String),
Column("approved_at", TIMESTAMP),
Column("approved_on", Date),
Column("journal_id", BigInteger),
Column("journal_status", String),
Column("journal_reason", String),
Column("hr_batch_id", String),
Column("cancel_reason", String),
*_stamps())
# The row a regular run of a period races for, so a period is paid once (a plain INSERT; released only by cancel_pay_run).
prl_period_claim = Table("prl_period_claim", md,
Column("company_code", String, primary_key=True),
Column("schedule_code", String, primary_key=True),
Column("tax_year", String, primary_key=True),
Column("period_code", String, primary_key=True),
Column("kind", String, primary_key=True),
Column("pay_run_id", String),
Column("claimed_at", TIMESTAMP),
*_stamps())
prl_payslip = Table("prl_payslip", md,
Column("pay_run_id", String, primary_key=True),
Column("payroll_id", String, primary_key=True),
Column("person_id", BigInteger),
Column("calc_no", Integer),
Column("gross_amount", MONEY),
Column("taxable_amount", MONEY),
Column("tax_amount", MONEY),
Column("ee_ni_amount", MONEY),
Column("er_ni_amount", MONEY),
Column("ee_pension_amount", MONEY),
Column("er_pension_amount", MONEY),
Column("student_loan_amount", MONEY),
Column("net_amount", MONEY),
Column("tax_code_used", String),
Column("basis_used", String),
Column("ni_category_used", String),
Column("ytd_json", String),
Column("cost_centre", String),
Column("bp_id", BigInteger),
Column("inputs_sha", String),
Column("stale", Boolean),
Column("stale_reason", String),
Column("approved_by", String),
Column("approved_sub", String),
Column("payout_ref", String),
Column("payout_status", String),
Column("payout_reason", String),
Column("paid_on", Date),
Column("payment_id", String),
Column("document_key", String),
Column("published_on", Date),
Column("leaving_date", Date),
*_stamps())
# A payslip's lines of ONE calculation (calc_no): recalculating writes the next calc_no's lines, so nothing is deleted and
# the lines a run was approved with are the lines of its last calc_no.
prl_payslip_line = Table("prl_payslip_line", md,
Column("pay_run_id", String, primary_key=True),
Column("payroll_id", String, primary_key=True),
Column("calc_no", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("element_code", String),
Column("kind", String),
Column("quantity", RATE),
Column("rate_amount", RATE),
Column("amount", MONEY),
Column("source_kind", String),
Column("source_ref", String),
Column("rate_rows_used", String),
*_stamps())
# A payslip (later a P45, P60) kept for the person and for HMRC's record-keeping period: the bytes in the bucket under
# the spec's object key raw/payroll/<company>/<tax year>/<kind>/<person_id>/<ref>.pdf.
prl_document = Table("prl_document", md,
Column("object_key", String, primary_key=True),
Column("kind", String),
Column("payroll_id", String),
Column("person_id", BigInteger),
Column("ref", String),
Column("pay_run_id", String),
Column("content_type", String),
Column("sha256", String),
Column("size_bytes", Integer),
Column("content_b64", String),
Column("published_on", Date),
Column("retain_until", Date),
*_stamps())
# Every read of another person's payslip by a payroll user (row_visibility: "every read of another person's payslip is logged").
prl_access_log = Table("prl_access_log", md,
Column("object_key", String, primary_key=True),
Column("reader", String, primary_key=True),
Column("read_at", TIMESTAMP, primary_key=True),
Column("reader_sub", String),
Column("person_id", BigInteger),
Column("what", String),
*_stamps())
# One guard row per bounded operation on money or on a person (scopes: approve, payslip_approval, paid, returned, cancel).
prl_claim = Table("prl_claim", md,
Column("claim_scope", String, primary_key=True),
Column("claim_key", String, primary_key=True),
Column("n", Integer, primary_key=True),
Column("ref", String),
Column("claimed_by", String),
Column("claimed_at", TIMESTAMP),
*_stamps())
# A side effect waiting for a neighbour, written in the SAME transaction as the state change that needs it and sent only
# after that commit. target: hr (extract sent) | gl (the journal) | payments (the run's payouts).
prl_outbox = Table("prl_outbox", md,
Column("target", String, primary_key=True),
Column("ref", String, primary_key=True),
Column("pay_run_id", String),
Column("payload_json", String),
Column("status", String),
Column("attempts", Integer),
Column("last_error", String),
Column("sent_at", TIMESTAMP),
*_stamps())
# One FPS, EPS or pension contribution file sent (or to be sent) through a capability: submission_id fps:<run>:<n> | eps:<paye_ref>:<tax year>:M<nn>:<n> |
# pension:<scheme>:<run>. The file itself is a prl_document under object_key; the claim is a plain INSERT, the status moves forward (an upsert).
prl_rti_submission = Table("prl_rti_submission", md,
Column("submission_id", String, primary_key=True),
Column("kind", String),
Column("pay_run_id", String),
Column("company_code", String),
Column("paye_ref", String),
Column("tax_year", String),
Column("tax_period", Integer),
Column("scheme_code", String),
Column("n", Integer),
Column("object_key", String),
Column("sha256", String),
Column("status", String),
Column("correlation_id", String),
Column("message", String),
Column("errors_json", String),
Column("totals_json", String),
Column("attempts", Integer),
Column("due_on", Date),
Column("sent_at", TIMESTAMP),
*_stamps())
# The year-to-date load (PRL-24): a run per load of the previous provider's figures (loaded -> reconciled -> signed | discarded) and one opening row per employment
# and tax year, keyed "ytd:<payroll_id>:<tax year>". An opening counts in the cumulative calculation only once its run is SIGNED.
prl_ytd_run = Table("prl_ytd_run", md,
Column("run_id", String, primary_key=True),
Column("tax_year", String),
Column("paye_ref", String),
Column("company_code", String),
Column("schedule_code", String),
Column("cutover_date", Date),
Column("source_ref", String),
Column("status", String),
Column("fps_json", String),
Column("checks_json", String),
Column("loaded_by", String),
Column("signed_by", String),
Column("signed_sub", String),
Column("signed_on", Date),
Column("discarded_by", String),
Column("discard_reason", String),
*_stamps())
prl_ytd_opening = Table("prl_ytd_opening", md,
Column("ytd_ref", String, primary_key=True),
Column("payroll_id", String),
Column("tax_year", String),
Column("paye_ref", String),
Column("run_id", String),
Column("tax_code", String),
Column("basis", String),
Column("gross_amount", MONEY),
Column("taxable_pay_amount", MONEY),
Column("tax_amount", MONEY),
Column("ee_ni_amount", MONEY),
Column("er_ni_amount", MONEY),
Column("ni_by_letter_json", String),
Column("statutory_json", String),
Column("pension_amount", MONEY),
Column("source_file_key", String),
*_stamps())
prl_settings = Table("prl_settings", md,
Column("setting_key", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", String),
Column("set_by", String),
*_stamps())
class AlreadyDone(RuleBroken):
"""The one-shot claim was already taken: a 409 with these words."""
def _utc(dt: Optional[datetime]) -> Optional[datetime]:
if dt is None:
return None
return dt.astimezone(timezone.utc).replace(tzinfo=None) if dt.tzinfo else dt
def _plain(conn, table: Table, rows, at: datetime) -> None:
"""FACT rows: plain INSERTs in ONE statement - a second writer of the same key is rejected (23505), never overwritten."""
rows = rows if isinstance(rows, list) else [rows]
if rows:
conn.execute(insert(table), [{**r, "created_at": _utc(at), "updated_at": _utc(at)} for r in rows])
def _all(conn, table: Table, cond=None, order=()) -> list:
q = select(table)
if cond is not None:
q = q.where(cond)
return [dict(m) for m in conn.execute(q.order_by(*order)).mappings()]
def _m(v) -> Decimal:
return money(v if v is not None else 0)
def _in_force(rows: list, keyf, on: date) -> dict:
"""key -> the row with the latest valid_from on or before `on`."""
out: dict = {}
for r in sorted(rows, key=lambda r: r["valid_from"]):
if r["valid_from"] <= on:
out[keyf(r)] = r
return out
# ── the pay run ───────────────────────────────────────────────────────────────────────────────
RUN_COLS = tuple(c.name for c in prl_pay_run.columns if c.name not in ("created_at", "updated_at"))
SLIP_MONEY = {"gross": "gross_amount", "taxable": "taxable_amount", "tax": "tax_amount", "ee_ni": "ee_ni_amount", "er_ni": "er_ni_amount",
"ee_pension": "ee_pension_amount", "er_pension": "er_pension_amount", "student_loan": "student_loan_amount", "net": "net_amount"}
class PayRunRepository:
"""A run whole: the root, its payslips of the current calculation, and their lines - three queries however many payslips."""
def __init__(self, conn):
self.conn = conn
def root(self, pay_run_id: str) -> Optional[pr.PayRun]:
row = self.conn.execute(select(prl_pay_run).where(prl_pay_run.c.pay_run_id == pay_run_id)).mappings().first()
return self._run(dict(row)) if row else None
def get(self, pay_run_id: str, lines: bool = True) -> Optional[pr.PayRun]:
run = self.root(pay_run_id)
if run is None:
return None
slips = _all(self.conn, prl_payslip, (prl_payslip.c.pay_run_id == pay_run_id) & (prl_payslip.c.calc_no == run.calc_no),
(prl_payslip.c.payroll_id,))
by_id = {}
for s in slips:
p = self._slip(s)
by_id[p.payroll_id] = p
run.payslips.append(p)
if lines and slips:
for ln in _all(self.conn, prl_payslip_line, (prl_payslip_line.c.pay_run_id == pay_run_id) & (prl_payslip_line.c.calc_no == run.calc_no),
(prl_payslip_line.c.payroll_id, prl_payslip_line.c.line_no)):
if ln["payroll_id"] in by_id:
by_id[ln["payroll_id"]].lines.append(pr.Line(ln["line_no"], ln["element_code"], ln["kind"], ln["quantity"], ln["rate_amount"],
_m(ln["amount"]), ln["source_kind"], ln["source_ref"] or "", ln["rate_rows_used"] or ""))
return run
@staticmethod
def _run(r: dict) -> pr.PayRun:
fields = pr.PayRun.__dataclass_fields__
return pr.PayRun(**{k: (r[k] if r[k] is not None else ("" if fields[k].type in (str, "str") else None))
for k in RUN_COLS if k in fields})
@staticmethod
def _slip(s: dict) -> pr.Payslip:
return pr.Payslip(
pay_run_id=s["pay_run_id"], payroll_id=s["payroll_id"], person_id=s["person_id"], calc_no=s["calc_no"],
**{k: _m(s[c]) for k, c in SLIP_MONEY.items()}, tax_code_used=s["tax_code_used"] or "", basis_used=s["basis_used"] or "",
ni_category_used=s["ni_category_used"] or "", ytd=json.loads(s["ytd_json"] or "{}"), cost_centre=s["cost_centre"] or "",
bp_id=s["bp_id"], inputs_sha=s["inputs_sha"] or "", stale=bool(s["stale"]), stale_reason=s["stale_reason"] or "",
approved_by=s["approved_by"] or "", approved_sub=s["approved_sub"] or "", payout_ref=s["payout_ref"] or "",
payout_status=s["payout_status"] or "", payout_reason=s["payout_reason"] or "", paid_on=s["paid_on"], payment_id=s["payment_id"] or "",
document_key=s["document_key"] or "", published_on=s["published_on"], leaving_date=s["leaving_date"])
def _root_row(self, run: pr.PayRun) -> dict:
row = {k: getattr(run, k) for k in RUN_COLS}
for k in ("calculated_at", "approved_at"):
row[k] = _utc(row[k])
return {k: (None if v == "" else v) for k, v in row.items()}
def create(self, run: pr.PayRun, at: datetime) -> None:
_plain(self.conn, prl_pay_run, self._root_row(run), at)
def save_root(self, run: pr.PayRun, at: datetime) -> None:
upsert(self.conn, prl_pay_run, [{**self._root_row(run), "updated_at": _utc(at)}], ("pay_run_id",))
@staticmethod
def slip_row(p: pr.Payslip, at: datetime) -> dict:
return {"pay_run_id": p.pay_run_id, "payroll_id": p.payroll_id, "person_id": p.person_id, "calc_no": p.calc_no,
**{c: getattr(p, k) for k, c in SLIP_MONEY.items()}, "tax_code_used": p.tax_code_used, "basis_used": p.basis_used,
"ni_category_used": p.ni_category_used, "ytd_json": json.dumps(p.ytd, sort_keys=True, default=str), "cost_centre": p.cost_centre or None,
"bp_id": p.bp_id, "inputs_sha": p.inputs_sha, "stale": p.stale, "stale_reason": p.stale_reason or None,
"approved_by": p.approved_by or None, "approved_sub": p.approved_sub or None, "payout_ref": p.payout_ref or None,
"payout_status": p.payout_status or None, "payout_reason": p.payout_reason or None, "paid_on": p.paid_on,
"payment_id": p.payment_id or None, "document_key": p.document_key or None, "published_on": p.published_on,
"leaving_date": p.leaving_date, "updated_at": _utc(at)}
def save_payslips(self, slips: list, at: datetime) -> None:
"""Payslips move forward while their run is calculated (a recalculation, a peer's approval, a payout's answer):
ONE keyed upsert for all of them."""
if slips:
upsert(self.conn, prl_payslip, [self.slip_row(p, at) for p in slips], ("pay_run_id", "payroll_id"))
def add_lines(self, slips: list, at: datetime) -> None:
"""The lines of this calculation, every payslip's, in ONE statement."""
rows = [{"pay_run_id": p.pay_run_id, "payroll_id": p.payroll_id, "calc_no": p.calc_no, "line_no": ln.line_no,
"element_code": ln.element_code, "kind": ln.kind, "quantity": ln.quantity, "rate_amount": ln.rate_amount, "amount": ln.amount,
"source_kind": ln.source_kind, "source_ref": ln.source_ref or None, "rate_rows_used": ln.rate_rows_used or None}
for p in slips for ln in p.lines]
_plain(self.conn, prl_payslip_line, rows, at)
def runs(self, company_code: Optional[str] = None, limit: int = 200) -> list:
q = select(prl_pay_run)
if company_code:
q = q.where(prl_pay_run.c.company_code == company_code)
return [dict(r) for r in self.conn.execute(q.order_by(prl_pay_run.c.pay_date.desc(), prl_pay_run.c.pay_run_id).limit(limit)).mappings()]
def of_schedule(self, company_code: str, schedule_code: str) -> list:
return _all(self.conn, prl_pay_run, (prl_pay_run.c.company_code == company_code) & (prl_pay_run.c.schedule_code == schedule_code),
(prl_pay_run.c.pay_date,))
def ytd_before(self, run: pr.PayRun, payroll_ids: list) -> dict:
"""This employer's figures to date per payroll_id: the payslips of every APPROVED run (and beyond) of the same PAYE
scheme and tax year paid before this run - ONE grouped query. Order-independent: cumulative tax from these sums."""
if not payroll_ids:
return {}
s, r = prl_payslip, prl_pay_run
q = (select(s.c.payroll_id, func.sum(s.c.gross_amount), func.sum(s.c.taxable_amount), func.sum(s.c.tax_amount),
func.sum(s.c.ee_ni_amount), func.sum(s.c.er_ni_amount), func.sum(s.c.ee_pension_amount), func.sum(s.c.er_pension_amount))
.select_from(s.join(r, and_(s.c.pay_run_id == r.c.pay_run_id, s.c.calc_no == r.c.calc_no)))
.where(r.c.paye_ref == run.paye_ref, r.c.company_code == run.company_code, r.c.tax_year == run.tax_year,
r.c.status.in_(pr.FROZEN), r.c.pay_run_id != run.pay_run_id, r.c.pay_date <= run.pay_date, s.c.payroll_id.in_(payroll_ids))
.group_by(s.c.payroll_id))
out = {}
for row in self.conn.execute(q):
out[row[0]] = dict(zip(pr.YTD_KEYS, (_m(v) for v in row[1:])))
o, y = prl_ytd_opening, prl_ytd_run # the previous provider's figures: counted once their run is signed (migration, PRL-24)
oq = (select(o.c.payroll_id, o.c.gross_amount, o.c.taxable_pay_amount, o.c.tax_amount, o.c.ee_ni_amount, o.c.er_ni_amount, o.c.pension_amount)
.select_from(o.join(y, o.c.run_id == y.c.run_id))
.where(y.c.status == "signed", o.c.paye_ref == run.paye_ref, o.c.tax_year == run.tax_year, o.c.payroll_id.in_(payroll_ids)))
for pid, gross, taxable, tax, ee_ni, er_ni, pension in self.conn.execute(oq):
have = out.get(pid) or {k: _m(0) for k in pr.YTD_KEYS}
add = {"gross": gross if gross is not None else taxable, "taxable": taxable, "tax": tax, "ee_ni": ee_ni, "er_ni": er_ni, "ee_pension": pension}
out[pid] = {k: _m(have[k]) + _m(add.get(k)) for k in pr.YTD_KEYS}
return out
def history(self, run: pr.PayRun, payroll_ids: list, lo, hi) -> dict:
"""What statutory pay averages and limits read, for the people who have a statutory absence: their gross pay by pay date in
[lo, hi] and the SSP days already paid per linked group - from APPROVED runs of this scheme and tax year (the current run
excluded), two queries however many people."""
out = {"gross": {}, "ssp_days": {}}
if not payroll_ids or lo is None:
return out
s, r, ln = prl_payslip, prl_pay_run, prl_payslip_line
join = s.join(r, and_(s.c.pay_run_id == r.c.pay_run_id, s.c.calc_no == r.c.calc_no))
base = [r.c.paye_ref == run.paye_ref, r.c.company_code == run.company_code, r.c.status.in_(pr.FROZEN), r.c.pay_run_id != run.pay_run_id,
s.c.payroll_id.in_(payroll_ids)]
for row in self.conn.execute(select(s.c.payroll_id, r.c.pay_date, s.c.gross_amount).select_from(join)
.where(*base, r.c.pay_date >= lo, r.c.pay_date <= hi).order_by(r.c.pay_date)):
out["gross"].setdefault(row[0], []).append((row[1], _m(row[2])))
lines = (select(s.c.payroll_id, ln.c.source_ref, ln.c.quantity).select_from(join.join(ln, and_(ln.c.pay_run_id == s.c.pay_run_id,
ln.c.payroll_id == s.c.payroll_id, ln.c.calc_no == s.c.calc_no)))
.where(*base, r.c.pay_date < run.pay_date, ln.c.element_code == "SSP"))
for pid, ref, qty in self.conn.execute(lines):
group = (ref or "").rsplit(" group ", 1)[-1]
out["ssp_days"].setdefault(pid, {})[group] = out["ssp_days"].get(pid, {}).get(group, 0) + int(qty or 0)
return out
def frozen_of_scheme(self, paye_ref: str, tax_year: str) -> list:
"""The approved (and later) runs reported under a PAYE reference in a tax year: what an EPS and the HMRC liability add up."""
r = prl_pay_run
return _all(self.conn, r, (r.c.paye_ref == paye_ref) & (r.c.tax_year == tax_year) & r.c.status.in_(pr.FROZEN), (r.c.pay_date, r.c.pay_run_id))
def element_by_run(self, pay_run_ids: list, element_code: str) -> dict:
"""pay_run_id -> the sum of one element over the run's current payslips (SMP paid), one query."""
if not pay_run_ids:
return {}
r, ln = prl_pay_run, prl_payslip_line
q = (select(ln.c.pay_run_id, func.sum(ln.c.amount)).select_from(ln.join(r, and_(ln.c.pay_run_id == r.c.pay_run_id, ln.c.calc_no == r.c.calc_no)))
.where(ln.c.pay_run_id.in_(pay_run_ids), ln.c.element_code == element_code).group_by(ln.c.pay_run_id))
return {row[0]: _m(row[1]) for row in self.conn.execute(q)}
def payslips_of_person(self, person_id: int) -> list:
"""A person's payslips in runs that are approved or beyond (what My payslips can show): ONE query."""
s, r = prl_payslip, prl_pay_run
q = (select(s, r.c.pay_date, r.c.status.label("run_status"), r.c.tax_year, r.c.tax_period, r.c.company_code)
.select_from(s.join(r, and_(s.c.pay_run_id == r.c.pay_run_id, s.c.calc_no == r.c.calc_no)))
.where(s.c.person_id == person_id, r.c.status.in_(pr.FROZEN)).order_by(r.c.pay_date.desc()))
return [dict(m) for m in self.conn.execute(q).mappings()]
class ClaimRepository:
def __init__(self, conn):
self.conn = conn
def count(self, scope: str, key: str) -> int:
return self.conn.execute(select(func.count()).select_from(prl_claim).where(
(prl_claim.c.claim_scope == scope) & (prl_claim.c.claim_key == key))).scalar() or 0
def once(self, scope: str, key: str, at: datetime, by: str = "", ref: str = "") -> None:
"""A key claimed exactly ONCE: two racers both see none, both insert n=1, the second is rejected at the INSERT (23505)."""
if self.count(scope, key):
raise AlreadyDone(f"Someone else did the same a moment ago ({scope} {key}) - load the run again")
_plain(self.conn, prl_claim, {"claim_scope": scope, "claim_key": key, "n": 1, "ref": ref, "claimed_by": by, "claimed_at": _utc(at)}, at)
class PeriodClaimRepository:
def __init__(self, conn):
self.conn = conn
def get(self, key: tuple) -> Optional[dict]:
c = prl_period_claim.c
row = self.conn.execute(select(prl_period_claim).where((c.company_code == key[0]) & (c.schedule_code == key[1]) & (c.tax_year == key[2])
& (c.period_code == key[3]) & (c.kind == key[4]))).mappings().first()
return dict(row) if row else None
def claim(self, key: tuple, pay_run_id: str, at: datetime) -> None:
if self.get(key) is not None:
raise RuleBroken(f"the period {key[3]} of {key[2]} is already being paid by another regular run", gate="period_free")
_plain(self.conn, prl_period_claim, {"company_code": key[0], "schedule_code": key[1], "tax_year": key[2], "period_code": key[3],
"kind": key[4], "pay_run_id": pay_run_id, "claimed_at": _utc(at)}, at)
def release(self, key: tuple) -> None:
c = prl_period_claim.c
self.conn.execute(delete(prl_period_claim).where((c.company_code == key[0]) & (c.schedule_code == key[1]) & (c.tax_year == key[2])
& (c.period_code == key[3]) & (c.kind == key[4])))
class OutboxRepository:
def __init__(self, conn):
self.conn = conn
def add(self, target: str, ref: str, pay_run_id: str, payload: dict, at: datetime) -> None:
_plain(self.conn, prl_outbox, {"target": target, "ref": ref, "pay_run_id": pay_run_id, "payload_json": json.dumps(payload, sort_keys=True),
"status": "pending", "attempts": 0}, at)
def add_many(self, rows: list, pay_run_id: str, at: datetime) -> None:
"""Several pending rows in ONE statement: [(target, ref, payload)]."""
_plain(self.conn, prl_outbox, [{"target": t, "ref": ref, "pay_run_id": pay_run_id, "payload_json": json.dumps(pl, sort_keys=True), "status": "pending",
"attempts": 0} for t, ref, pl in rows], at)
def get(self, target: str, ref: str) -> Optional[dict]:
row = self.conn.execute(select(prl_outbox).where((prl_outbox.c.target == target) & (prl_outbox.c.ref == ref))).mappings().first()
return dict(row) if row else None
def pending(self, pay_run_id: Optional[str] = None) -> list:
q = select(prl_outbox).where(prl_outbox.c.status == "pending")
if pay_run_id:
q = q.where(prl_outbox.c.pay_run_id == pay_run_id)
return [dict(r) for r in self.conn.execute(q.order_by(prl_outbox.c.created_at, prl_outbox.c.target, prl_outbox.c.ref)).mappings()]
def of_run(self, pay_run_id: str) -> list:
return _all(self.conn, prl_outbox, prl_outbox.c.pay_run_id == pay_run_id, (prl_outbox.c.target, prl_outbox.c.ref))
def mark(self, row: dict, status: str, error: str, at: datetime) -> None:
keep = {k: row[k] for k in ("target", "ref", "pay_run_id", "payload_json")}
upsert(self.conn, prl_outbox, [{**keep, "status": status, "attempts": (row["attempts"] or 0) + 1, "last_error": error or None,
"sent_at": _utc(at) if status == "sent" else None, "updated_at": _utc(at)}], ("target", "ref"))
class EmployeeRepository:
"""The payroll record of each employment, its dated tax code and NI rows and its pension memberships."""
def __init__(self, conn):
self.conn = conn
def get(self, payroll_id: str) -> Optional[dict]:
row = self.conn.execute(select(prl_employee).where(prl_employee.c.payroll_id == payroll_id)).mappings().first()
return dict(row) if row else None
def many(self, payroll_ids: list) -> dict:
if not payroll_ids:
return {}
return {r["payroll_id"]: r for r in _all(self.conn, prl_employee, prl_employee.c.payroll_id.in_(payroll_ids))}
def all(self, company_code: Optional[str] = None, limit: int = 500) -> list:
q = select(prl_employee)
if company_code:
q = q.where(prl_employee.c.company_code == company_code)
return [dict(r) for r in self.conn.execute(q.order_by(prl_employee.c.payroll_id).limit(limit)).mappings()]
def add(self, rows: list, at: datetime) -> None:
_plain(self.conn, prl_employee, rows, at)
def save(self, row: dict, at: datetime) -> None:
upsert(self.conn, prl_employee, [{**{k: v for k, v in row.items() if k != "created_at"}, "updated_at": _utc(at)}], ("payroll_id",))
def add_tax_code(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_tax_code, row, at)
def add_ni_category(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_ni_category, row, at)
def add_membership(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_pension_membership, row, at)
def facts(self, payroll_ids: list, on: date) -> dict:
"""payroll_id -> {tax_code, ni, membership}: the rows in force on `on` - three queries however many people."""
if not payroll_ids:
return {}
tc = _in_force(_all(self.conn, prl_tax_code, prl_tax_code.c.payroll_id.in_(payroll_ids)), lambda r: r["payroll_id"], on)
ni = _in_force(_all(self.conn, prl_ni_category, prl_ni_category.c.payroll_id.in_(payroll_ids)), lambda r: r["payroll_id"], on)
ms = [r for r in _all(self.conn, prl_pension_membership, prl_pension_membership.c.payroll_id.in_(payroll_ids))
if r["enrolled_on"] <= on and r["status"] == "enrolled"]
mem = {}
for r in sorted(ms, key=lambda r: r["enrolled_on"]):
mem[r["payroll_id"]] = r
return {p: {"tax_code": tc.get(p), "ni": ni.get(p), "membership": mem.get(p)} for p in payroll_ids}
def add_p45(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_p45, row, at)
def p45s(self, payroll_ids: list) -> dict:
"""{(payroll_id, direction): row}: the P45s in and out of these employments, one query."""
if not payroll_ids:
return {}
return {(r["payroll_id"], r["direction"]): r for r in _all(self.conn, prl_p45, prl_p45.c.payroll_id.in_(payroll_ids))}
def history(self, payroll_id: str) -> dict:
return {"p45": _all(self.conn, prl_p45, prl_p45.c.payroll_id == payroll_id, (prl_p45.c.direction,)),
"tax_codes": _all(self.conn, prl_tax_code, prl_tax_code.c.payroll_id == payroll_id, (prl_tax_code.c.valid_from,)),
"ni_categories": _all(self.conn, prl_ni_category, prl_ni_category.c.payroll_id == payroll_id, (prl_ni_category.c.valid_from,)),
"pension": _all(self.conn, prl_pension_membership, prl_pension_membership.c.payroll_id == payroll_id,
(prl_pension_membership.c.enrolled_on,))}
class ConfigRepository:
"""The customer's rows: PAYE schemes, pay schedules, pension schemes, elements, statutory rates (all dated where they change)."""
def __init__(self, conn):
self.conn = conn
def schedule(self, company_code: str, schedule_code: str) -> Optional[dict]:
c = prl_pay_schedule.c
row = self.conn.execute(select(prl_pay_schedule).where((c.company_code == company_code) & (c.schedule_code == schedule_code))).mappings().first()
return dict(row) if row else None
def put_schedule(self, row: dict, at: datetime) -> None:
upsert(self.conn, prl_pay_schedule, [{**row, "updated_at": _utc(at)}], ("company_code", "schedule_code"))
def paye_scheme(self, company_code: str, paye_ref: str) -> Optional[dict]:
c = prl_paye_scheme.c
row = self.conn.execute(select(prl_paye_scheme).where((c.company_code == company_code) & (c.paye_ref == paye_ref))).mappings().first()
return dict(row) if row else None
def paye_scheme_of(self, paye_ref: str) -> Optional[dict]:
rows = _all(self.conn, prl_paye_scheme, prl_paye_scheme.c.paye_ref == paye_ref, (prl_paye_scheme.c.company_code,))
return rows[0] if rows else None
def put_paye_scheme(self, row: dict, at: datetime) -> None:
upsert(self.conn, prl_paye_scheme, [{**row, "updated_at": _utc(at)}], ("company_code", "paye_ref"))
def put_pension_scheme(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_pension_scheme, row, at)
def pension_schemes(self, company_code: str, on: date) -> dict:
rows = _all(self.conn, prl_pension_scheme, prl_pension_scheme.c.company_code == company_code)
return _in_force(rows, lambda r: r["scheme_code"], on)
def pension_scheme_rows(self, company_code: str) -> list:
return _all(self.conn, prl_pension_scheme, prl_pension_scheme.c.company_code == company_code,
(prl_pension_scheme.c.scheme_code, prl_pension_scheme.c.valid_from))
def elements(self, company_code: str, on: date) -> dict:
"""element_code -> the row in force: the company's own over the '*' default."""
rows = _all(self.conn, prl_element, prl_element.c.company_code.in_((company_code, "*")))
base = _in_force([r for r in rows if r["company_code"] == "*"], lambda r: r["element_code"], on)
own = _in_force([r for r in rows if r["company_code"] == company_code], lambda r: r["element_code"], on)
return {**base, **own}
def rates(self, on: date, regions: tuple = ("uk", "ruk")) -> list:
"""The statutory rows in force on `on` (one query): per (rate_key, region, frequency), the latest valid_from <= on."""
rows = _all(self.conn, prl_statutory_rate, prl_statutory_rate.c.region.in_(regions))
return list(_in_force(rows, lambda r: (r["rate_key"], r["region"], r["frequency"]), on).values())
def all_rates(self) -> list:
return _all(self.conn, prl_statutory_rate, None, (prl_statutory_rate.c.rate_key, prl_statutory_rate.c.frequency, prl_statutory_rate.c.valid_from))
class DocumentRepository:
def __init__(self, conn):
self.conn = conn
def add(self, rows: list, at: datetime) -> None:
_plain(self.conn, prl_document, rows, at)
def get(self, object_key: str) -> Optional[dict]:
row = self.conn.execute(select(prl_document).where(prl_document.c.object_key == object_key)).mappings().first()
return dict(row) if row else None
def log_read(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_access_log, row, at)
def reads(self, object_key: str) -> list:
return _all(self.conn, prl_access_log, prl_access_log.c.object_key == object_key, (prl_access_log.c.read_at,))
class RtiRepository:
"""The submissions to HMRC and to the pension scheme (rti_submission): claimed once by a plain INSERT, then moved forward."""
def __init__(self, conn):
self.conn = conn
def get(self, submission_id: str) -> Optional[dict]:
row = self.conn.execute(select(prl_rti_submission).where(prl_rti_submission.c.submission_id == submission_id)).mappings().first()
return dict(row) if row else None
def of_run(self, pay_run_id: str) -> list:
return _all(self.conn, prl_rti_submission, prl_rti_submission.c.pay_run_id == pay_run_id, (prl_rti_submission.c.submission_id,))
def waiting(self) -> list:
"""What still needs a person or a retry: every submission that is not accepted, oldest deadline first."""
c = prl_rti_submission.c
return _all(self.conn, prl_rti_submission, c.status != "accepted", (c.due_on, c.submission_id))
def eps_of(self, paye_ref: str, tax_year: str) -> list:
c = prl_rti_submission.c
return _all(self.conn, prl_rti_submission, (c.kind == "eps") & (c.paye_ref == paye_ref) & (c.tax_year == tax_year), (c.tax_period, c.n))
def claim(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_rti_submission, {"attempts": 0, **row}, at)
def save(self, row: dict, at: datetime) -> None:
keep = {k: v for k, v in row.items() if k != "created_at"}
upsert(self.conn, prl_rti_submission, [{**keep, "updated_at": _utc(at)}], ("submission_id",))
class YtdRepository:
"""The previous provider's year to date: runs and opening rows (migration, PRL-24)."""
def __init__(self, conn):
self.conn = conn
def run(self, run_id: str) -> Optional[dict]:
row = self.conn.execute(select(prl_ytd_run).where(prl_ytd_run.c.run_id == run_id)).mappings().first()
return dict(row) if row else None
def runs(self) -> list:
return _all(self.conn, prl_ytd_run, None, (prl_ytd_run.c.created_at, prl_ytd_run.c.run_id))
def add_run(self, row: dict, at: datetime) -> None:
_plain(self.conn, prl_ytd_run, row, at)
def save_run(self, row: dict, at: datetime) -> None:
upsert(self.conn, prl_ytd_run, [{**{k: v for k, v in row.items() if k != "created_at"}, "updated_at": _utc(at)}], ("run_id",))
def add_openings(self, rows: list, at: datetime) -> None:
_plain(self.conn, prl_ytd_opening, rows, at)
def of_run(self, run_id: str) -> list:
return _all(self.conn, prl_ytd_opening, prl_ytd_opening.c.run_id == run_id, (prl_ytd_opening.c.payroll_id,))
def held(self, refs: list) -> dict:
"""ytd_ref -> the opening row already held (by any run), one query."""
if not refs:
return {}
return {r["ytd_ref"]: r for r in _all(self.conn, prl_ytd_opening, prl_ytd_opening.c.ytd_ref.in_(refs))}
def delete_of_run(self, run_id: str) -> None:
self.conn.execute(delete(prl_ytd_opening).where(prl_ytd_opening.c.run_id == run_id))
class SettingsRepository:
"""Dated settings. The business date is counted in `business_timezone` - read HERE and nowhere else."""
def __init__(self, conn):
self.conn = conn
self._rows: Optional[list] = None
def rows(self) -> list:
if self._rows is None:
self._rows = _all(self.conn, prl_settings, None, (prl_settings.c.setting_key, prl_settings.c.valid_from))
return self._rows
def on(self, key: str, day: date) -> str:
value = SETTING_DEFS[key][0]
for r in self.rows():
if r["setting_key"] == key and r["valid_from"] <= day:
value = r["value"]
return value
def timezone(self) -> str:
name = SETTING_DEFS["business_timezone"][0]
now = work.now()
for r in self.rows():
if r["setting_key"] == "business_timezone" and r["valid_from"] <= now.astimezone(ZoneInfo(name)).date():
name = r["value"] or name
return name
def put(self, key: str, value: str, valid_from: date, by: str, at: datetime) -> None:
_plain(self.conn, prl_settings, {"setting_key": key, "valid_from": valid_from, "value": str(value), "set_by": by}, at)
self._rows = None
class Repository:
"""Every repository on one connection (one unit of work)."""
def __init__(self, conn):
self.conn = conn
self.runs = PayRunRepository(conn)
self.claims = ClaimRepository(conn)
self.periods = PeriodClaimRepository(conn)
self.outbox = OutboxRepository(conn)
self.employees = EmployeeRepository(conn)
self.config = ConfigRepository(conn)
self.documents = DocumentRepository(conn)
self.settings = SettingsRepository(conn)
self.rti = RtiRepository(conn)
self.ytd = YtdRepository(conn)
def seed(self, at: Optional[datetime] = None) -> None:
"""The elements and the 2026/27 statutory rows every bucket starts with, once - the same rows migration 00002 inserts."""
at = at or work.now()
have = {r["element_code"] for r in _all(self.conn, prl_element, prl_element.c.company_code == "*")}
_plain(self.conn, prl_element, [{"company_code": "*", "element_code": c, "valid_from": date(2000, 1, 1), "name": n, "kind": k, "taxable": t,
"niable": ni, "pensionable": p, "gl_purpose": g, "maps_from": mf or None}
for c, n, k, t, ni, p, g, mf in (*DEFAULT_ELEMENTS, *STATUTORY_ELEMENTS) if c not in have], at)
held = {(r["rate_key"], r["region"], r["frequency"], r["valid_from"]) for r in _all(self.conn, prl_statutory_rate)}
every = [(k, rg, f, SEED_FROM, v, u, s) for k, rg, f, v, u, s in DEFAULT_RATES] + [(k, rg, f, vf, v, u, s) for k, rg, f, vf, v, u, s in (*STATUTORY_RATES, *PERIOD_RATES)]
_plain(self.conn, prl_statutory_rate, [{"rate_key": k, "region": rg, "frequency": f, "valid_from": vf, "value": Decimal(v), "unit": u,
"source": s, "to_confirm": True, "set_by": "seed"}
for k, rg, f, vf, v, u, s in every if (k, rg, f, vf) not in held], at)