dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/application/use_cases.py
"""The use cases - what a payroll administrator, a payroll manager or an employee (or their agent) can do with pay.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere else: WHO MAY
(a PERMISSION, never a role's name) and WHICH ROWS (row_visibility: payroll data only to payroll roles; an employee sees only
their own published payslips; a row the caller may not see is a 404, never a 403).
The rules that shape every use case (module.yaml in_code_not_rows; tests/test_invariants.py and test_stories.py):
- nobody calculates and approves the same run, approves their own payslip, or changes their own payroll record, under
either login (email AND sub are both checked); the approver's own payslip is approved by a peer first
- one claim key per pay run or person operation (prl_claim, prl_period_claim): a racer's loss is a 409, never a second effect
- side effects (hr's extract marked sent, the journal, the payouts) only AFTER the local commit, from pending prl_outbox rows
written in the approval's transaction; a rolled-back approval sends nothing; a crash is sent by the next drain
- states only move forward (domain/pay_run.py MOVES); an approved run is never edited - a correction is a later run
- 'today' is the date in business_timezone; bounds are 422 {gate: bounds}, never a 500
- no read-your-writes inside an open transaction (DataK3): each use case reads committed rows, writes, and answers from what it holds
"""
import json
from contextlib import contextmanager
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Literal, Optional
from zoneinfo import ZoneInfo
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
from sqlalchemy.exc import IntegrityError
from brick_payroll.application import gather as G
from brick_payroll.application import statutory_pay as SP
from brick_payroll.application.inputs import AmountText, Code, Company, Day, Note, PercentText, Period, Reason, Ref, day, parse_money, parse_percent
from brick_payroll.domain import pay_run as pr
from brick_payroll.domain import paye
from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import money
from brick_payroll.infrastructure import documents, seams
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
EMP_READ, EMP_WRITE, TAX_WRITE, PENSION_WRITE = "prl:employee:read", "prl:employee:write", "prl:tax_code:write", "prl:pension:write"
RUN_READ, RUN_OPEN, RUN_CALC, RUN_APPROVE, RUN_CANCEL = ("prl:pay_run:read", "prl:pay_run:open", "prl:pay_run:calculate", "prl:pay_run:approve",
"prl:pay_run:cancel")
SCHEME_WRITE, READ_OWN, DOC_READ = "prl:scheme:write", "prl:payslip:read_own", "prl:document:read"
# machine: held only by a service client, by an EXACT grant - no wildcard reaches them (module.yaml permissions.machine)
MARK_PAID, POST_AUTO = "prl:payout:mark_paid", "prl:journal:post_auto"
MACHINE = ("prl:pay_run:open_auto", "prl:payslip:derive", POST_AUTO, "prl:payout:request", MARK_PAID, "prl:rti:submit", "prl:pension:submit",
"prl:document:publish", "prl:year_end:run", "prl:migration:load", "prl:migration:reconcile")
PRL = Brick("payroll", namespace="prl", version="0.2.1", ids_as_strings=True, machine_permissions=MACHINE, instructions=(
"Payroll (UK): pay salaried employees monthly or weekly from hr's payroll extract, with SSP and SMP from time-off's absences. Set up once: the PAYE scheme (put_paye_scheme), "
"the monthly schedule (put_pay_schedule) and the pension scheme (put_pension_scheme). Each employment gets its tax code "
"and NI category (complete_payroll_details) and pension membership (enrol_pension). Each month: open the period "
"(open_pay_run), calculate it (calculate_pay_run: gross from hr, PAYE cumulative, NI category A, pension on qualifying "
"earnings, every figure on a line with its source), then ANOTHER payroll manager approves it (approve_pay_run) - never "
"the person who calculated it, and a manager's own payslip is approved by a peer first (approve_approver_payslip). The "
"approval posts one journal to the ledger and asks payments to pay each net; payslips are published to employees on "
"the pay date (list_my_payslips). A starter gets a tax code from a P45 (record_p45_in) or a starter statement (record_starter_declaration); a leaver's P45 "
"comes with their final payslip. After approval the FPS and the pension file are built and handed to a provider (none: list_rti_submissions shows the manual "
"tasks, complete_manual_submission closes one). Amounts are exact strings in GBP."))
PRL.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text
@contextmanager
def _on_engine(engine):
"""One transaction. A racer's duplicate key can surface at COMMIT on DataK3, after the use case returned - so it is
caught around the commit and answered as the retry-safe 409, never a 500."""
from brick_payroll.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else did the same a moment ago - load the pay run again and retry") from None
repository = _repository_default # tests and a composition swap in their own
@contextmanager
def _decide():
"""A broken rule is a 409 with its own words and gate (a bound is a 422), through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(422 if broken.gate == "bounds" else 409, str(broken), gate=broken.gate, fix=broken.fix) from None
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
except IntegrityError:
raise Refused(409, "Someone else did the same a moment ago - load the pay run again and retry") from None
def _who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _me(caller: dict) -> pr.Identity:
return pr.Identity((caller.get("email") or "").strip().lower(), (caller.get("sub") or "").strip())
def _now() -> datetime:
return work.now()
def _today(repo) -> date:
"""Business today: work.now() in business_timezone (read from ONE place), never the UTC date."""
return _now().astimezone(ZoneInfo(repo.settings.timezone())).date()
def _has(caller: dict, perm: str) -> bool:
return auth.has_permission(caller, perm)
def _need(caller: dict, *perms: str) -> None:
for perm in perms:
if not _has(caller, perm):
raise Refused(403, f"missing permission: {perm}")
def _bounds(reason: str, fix: str = "check the value and send it again") -> Refused:
return Refused(422, reason, gate="bounds", fix=fix)
def _date(v: Optional[str], what: str) -> date:
try:
d = day(v)
except ValueError:
raise _bounds(f"{what} is a date like 2026-11-30") from None
if not date(2000, 1, 1) <= d <= date(2100, 1, 1):
raise _bounds(f"{what} {d} is outside 2000-01-01 .. 2100-01-01")
return d
def _money(v, what: str) -> Decimal:
try:
return parse_money(v, what)
except ValueError as e:
raise _bounds(str(e)) from None
def _pct(v, what: str) -> Decimal:
try:
return parse_percent(v, what)
except ValueError as e:
raise _bounds(str(e)) from None
def _company(v: str) -> str:
v = (v or "").strip().upper()
if not (1 <= len(v) <= 4 and v.isalnum() and v.isascii()):
raise _bounds(f"{v!r} is a company code of 1 to 4 letters or digits (UK01)")
return v
def _my_person(caller: dict) -> Optional[int]:
"""The caller's person in hr, by their sign-in (HR-58), or None. Used for 'your own' - never for permission."""
me = _me(caller)
got = seams.person_by_sub(me.sub, _who(caller)) if me.sub else None
return int(got["person_id"]) if got and got.get("person_id") is not None else None
def _s(v) -> Optional[str]:
return None if v is None else f"{Decimal(v):.2f}"
# ── views ─────────────────────────────────────────────────────────────────────────────────────
def _line_view(ln: pr.Line) -> dict:
return {"line_no": ln.line_no, "element_code": ln.element_code, "kind": ln.kind,
"quantity": None if ln.quantity is None else f"{Decimal(ln.quantity):.4f}",
"rate_amount": None if ln.rate_amount is None else f"{Decimal(ln.rate_amount):.4f}", "amount": _s(ln.amount),
"source_kind": ln.source_kind, "source_ref": ln.source_ref or None, "rate_rows_used": ln.rate_rows_used or None}
def _slip_view(p: pr.Payslip, lines: bool = True) -> dict:
out = {"pay_run_id": p.pay_run_id, "payroll_id": p.payroll_id, "person_id": str(p.person_id), "gross": _s(p.gross), "taxable": _s(p.taxable),
"tax": _s(p.tax), "employee_ni": _s(p.ee_ni), "employer_ni": _s(p.er_ni), "employee_pension": _s(p.ee_pension),
"employer_pension": _s(p.er_pension), "student_loan": _s(p.student_loan), "net": _s(p.net),
"cost_to_employer": _s(p.gross + p.er_ni + p.er_pension), "tax_code": p.tax_code_used, "basis": p.basis_used,
"ni_category": p.ni_category_used, "cost_centre": p.cost_centre or None, "stale": p.stale, "stale_reason": p.stale_reason or None,
"approved_by_peer": p.approved_by 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.isoformat() if p.paid_on else None,
"leaving_date": p.leaving_date.isoformat() if p.leaving_date else None, "document_key": p.document_key or None,
"published_on": p.published_on.isoformat() if p.published_on else None,
"to_date": {k: (p.ytd or {}).get(k) for k in pr.YTD_KEYS}}
if lines:
out["lines"] = [_line_view(ln) for ln in p.lines]
out["tax_detail"] = (p.ytd or {}).get("tax_detail")
out["statutory"] = (p.ytd or {}).get("statutory")
out["warnings"] = (p.ytd or {}).get("warnings") or []
return out
def _run_view(repo, run: pr.PayRun, lines: bool = False, gates: Optional[list] = None) -> dict:
slips = run.payslips
out = {"pay_run_id": run.pay_run_id, "company_code": run.company_code, "schedule_code": run.schedule_code, "kind": run.kind,
"tax_year": run.tax_year, "tax_period": run.tax_period, "period": run.period_code, "period_from": run.period_from.isoformat(),
"period_to": run.period_to.isoformat(), "pay_date": run.pay_date.isoformat(), "paye_ref": run.paye_ref, "status": run.status,
"calculation": run.calc_no, "prepared_by": run.prepared_by or None, "approved_by": run.approved_by or None,
"approved_on": run.approved_on.isoformat() if run.approved_on else None,
"journal_id": None if run.journal_id is None else str(run.journal_id), "journal_status": run.journal_status or None,
"journal_reason": run.journal_reason or None, "cancel_reason": run.cancel_reason or None, "payslip_count": len(slips),
"totals": {k: _s(sum((getattr(p, a) for p in slips), Decimal(0))) for k, a in
(("gross", "gross"), ("tax", "tax"), ("employee_ni", "ee_ni"), ("employer_ni", "er_ni"), ("employee_pension", "ee_pension"),
("employer_pension", "er_pension"), ("net", "net"))},
"payslips": [_slip_view(p, lines) for p in slips]}
if gates is not None:
out["readiness"] = gates
box = repo.outbox.of_run(run.pay_run_id) if run.status in pr.FROZEN else []
if run.status in pr.FROZEN:
out["filings"] = [{"submission_id": f["submission_id"], "kind": f["kind"], "status": f["status"], "due_on": f["due_on"].isoformat() if f["due_on"] else None,
"correlation_id": f["correlation_id"] or None} for f in repo.rti.of_run(run.pay_run_id)]
if box:
out["outbox"] = [{"target": b["target"], "ref": b["ref"], "status": b["status"], "attempts": b["attempts"], "last_error": b["last_error"]}
for b in box]
return out
def _run(repo, pay_run_id: str, lines: bool = True) -> pr.PayRun:
run = repo.runs.get(pay_run_id, lines=lines)
if run is None:
raise Refused(404, f"no pay run {pay_run_id}")
return run
# ── setting up: the PAYE scheme, the schedule, the pension scheme ───────────────────────────────
@PRL.use_case(kind="command", route="POST /paye-schemes", permission=SCHEME_WRITE, idempotent=True)
def put_paye_scheme(caller: dict, company_code: Company, paye_ref: Code, accounts_office_ref: Code) -> dict:
"""Record (or correct) the employer PAYE scheme HMRC registered for a company: the PAYE reference (123/AB45678) and the
accounts office reference (123PA00045678). A pay schedule names the scheme it reports under."""
cc, ref, aor = _company(company_code), paye_ref.strip().upper(), accounts_office_ref.strip().upper()
if not (len(ref) >= 5 and ref[:3].isdigit() and ref[3] == "/" and ref[4:].isalnum()):
raise _bounds(f"{paye_ref!r} is a PAYE reference like 123/AB45678")
if not (len(aor) == 13 and aor[:3].isdigit() and aor[3] == "P" and aor[4].isalpha() and aor[5:].isalnum()):
raise _bounds(f"{accounts_office_ref!r} is an accounts office reference like 123PA00045678 (13 characters)")
with repository() as repo:
row = {"company_code": cc, "paye_ref": ref, "accounts_office_ref": aor, "set_by": _who(caller)}
repo.config.put_paye_scheme(row, _now())
return row
@PRL.use_case(kind="command", route="POST /pay-schedules", permission=SCHEME_WRITE, idempotent=True)
def put_pay_schedule(caller: dict, company_code: Company, schedule_code: Code, paye_ref: Code, frequency: Literal["monthly", "weekly"] = "monthly",
pay_day_rule: Optional[Literal["last_working_day", "friday_after"]] = None) -> dict:
"""Record how a group is paid: the schedule code that goes into each run's id (M, W), monthly by calendar month paid on the
last working day, or weekly by ISO week (Monday to Sunday, worked in arrears) paid the Friday after, reported under the PAYE
scheme named. Fortnightly and four-weekly tax and NI tables exist, but hr's extract has no multi-week period to pay them from."""
cc, code = _company(company_code), schedule_code.strip().upper()
if not (code.isalnum() and len(code) <= 4):
raise _bounds(f"{schedule_code!r} is a schedule code of 1 to 4 letters or digits (M, W)")
with repository() as repo:
if repo.config.paye_scheme(cc, paye_ref.strip().upper()) is None:
raise Refused(409, f"no PAYE scheme {paye_ref} for {cc}", gate="paye_scheme", fix="record it first (put_paye_scheme)")
rule = {"monthly": "last_working_day", "weekly": "friday_after"}[frequency]
if pay_day_rule not in (None, rule):
raise _bounds(f"a {frequency} schedule is paid {rule}, not {pay_day_rule}")
row = {"company_code": cc, "schedule_code": code, "frequency": frequency,
"period_rule": {"monthly": "calendar_month", "weekly": "iso_week_in_arrears"}[frequency], "pay_day_rule": rule,
"paye_ref": paye_ref.strip().upper(), "set_by": _who(caller)}
repo.config.put_schedule(row, _now())
return row
@PRL.use_case(kind="command", route="POST /pension-schemes", permission=SCHEME_WRITE, status=201)
def put_pension_scheme(caller: dict, company_code: Company, scheme_code: Code, employee_pct: PercentText, employer_pct: PercentText,
valid_from: Day, basis: Literal["qualifying_earnings"] = "qualifying_earnings",
tax_relief: Literal["net_pay"] = "net_pay") -> dict:
"""Add a dated row of a workplace pension scheme: contributions on qualifying earnings, the employee's and the employer's
percentage, tax relief by the net pay arrangement. A qualifying scheme takes at least 8% in all with at least 3% from the
employer (Pensions Act 2008 s. 20). A change is a new row from a later date; a row already in force is never rewritten."""
cc, code = _company(company_code), scheme_code.strip().upper()
ee, er, frm = _pct(employee_pct, "employee_pct"), _pct(employer_pct, "employer_pct"), _date(valid_from, "valid_from")
if er < 3 or ee + er < 8:
raise Refused(422, f"{code}: {ee}% + {er}% is below the legal minimum on qualifying earnings (8% in all, 3% from the employer)",
gate="minimum_contribution", fix="raise the employer's or the employee's percentage")
with repository() as repo, _decide():
today = _today(repo)
have = [r for r in repo.config.pension_scheme_rows(cc) if r["scheme_code"] == code]
if have and frm <= today:
raise RuleBroken(f"{code} already has rows: a change starts after today ({today})", gate="dated_after_today")
row = {"company_code": cc, "scheme_code": code, "valid_from": frm, "basis": basis, "employee_pct": ee, "employer_pct": er,
"tax_relief": tax_relief, "provider_capability": "pension.submit_contributions", "set_by": _who(caller)}
repo.config.put_pension_scheme(row, _now())
return {**row, "valid_from": frm.isoformat(), "employee_pct": f"{ee:.4f}", "employer_pct": f"{er:.4f}"}
@PRL.use_case(kind="query", route="GET /statutory-rates", permission=RUN_READ)
def list_statutory_rates(caller: dict) -> dict:
"""The statutory rates, bands and thresholds as dated rows (tax bands, NI thresholds per frequency, auto-enrolment, the
minimum wage), each with its source and whether a payroll manager still has to confirm it against HMRC's published figure."""
with repository() as repo:
return {"rates": [{"rate_key": r["rate_key"], "region": r["region"], "frequency": r["frequency"], "valid_from": r["valid_from"].isoformat(),
"value": f"{Decimal(r['value']):.4f}", "unit": r["unit"], "source": r["source"], "to_confirm": bool(r["to_confirm"])}
for r in repo.config.all_rates()]}
# ── employees on the payroll ──────────────────────────────────────────────────────────────────
def _emp_view(row: dict, history: Optional[dict] = None) -> dict:
out = {"payroll_id": row["payroll_id"], "person_id": str(row["person_id"]), "company_code": row["company_code"],
"schedule_code": row["schedule_code"], "status": row["status"], "start_date": row["start_date"].isoformat() if row["start_date"] else None,
"leaving_date": row["leaving_date"].isoformat() if row["leaving_date"] else None, "starter_declaration": row["starter_declaration"],
"p45_pay_to_date": _s(row["p45_pay_to_date"]), "p45_tax_to_date": _s(row["p45_tax_to_date"])}
if history is not None:
out["tax_codes"] = [{"valid_from": r["valid_from"].isoformat(), "tax_code": r["tax_code"], "basis": r["basis"], "source": r["source"],
"notice_ref": r["notice_ref"]} for r in history["tax_codes"]]
out["ni_categories"] = [{"valid_from": r["valid_from"].isoformat(), "category": r["category"]} for r in history["ni_categories"]]
out["p45"] = [{"direction": r["direction"], "paye_ref": r["paye_ref"], "leaving_date": r["leaving_date"].isoformat() if r["leaving_date"] else None,
"tax_code": r["tax_code"], "basis": r["basis"], "pay_to_date": _s(r["pay_to_date"]), "tax_to_date": _s(r["tax_to_date"]),
"student_loan": bool(r["student_loan"]), "document_key": r["document_key"]} for r in history.get("p45", [])]
out["pension"] = [{"scheme_code": r["scheme_code"], "enrolled_on": r["enrolled_on"].isoformat(), "status": r["status"]} for r in history["pension"]]
return out
@PRL.use_case(kind="query", route="GET /payroll-employees", permission=EMP_READ)
def list_payroll_employees(caller: dict, company_code: Optional[Company] = None,
status: Optional[Literal["pending_details", "active", "left"]] = None) -> dict:
"""The employments on the payroll (one record per employment as HMRC knows it, payroll_id <person_id>-<n>), with the
starters still waiting for their tax details. Salaries are never here: they stay in hr."""
with repository() as repo:
rows = repo.employees.all(_company(company_code) if company_code else None)
return {"employees": [_emp_view(r) for r in rows if status is None or r["status"] == status]}
@PRL.use_case(kind="query", route="GET /payroll-employees/{payroll_id}", permission=EMP_READ)
def get_payroll_employee(caller: dict, payroll_id: Ref) -> dict:
"""One employment's payroll record: its status, its dated tax codes (with their source) and NI categories, its P45 figures
and its pension membership."""
with repository() as repo:
row = repo.employees.get(payroll_id)
if row is None:
raise Refused(404, f"no payroll employee {payroll_id}")
return _emp_view(row, repo.employees.history(payroll_id))
NI_LETTERS = set("ABCHJMZFILSVNEDK")
def _complete_details(caller: dict, payroll_id: str, tax_code: str, ni_category: str, source: str, basis: str, valid_from: Optional[str],
starter_declaration: Optional[str], p45_pay_to_date: Optional[str], p45_tax_to_date: Optional[str], notice_ref: Optional[str],
p45: Optional[dict] = None) -> dict:
"""pending_details -> active: the one place a payroll record is completed, whichever door (a code from a notice or a manual entry, a
starter's statement, a P45, the emergency code). Needs prl:tax_code:write too; nobody completes their own record, under either login."""
_need(caller, TAX_WRITE)
try:
code = paye.parse_tax_code(tax_code).code
except paye.UnsupportedHere:
code = tax_code.strip().upper() # kept as HMRC issued it; the calculation names what it cannot do yet
except RuleBroken as bad:
raise _bounds(str(bad)) from None
letter = ni_category.strip().upper()
if letter not in NI_LETTERS or len(letter) != 1:
raise _bounds(f"{ni_category!r} is an NI category letter (A, B, C, H, J, M, Z, ...)")
pay45 = _money(p45_pay_to_date, "p45_pay_to_date") if p45_pay_to_date is not None else None
try:
tax45 = parse_money(p45_tax_to_date, "p45_tax_to_date", negative_ok=True) if p45_tax_to_date is not None else None
except ValueError as e:
raise _bounds(str(e)) from None
if source == "p45" and (pay45 is None or tax45 is None):
raise _bounds("a P45 gives pay to date and tax to date: send p45_pay_to_date and p45_tax_to_date")
if source == "starter_checklist" and starter_declaration is None:
raise _bounds("a starter checklist gives a statement: send starter_declaration A, B or C")
frm = _date(valid_from, "valid_from") if valid_from else None
with repository() as repo, _decide():
row = repo.employees.get(payroll_id)
if row is None:
raise Refused(404, f"no payroll employee {payroll_id}")
if _my_person(caller) == row["person_id"]:
raise RuleBroken("You cannot change your own payroll record: another payroll administrator does it", gate="not_own")
row["status"] = pr.move(row["status"], "complete_payroll_details", pr.EMPLOYEE_MOVES)
frm = frm or row["start_date"] or _today(repo)
me = _me(caller)
row.update({"starter_declaration": starter_declaration, "p45_pay_to_date": pay45, "p45_tax_to_date": tax45, "details_by": me.email,
"details_sub": me.sub})
repo.employees.add_tax_code({"payroll_id": payroll_id, "valid_from": frm, "tax_code": code, "basis": basis, "source": source,
"notice_ref": notice_ref, "set_by": me.email, "set_sub": me.sub}, _now())
repo.employees.add_ni_category({"payroll_id": payroll_id, "valid_from": frm, "category": letter, "reason": source, "set_by": me.email}, _now())
if p45:
repo.employees.add_p45({"payroll_id": payroll_id, "direction": "in", "tax_code": code, "basis": basis, "pay_to_date": pay45, "tax_to_date": tax45,
"ni_category": letter, "recorded_by": me.email, **p45}, _now())
repo.employees.save(row, _now())
hist = {"tax_codes": [{"valid_from": frm, "tax_code": code, "basis": basis, "source": source, "notice_ref": notice_ref}],
"ni_categories": [{"valid_from": frm, "category": letter}], "pension": [], "p45": []}
return _emp_view(row, hist)
@PRL.use_case(kind="command", route="POST /payroll-employees/{payroll_id}/details", permission=EMP_WRITE)
def complete_payroll_details(caller: dict, payroll_id: Ref, tax_code: Code, ni_category: Code,
source: Literal["p45", "starter_checklist", "p6", "manual"], basis: Literal["cumulative", "week1_month1"] = "cumulative",
valid_from: Optional[Day] = None, starter_declaration: Optional[Literal["A", "B", "C"]] = None,
p45_pay_to_date: Optional[AmountText] = None, p45_tax_to_date: Optional[AmountText] = None,
notice_ref: Optional[Note] = None) -> dict:
"""Complete a starter's payroll record (payroll_employee: pending_details -> active): the tax code and its basis, dated from
the start (or valid_from), the NI category, and where the code came from - a P45 (give its pay and tax to date: they count
in the cumulative tax, never in this employer's year to date), the starter checklist (its statement A, B or C), an HMRC
notice, or a migration. Needs prl:tax_code:write too. Nobody completes their own record, under either login. A starter with a
statement or a P45 has their own doors (record_starter_declaration, record_p45_in)."""
return _complete_details(caller, payroll_id, tax_code, ni_category, source, basis, valid_from, starter_declaration, p45_pay_to_date, p45_tax_to_date,
notice_ref)
def _emergency_code() -> str:
with repository() as repo:
return repo.settings.on("emergency_tax_code", _today(repo))
@PRL.use_case(kind="command", route="POST /payroll-employees/{payroll_id}/starter-declaration", permission=EMP_WRITE)
def record_starter_declaration(caller: dict, payroll_id: Ref, statement: Literal["A", "B", "C"], ni_category: Code = "A", valid_from: Optional[Day] = None,
notice_ref: Optional[Note] = None) -> dict:
"""Record the starter checklist statement of a starter with no P45 (PRL-02) and set their tax code from it: A (first job since 6 April,
no benefits) the emergency code (1257L) cumulative; B (the only job now, but another since 6 April) the emergency code on week 1 / month 1;
C (another job or a pension) BR cumulative. Dated from the start date, source starter_checklist; the statement goes on the first FPS.
NI category A unless told otherwise. payroll_employee: pending_details -> active. Nobody does their own."""
code, basis = {"A": (_emergency_code, "cumulative"), "B": (_emergency_code, "week1_month1"), "C": (lambda: "BR", "cumulative")}[statement]
return _complete_details(caller, payroll_id, code(), ni_category, "starter_checklist", basis, valid_from, statement, None, None, notice_ref)
@PRL.use_case(kind="command", route="POST /payroll-employees/{payroll_id}/emergency-code", permission=EMP_WRITE)
def use_emergency_code(caller: dict, payroll_id: Ref, ni_category: Code = "A", valid_from: Optional[Day] = None) -> dict:
"""A starter with neither a P45 nor a statement by the first run: the emergency code (1257L) on week 1 / month 1 (PRL-02's emergency rule),
source emergency. The run it is used in carries a warning on that payslip until a statement, a P45 or a notice replaces it. A person
decides this, never the calculation silently."""
return _complete_details(caller, payroll_id, _emergency_code(), ni_category, "emergency", "week1_month1", valid_from, None, None, None,
"emergency code: no P45 and no starter statement")
@PRL.use_case(kind="command", route="POST /payroll-employees/{payroll_id}/p45", permission=EMP_WRITE)
def record_p45_in(caller: dict, payroll_id: Ref, tax_code: Code, previous_pay_to_date: AmountText, previous_tax_to_date: AmountText,
previous_paye_ref: Ref, previous_leaving_date: Day, week1_month1: bool = False, student_loan: bool = False, ni_category: Code = "A",
valid_from: Optional[Day] = None) -> dict:
"""Record the P45 a starter brought (PRL-01): the tax code and whether it is week 1 / month 1, the pay and tax to date at the previous
employer (they count in the cumulative tax, never in this employer's year to date), the previous PAYE reference and leaving date, and
whether a student loan was being deducted. Kept as data (prl_p45, direction in). payroll_employee: pending_details -> active."""
left = _date(previous_leaving_date, "previous_leaving_date")
return _complete_details(caller, payroll_id, tax_code, ni_category, "p45", "week1_month1" if week1_month1 else "cumulative", valid_from, None,
previous_pay_to_date, previous_tax_to_date, f"P45 from {previous_paye_ref}",
{"paye_ref": previous_paye_ref.strip().upper(), "leaving_date": left, "student_loan": student_loan, "source_ref": previous_paye_ref.strip().upper()})
@PRL.use_case(kind="query", route="GET /payroll-employees/{payroll_id}/p45/document", permission=None)
def get_p45_document(caller: dict, payroll_id: Ref) -> dict:
"""The P45 a leaver was given (Part 1A, PDF in base64 with its object key and sha256): the leaver's own from its pay date, a payroll
reader's of anyone (that read is written to the access log). Anyone else, or no P45, is a 404."""
with repository() as repo:
emp = repo.employees.get(payroll_id)
row = repo.employees.p45s([payroll_id]).get((payroll_id, "out")) if emp else None
doc = repo.documents.get(row["document_key"]) if row and row["document_key"] else None
if doc is None:
raise Refused(404, f"no P45 for {payroll_id}")
mine = _has(caller, READ_OWN) and _my_person(caller) == emp["person_id"]
today = _today(repo)
if not (mine and doc["published_on"] is not None and doc["published_on"] <= today) and not _has(caller, DOC_READ):
raise Refused(404, f"no P45 for {payroll_id}")
if not mine:
_log(repo, caller, pr.Payslip("", payroll_id, emp["person_id"], 0), doc["object_key"], "p45 document")
return {"object_key": doc["object_key"], "content_type": doc["content_type"], "sha256": doc["sha256"], "size_bytes": doc["size_bytes"],
"content_base64": doc["content_b64"], "published_on": doc["published_on"].isoformat() if doc["published_on"] else None}
@PRL.use_case(kind="command", route="POST /payroll-employees/{payroll_id}/pension", permission=PENSION_WRITE, status=201)
def enrol_pension(caller: dict, payroll_id: Ref, scheme_code: Code, enrolled_on: Day) -> dict:
"""Enrol an employment in a pension scheme from a date (pension_membership: none -> enrolled); contributions start in the
run whose pay date is on or after it. Nobody enrols themselves through this door (an employee opts out as an employee)."""
when = _date(enrolled_on, "enrolled_on")
with repository() as repo, _decide():
row = repo.employees.get(payroll_id)
if row is None:
raise Refused(404, f"no payroll employee {payroll_id}")
if _my_person(caller) == row["person_id"]:
raise RuleBroken("You cannot change your own pension membership here", gate="not_own")
code = scheme_code.strip().upper()
if code not in repo.config.pension_schemes(row["company_code"], date(2100, 1, 1)):
raise RuleBroken(f"no pension scheme {code} for {row['company_code']}", gate="pension_scheme")
repo.employees.add_membership({"payroll_id": payroll_id, "scheme_code": code, "enrolled_on": when, "status": "enrolled",
"assessed_as": "eligible", "set_by": _who(caller)}, _now())
return {"payroll_id": payroll_id, "scheme_code": code, "enrolled_on": when.isoformat(), "status": "enrolled"}
# ── the pay run: open, calculate, approve ──────────────────────────────────────────────────────
@PRL.use_case(kind="command", route="POST /pay-runs", permission=RUN_OPEN, status=201)
def open_pay_run(caller: dict, company_code: Company, schedule_code: Code, period: Period) -> dict:
"""Open the pay run of a period (pay_run: none -> open), e.g. UK01, M, 2026-11 -> UK01-M-2026-27-M08 paid on the last
working day (2026-11-30, tax month 8); a weekly schedule takes an ISO week: UK01, W, 2026-W46 -> UK01-W-2026-27-W33 for the week
worked 9 to 15 November, paid Friday 2026-11-20 (tax week 33). The period is paid by ONE regular run (the period claim), the schedule's runs go in
order (the previous one approved first), and hr is asked to produce the period's payroll extract."""
cc, code = _company(company_code), schedule_code.strip().upper()
with repository() as repo, _decide():
sch = repo.config.schedule(cc, code)
if sch is None:
raise Refused(409, f"no pay schedule {code} for {cc}", gate="schedule", fix="record it first (put_pay_schedule)")
first, last, pay_date = pr.period_dates(sch["frequency"], period)
tax_year = paye.tax_year_label(paye.tax_year_of(pay_date))
runs = [r for r in repo.runs.of_schedule(cc, code) if r["kind"] == "regular" and r["status"] != "cancelled"]
later = [r["pay_run_id"] for r in runs if r["pay_date"] >= pay_date]
if later:
raise RuleBroken(f"{', '.join(later)} already pays this or a later period", gate="period_free")
before = [r for r in runs if r["pay_date"] < pay_date]
if before and before[-1]["status"] not in pr.FROZEN:
raise RuleBroken(f"the previous run {before[-1]['pay_run_id']} is {before[-1]['status']}: approve it first", gate="previous_approved")
base = pr.pay_run_id(cc, code, pay_date, 1, sch["frequency"])
attempt = 1 + sum(1 for r in repo.runs.of_schedule(cc, code) if r["pay_run_id"].split("-R")[0] == base)
run = pr.PayRun(pr.pay_run_id(cc, code, pay_date, attempt, sch["frequency"]), cc, code, sch["frequency"], "regular", tax_year,
paye.tax_period_of(sch["frequency"], pay_date),
period, first, last, pay_date, sch["paye_ref"])
me = _me(caller)
run.opened_by, run.opened_sub, run.hr_batch_id = me.email, me.sub, seams.batch_id(cc, period)
repo.periods.claim((cc, code, tax_year, pr.period_key(run.frequency, run.tax_period), "regular"), run.pay_run_id, _now())
repo.runs.create(run, _now())
view = _run_view(repo, run)
try: # after the commit: hr produces the period's draft extract (re-produced at calculation)
got = seams.produce_extract(cc, period, _who(caller))
view["extract"] = {"batch_id": got.get("batch_id"), "state": got.get("state"), "lines": got.get("lines")}
except seams.Unavailable as down:
view["extract"] = {"batch_id": run.hr_batch_id, "error": down.reason}
return view
def _load_for_calc(repo, run: pr.PayRun, gathered: dict) -> dict:
ids = [G.payroll_id(*k) for k in gathered["people"]]
today = _today(repo)
stat_ids, lo, hi = SP.history_window(gathered["people"]) # statutory pay averages earlier pay: read only when someone needs it
return {"employees": repo.employees.many(ids), "facts": repo.employees.facts(ids, run.pay_date),
"history": repo.runs.history(run, [G.payroll_id(i, n) for (i, n) in gathered["people"] if i in stat_ids], lo, hi),
"schemes": repo.config.pension_schemes(run.company_code, run.pay_date), "elements": repo.config.elements(run.company_code, run.pay_date),
"ytd": repo.runs.ytd_before(run, ids), "rates": repo.config.rates(run.pay_date),
"divisor": int(repo.settings.on("day_rate_divisor", today)),
"deduct_overtaken": repo.settings.on("deduct_overtaken_holiday", today) == "true",
"proration": repo.settings.on("proration", today)}
def _inputs(run: pr.PayRun, gathered: dict, loaded: dict) -> tuple:
if loaded["proration"] != "working_days":
raise paye.UnsupportedHere(f"proration by {loaded['proration']} is not calculated yet", gate="proration")
return G.inputs(run, gathered, loaded["employees"], loaded["facts"], loaded["schemes"], loaded["elements"], loaded["ytd"], loaded["rates"],
loaded["divisor"], loaded["deduct_overtaken"], loaded["history"])
@PRL.use_case(kind="command", route="POST /pay-runs/{pay_run_id}/calculate", permission=RUN_CALC)
def calculate_pay_run(caller: dict, pay_run_id: Ref) -> dict:
"""Calculate the run (open -> calculated), or recalculate it (calculated -> calculated: the draft payslips are replaced).
Reads hr's extract, time-off's absences and the dated rows in force on the pay date, and works out for each person: gross
(salary / 12, a part period prorated by working days in hr's work pattern; unpaid leave at salary / 260 a day), pension on
qualifying earnings by the net pay arrangement, PAYE cumulative on taxable pay to date, Class 1 NI category A, net pay.
Every line names its source and the rate rows it used. Refused, naming each person, while anyone has no tax code or NI
category; a starter becomes a payroll employee waiting for those details."""
actor = _who(caller)
with repository() as repo, _decide():
run = _run(repo, pay_run_id, lines=False)
pr.move(run.status, "calculate_pay_run" if run.status == "open" else "recalculate_pay_run")
with _decide():
gathered = G.gather(run, actor)
with repository() as repo: # starters (and anyone new to this payroll) wait for their tax details - committed first
have = repo.employees.many([G.payroll_id(*k) for k in gathered["people"]])
new = [{"payroll_id": G.payroll_id(*k), "person_id": k[0], "n": k[1], "company_code": run.company_code, "schedule_code": run.schedule_code,
"status": pr.move("none", "take_starter", pr.EMPLOYEE_MOVES),
"start_date": date.fromisoformat(p["starter"]["start_date"]) if p.get("starter") else None,
"leaving_date": date.fromisoformat(p["leaver"]["last_day"]) if p.get("leaver") else None}
for k, p in sorted(gathered["people"].items()) if G.payroll_id(*k) not in have]
repo.employees.add(new, _now())
with repository() as repo, _decide():
run = _run(repo, pay_run_id, lines=False)
old = {p.payroll_id: p for p in run.payslips}
run.move("calculate_pay_run" if run.status == "open" else "recalculate_pay_run")
loaded = _load_for_calc(repo, run, gathered)
people, missing = _inputs(run, gathered, loaded)
if missing:
raise RuleBroken("; ".join(f"{m} has no tax code or NI category" for m in missing), gate="tax_details",
fix="Employees: complete the record (P45 or starter declaration)")
run.calc_no += 1
rates = paye.Rates([{**r, "frequency": r["frequency"]} for r in loaded["rates"]])
slips = [pr.gross_to_net(run, i, rates) for i in people] # a payslip that left the run keeps its row at the old calc_no (never shown)
me = _me(caller)
run.prepared_by, run.prepared_sub, run.calculated_at = me.email, me.sub, _now()
stale_gone = [p for pid, p in old.items() if pid not in {s.payroll_id for s in slips}]
repo.runs.save_payslips(slips, _now())
repo.runs.add_lines(slips, _now())
repo.runs.save_root(run, _now())
run.payslips = slips
out = _run_view(repo, run, lines=True)
out["removed"] = [p.payroll_id for p in stale_gone]
return out
def _gates(repo, caller: dict, run: pr.PayRun, stale: dict, my_person: Optional[int]) -> list:
today = _today(repo)
return pr.approval_gates(run, _me(caller), my_person, today, int(repo.settings.on("bacs_lead_working_days", today)), stale)
@PRL.use_case(kind="query", route="GET /pay-runs", permission=RUN_READ)
def list_pay_runs(caller: dict, company_code: Optional[Company] = None) -> dict:
"""The pay runs, newest pay date first: status, pay date, who calculated and who approved. Payroll roles only."""
with repository() as repo:
rows = repo.runs.runs(_company(company_code) if company_code else None)
return {"pay_runs": [{"pay_run_id": r["pay_run_id"], "company_code": r["company_code"], "kind": r["kind"], "status": r["status"],
"period": r["period_code"], "pay_date": r["pay_date"].isoformat(), "tax_year": r["tax_year"], "tax_period": r["tax_period"],
"prepared_by": r["prepared_by"], "approved_by": r["approved_by"]} for r in rows]}
@PRL.use_case(kind="query", route="GET /pay-runs/{pay_run_id}")
def get_pay_run(caller: dict, pay_run_id: Ref, lines: bool = False) -> dict:
"""One run whole: its payslips (lines=true: every line with its source and rate rows), the totals, and while it is calculated
the gates approve_pay_run checks (stale payslips, negative net, the Bacs lead time, the second person, the approver's own
payslip). After approval: the journal and the payouts it led to. Payroll roles only: anyone else gets a 404."""
with repository() as repo:
if not _has(caller, RUN_READ):
raise Refused(404, f"no pay run {pay_run_id}")
run = _run(repo, pay_run_id, lines=lines)
gates = None
if run.status == "calculated":
try: # a read of the run never fails because hr is down: the own-payslip gate says so
mine, unknown = _my_person(caller), ""
except seams.Unavailable as down:
mine, unknown = None, down.reason
gates = _gates(repo, caller, run, {p.payroll_id: p.stale_reason for p in run.payslips if p.stale}, mine)
if unknown:
gates = [{**g, "met": None, "reason": f"not known: {unknown}"} if g["gate"] == "own_payslip_by_peer" else g for g in gates]
return _run_view(repo, run, lines=lines, gates=gates)
@PRL.use_case(kind="command", route="POST /pay-runs/{pay_run_id}/payslips/{payroll_id}/approve", permission=RUN_APPROVE)
def approve_approver_payslip(caller: dict, pay_run_id: Ref, payroll_id: Ref) -> dict:
"""Approve one payslip of a calculated run on its own - the approver's own payslip, approved by a PEER payroll manager
before the approver approves the run (approve_approver_payslip: calculated -> calculated). Never by the payslip's person
nor by the run's preparer, under either login. A recalculation needs it approved again."""
with repository() as repo, _decide():
run = _run(repo, pay_run_id, lines=False)
pr.move(run.status, "approve_approver_payslip")
slip = next((p for p in run.payslips if p.payroll_id == payroll_id), None)
if slip is None:
raise Refused(404, f"pay run {pay_run_id} has no payslip {payroll_id}")
me = _me(caller)
if _my_person(caller) == slip.person_id:
raise RuleBroken("Nobody approves their own payslip: a peer payroll manager does it", gate="own_payslip_by_peer")
if me.same_as(run.preparer()):
raise RuleBroken("You calculated this run: another payroll manager approves this payslip", gate="second_person")
if slip.stale:
raise RuleBroken(f"{payroll_id} is stale ({slip.stale_reason}): recalculate first", gate="not_stale")
repo.claims.once("payslip_approval", f"{pay_run_id}:{payroll_id}:c{run.calc_no}", _now(), _who(caller))
slip.approved_by, slip.approved_sub = me.email, me.sub
repo.runs.save_payslips([slip], _now())
return _slip_view(slip, lines=False)
def _issue_p45s(repo, run: pr.PayRun, idents: dict, publish: date, retain: date) -> list:
"""The leaver's P45 (issue_p45: active -> left), in the approval's transaction: the figures to date as data (prl_p45, direction out), the
document, and the payroll employee moved to left with the leaving date. Someone already left (a late payment) gets no second P45."""
slips = [p for p in run.payslips if p.payroll_id in idents]
emps = repo.employees.many([p.payroll_id for p in slips])
docs, rows = [], []
for p in slips:
emp = emps.get(p.payroll_id)
if emp is None or emp["status"] != "active":
continue
cumulative = p.basis_used == "cumulative"
pay = money(Decimal((p.ytd or {}).get("taxable", "0")) + (Decimal(emp["p45_pay_to_date"] or 0) if cumulative else 0))
tax = money(Decimal((p.ytd or {}).get("tax", "0")) + (Decimal(emp["p45_tax_to_date"] or 0) if cumulative else 0))
doc = documents.p45_document(run, p, idents[p.payroll_id], pay, tax, publish, retain)
docs.append(doc)
rows.append({"payroll_id": p.payroll_id, "direction": "out", "paye_ref": run.paye_ref, "leaving_date": p.leaving_date, "tax_code": p.tax_code_used,
"basis": p.basis_used, "pay_to_date": pay, "tax_to_date": tax, "student_loan": False, "ni_category": p.ni_category_used,
"source_ref": run.pay_run_id, "document_key": doc["object_key"], "recorded_by": run.approved_by})
emp["status"] = pr.move(emp["status"], "issue_p45", pr.EMPLOYEE_MOVES)
emp["leaving_date"] = p.leaving_date
repo.employees.save(emp, _now())
for r in rows:
repo.employees.add_p45(r, _now())
return docs
def _approval_effects(repo, run: pr.PayRun, today: date, idents: Optional[dict] = None) -> None:
"""In the approval's transaction: the journal and hr's 'sent' as PENDING rows, the payslip documents (and a leaver's P45), publication date."""
journal = {"journal_id": str(seams.journal_id(f"prl:{run.pay_run_id}", "approve")), "source_module": "payroll",
"source_document": f"prl:{run.pay_run_id}", "event": "approve", "company_code": run.company_code,
"journal_date": run.pay_date.isoformat(), "currency": "GBP", "memo": f"Payroll {run.pay_run_id}",
"lines": pr.journal_lines(run.payslips)}
run.journal_id, run.journal_status = int(journal["journal_id"]), "pending"
repo.outbox.add_many([("hr", f"{run.hr_batch_id}:sent", {"batch_id": run.hr_batch_id, "sent_ref": run.pay_run_id}),
("gl", f"prl:{run.pay_run_id}:approve", journal),
("fps", f"fps:{run.pay_run_id}:1", {"n": 1}), # the FPS and the pension file: built from the committed run after this commit
("pension", f"pension:{run.pay_run_id}", {})], run.pay_run_id, _now())
publish = run.pay_date if repo.settings.on("publish_on", today) == "pay_date" else today
retain = date(int(run.tax_year[:4]) + 1 + 6, 4, 5) # the tax year's end + 6 years (NMW regs. 59; PAYE reg. 97 needs 3)
docs = []
for p in run.payslips:
p.published_on = publish
doc = documents.payslip_document(run, p, publish, retain)
p.document_key = doc["object_key"]
docs.append(doc)
if idents:
docs += _issue_p45s(repo, run, idents, publish, retain)
repo.documents.add(docs, _now())
@PRL.use_case(kind="command", route="POST /pay-runs/{pay_run_id}/approve", permission=RUN_APPROVE)
def approve_pay_run(caller: dict, pay_run_id: Ref) -> dict:
"""Approve a calculated run (calculated -> approved; its payslips are frozen). Refused, with every unmet gate named, when a
payslip is stale (hr's extract, an absence, a payroll record or a rate changed since the calculation - checked again
now), a net pay is below 0.00, the pay date is fewer than bacs_lead_working_days (3) working days away, the caller
calculated the run, or the caller's own payslip is in it and no peer approved it. After the commit, from pending rows: hr's
extract is marked sent, ONE journal posts to gl, and once it has posted payments is asked to pay each net."""
actor = _who(caller)
with repository() as repo, _decide():
run = _run(repo, pay_run_id)
pr.move(run.status, "approve_pay_run")
with _decide(): # PRL-25: are these payslips still what hr, time-off and our rows say?
gathered = G.gather(run, actor)
with repository() as repo, _decide():
fresh, missing = _inputs(run, gathered, _load_for_calc(repo, run, gathered))
stale = G.stale_causes(run, fresh, missing)
if stale:
with repository() as repo:
marked = [p for p in run.payslips if p.payroll_id in stale and (not p.stale or p.stale_reason != stale[p.payroll_id])]
for p in marked:
p.stale, p.stale_reason = True, stale[p.payroll_id]
repo.runs.save_payslips(marked, _now())
raise Refused(409, "Cannot approve: not_stale: " + "; ".join(f"{k}: {v}" for k, v in sorted(stale.items())), gate="not_stale",
fix="recalculate the run")
with _decide(): # a leaver's identity goes into their P45 and nowhere else (HR-61): read before the commit
idents = {G.payroll_id(*k): seams.identity(k[0], run.pay_date, actor) for k, p in sorted(gathered["people"].items()) if p.get("leaver")}
with repository() as repo, _decide():
run = _run(repo, pay_run_id)
run.move("approve_pay_run")
gates = _gates(repo, caller, run, {p.payroll_id: p.stale_reason for p in run.payslips if p.stale}, _my_person(caller))
unmet = [g for g in gates if not g["met"]]
if unmet:
raise Refused(409, "Cannot approve: " + "; ".join(f"{g['gate']}: {g['reason']}" for g in unmet), gate=unmet[0]["gate"],
fix=unmet[0]["fix"], fields={"gates": gates})
repo.claims.once("approve", pay_run_id, _now(), actor)
me, today = _me(caller), _today(repo)
run.approved_by, run.approved_sub, run.approved_at, run.approved_on = me.email, me.sub, _now(), today
_approval_effects(repo, run, today, idents)
repo.runs.save_payslips(run.payslips, _now())
repo.runs.save_root(run, _now())
_drain(pay_run_id, actor)
with repository() as repo:
return _run_view(repo, _run(repo, pay_run_id, lines=False))
@PRL.use_case(kind="command", route="POST /pay-runs/{pay_run_id}/cancel", permission=RUN_CANCEL, destructive=True)
def cancel_pay_run(caller: dict, pay_run_id: Ref, reason: Reason) -> dict:
"""Cancel an open or calculated run with a reason (-> cancelled, terminal): nothing was sent, and the period is free again -
it is opened again as <run>-R2. An approved run is never cancelled: a correction is the next period's."""
if not reason.strip():
raise _bounds("a reason is required to cancel a run")
with repository() as repo, _decide():
run = _run(repo, pay_run_id, lines=False)
run.move("cancel_pay_run")
repo.claims.once("cancel", pay_run_id, _now(), _who(caller))
run.cancel_reason = reason.strip()
repo.periods.release((run.company_code, run.schedule_code, run.tax_year, pr.period_key(run.frequency, run.tax_period), run.kind))
repo.runs.save_root(run, _now())
return _run_view(repo, run)
# ── after the commit: hr's sent, the journal, the payouts - from pending rows ──────────────────────
def _payouts(run: pr.PayRun) -> list:
return [{"payout_ref": pr.payout_ref(run.pay_run_id, p.person_id), "batch_ref": run.pay_run_id, "payee_bp_id": str(p.bp_id),
"person_id": str(p.person_id), "company_code": run.company_code, "currency": "GBP", "amount": f"{p.net:.2f}",
"due_on": run.pay_date.isoformat(), "reference": f"SALARY {run.period_code}"[:18], "source_module": "payroll",
"source_document": run.pay_run_id, "clearing_purpose": "net_pay_payable"}
for p in run.payslips if p.net > 0 and p.bp_id is not None]
def _deliver_hr(row: dict, who: str) -> dict:
body = json.loads(row["payload_json"])
out = seams.mark_extract_sent(body["batch_id"], body["sent_ref"], who)
with repository() as repo:
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is None or fresh["status"] != "pending":
return {"ref": row["ref"], "status": "already sent"}
why = "" if out["ok"] else f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
repo.outbox.mark(fresh, "sent" if out["ok"] else ("refused" if not out["retry"] else "pending"), why, _now())
return {"ref": row["ref"], "status": "sent" if out["ok"] else ("refused" if not out["retry"] else "retry"), "reason": why or None}
def _deliver_gl(row: dict, who: str) -> dict:
"""Send the stored journal (after its commit); gl replays a repeat. Posted -> the payouts row is written in the SAME
transaction that records it. A closed period keeps it pending with gl's reason and holds the payouts until it posts."""
out = seams.post_journal(json.loads(row["payload_json"]), who)
with repository() as repo, _decide():
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is None or fresh["status"] != "pending":
return {"ref": row["ref"], "status": "already sent"}
run = _run(repo, row["pay_run_id"])
if out["ok"]:
repo.outbox.mark(fresh, "sent", "", _now())
run.journal_status, run.journal_reason = "posted", ""
pays = _payouts(run)
for p in run.payslips:
p.payout_ref = pr.payout_ref(run.pay_run_id, p.person_id) if p.net > 0 and p.bp_id is not None else ""
p.payout_status = "pending" if p.payout_ref else ("no_payee" if p.net > 0 else "nothing_to_pay")
if pays:
repo.outbox.add("payments", f"prl:{run.pay_run_id}:payouts", run.pay_run_id, {"payouts": pays}, _now())
repo.runs.save_payslips(run.payslips, _now())
repo.runs.save_root(run, _now())
file_out.advance_run(repo, run)
return {"ref": row["ref"], "status": "posted"}
why = f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
repo.outbox.mark(fresh, "pending", why, _now())
if not out["retry"]:
run.journal_status, run.journal_reason = "refused", why
repo.runs.save_root(run, _now())
return {"ref": row["ref"], "status": "refused" if not out["retry"] else "retry", "reason": why}
def _deliver_payouts(row: dict, who: str) -> dict:
"""Ask payments for each payout of the run (payments replays a known payout_ref). A refusal is recorded on that payslip;
a transient failure keeps the row pending, and the next pass asks again for all of them (replays are free)."""
asks = json.loads(row["payload_json"])["payouts"]
answers = {a["payout_ref"]: seams.request_payout(a, who) for a in asks}
with repository() as repo, _decide():
fresh = repo.outbox.get(row["target"], row["ref"])
if fresh is None or fresh["status"] != "pending":
return {"ref": row["ref"], "status": "already sent"}
run = _run(repo, row["pay_run_id"], lines=False)
changed = []
for p in run.payslips:
a = answers.get(p.payout_ref)
if a is None or p.payout_status in ("paid", "returned"):
continue
status = "requested" if a["ok"] else ("refused" if not a["retry"] else "pending")
reason = "" if a["ok"] else f"{a['gate'] + ': ' if a['gate'] else ''}{a['reason']}"
if (status, reason) != (p.payout_status, p.payout_reason):
p.payout_status, p.payout_reason = status, reason
changed.append(p)
retry = [r for r, a in answers.items() if not a["ok"] and a["retry"]]
repo.runs.save_payslips(changed, _now())
repo.outbox.mark(fresh, "pending" if retry else "sent", f"{len(retry)} payouts not answered: {answers[retry[0]]['reason']}" if retry else "",
_now())
return {"ref": row["ref"], "status": "retry" if retry else "requested", "payouts": len(asks), "unanswered": len(retry)}
def _drain(pay_run_id: Optional[str] = None, who: str = "payroll") -> list:
"""Send what is pending, oldest first: after a commit for one run, or everything (the worker's tick). One row failing
never stops the others; a journal that posts writes the payouts row, which the next pass takes."""
done, seen = [], set()
deliver = {"hr": _deliver_hr, "gl": _deliver_gl, "payments": _deliver_payouts, "fps": file_out.deliver_fps, "pension": file_out.deliver_pension,
"eps": file_out.deliver_eps, "gl_eps": file_out.deliver_eps_journal}
while True:
with repository() as repo:
rows = [r for r in repo.outbox.pending(pay_run_id) if (r["target"], r["ref"]) not in seen]
if not rows:
return done
for row in rows:
seen.add((row["target"], row["ref"]))
try:
done.append(deliver[row["target"]](row, who))
except Refused as lost:
done.append({"ref": row["ref"], "status": "lost a race", "reason": lost.reason})
@PRL.use_case(kind="command", route="POST /outbox/drain", permission=POST_AUTO, idempotent=True, tool=False)
def drain_outbox(caller: dict) -> dict:
"""The worker's tick (no server-side scheduler): send every pending hr 'sent', journal and payout batch. Safe at any time and
from two workers: every send is idempotent at the neighbour."""
return {"sent": _drain(None, _who(caller))}
@PRL.use_case(kind="command", route="POST /pay-runs/{pay_run_id}/payouts/{payout_ref}/paid", permission=MARK_PAID, idempotent=True, tool=False)
def confirm_payout(caller: dict, pay_run_id: Ref, payout_ref: Ref, status: Literal["paid", "returned"], paid_on: Day, amount: AmountText,
payment_id: Optional[Ref] = None, reason: Optional[Reason] = None) -> dict:
"""payments tells us the bank paid a net pay payout, or that it came back (PRL-28). Recorded once per payout_ref: the same
call answers the same; another amount is 409 {gate: amount_matches}. A returned payout leaves the payslip unpaid with the
bank's reason for the admin. Held by payments' service client alone."""
when = _date(paid_on, "paid_on")
paid = _money(amount, "amount")
with repository() as repo, _decide():
run = repo.runs.get(pay_run_id, lines=False)
slip = next((p for p in (run.payslips if run else []) if p.payout_ref and p.payout_ref == payout_ref), None)
if slip is None:
raise Refused(404, f"pay run {pay_run_id} requested no payout {payout_ref}")
if money(paid) != slip.net:
raise Refused(409, f"the bank moved {paid:.2f} but the payslip's net pay is {slip.net:.2f}", gate="amount_matches",
fix="payments and payroll must agree on the amount")
if slip.payout_status == status:
return _slip_view(slip, lines=False) # the same answer for the same call
if slip.payout_status in ("paid", "returned"):
raise RuleBroken(f"{payout_ref} is already {slip.payout_status}", gate="state")
repo.claims.once(status, payout_ref, _now(), _who(caller), payment_id or "")
slip.payout_status = status
if status == "paid":
slip.paid_on, slip.payment_id, slip.payout_reason = when, payment_id or "", ""
else:
slip.payout_reason = (reason or "returned by the bank").strip()
repo.runs.save_payslips([slip], _now())
file_out.advance_run(repo, run)
return _slip_view(slip, lines=False)
# ── payslips: payroll roles see a run's; an employee sees only their own, from the pay date ─────────
def _published(p: pr.Payslip, run_status: str, today: date) -> bool:
return run_status in pr.FROZEN and p.published_on is not None and p.published_on <= today
def _visible_slip(repo, caller: dict, pay_run_id: str, payroll_id: str) -> tuple:
"""(run, payslip, logged?) - a payroll reader sees any payslip (and their read of another person's is logged); an employee
only their own, once published; anyone else, or anything else, is a 404."""
run = repo.runs.get(pay_run_id)
slip = next((p for p in (run.payslips if run else []) if p.payroll_id == payroll_id), None)
if slip is None:
raise Refused(404, f"no payslip {payroll_id} in {pay_run_id}")
mine = _has(caller, READ_OWN) and _my_person(caller) == slip.person_id
if mine and _published(slip, run.status, _today(repo)):
return run, slip, False
if _has(caller, RUN_READ):
return run, slip, not mine
raise Refused(404, f"no payslip {payroll_id} in {pay_run_id}")
def _log(repo, caller: dict, slip: pr.Payslip, key: str, what: str) -> None:
me = _me(caller)
at = datetime.now(timezone.utc).replace(tzinfo=None) # an audit stamp is real time, never the business clock
repo.documents.log_read({"object_key": key, "reader": me.email or me.sub, "read_at": at, "reader_sub": me.sub,
"person_id": slip.person_id, "what": what}, _now())
@PRL.use_case(kind="query", route="GET /pay-runs/{pay_run_id}/payslips/{payroll_id}")
def get_payslip(caller: dict, pay_run_id: Ref, payroll_id: Ref) -> dict:
"""One payslip with every line, its source and the rate rows it used, and the figures to date. Payroll roles see any
payslip (a read of another person's is recorded in the access log); an employee sees their own from its pay date."""
with repository() as repo:
run, slip, logged = _visible_slip(repo, caller, pay_run_id, payroll_id)
if logged:
_log(repo, caller, slip, slip.document_key or f"payslip:{pay_run_id}:{payroll_id}", "payslip")
return {**_slip_view(slip), "pay_date": run.pay_date.isoformat(), "tax_year": run.tax_year, "tax_period": run.tax_period}
@PRL.use_case(kind="query", route="GET /pay-runs/{pay_run_id}/payslips/{payroll_id}/document")
def get_payslip_document(caller: dict, pay_run_id: Ref, payroll_id: Ref) -> dict:
"""The payslip as its stored PDF (base64) with its object key and sha256: the employee's own from the pay date; a payroll
role's read of another person's is recorded in the access log."""
with repository() as repo:
run, slip, logged = _visible_slip(repo, caller, pay_run_id, payroll_id)
doc = repo.documents.get(slip.document_key) if slip.document_key else None
if doc is None:
raise Refused(404, f"payslip {payroll_id} in {pay_run_id} has no document yet: it is made when the run is approved")
if logged:
_log(repo, caller, slip, doc["object_key"], "payslip document")
return {"object_key": doc["object_key"], "content_type": doc["content_type"], "sha256": doc["sha256"], "size_bytes": doc["size_bytes"],
"content_base64": doc["content_b64"], "published_on": doc["published_on"].isoformat() if doc["published_on"] else None}
@PRL.use_case(kind="query", route="GET /my/payslips", permission=READ_OWN)
def list_my_payslips(caller: dict) -> dict:
"""My payslips: the caller's own (their person in hr, by their sign-in), each from its pay date - never before, never anyone
else's. Summaries; get_payslip and get_payslip_document give the lines and the PDF."""
with repository() as repo:
pid = _my_person(caller)
if pid is None:
return {"payslips": []}
today = _today(repo)
rows = [r for r in repo.runs.payslips_of_person(pid) if r["published_on"] is not None and r["published_on"] <= today]
return {"payslips": [{"pay_run_id": r["pay_run_id"], "payroll_id": r["payroll_id"], "pay_date": r["pay_date"].isoformat(),
"tax_year": r["tax_year"], "tax_period": r["tax_period"], "gross": _s(r["gross_amount"]), "net": _s(r["net_amount"]),
"tax_code": r["tax_code_used"], "document_key": r["document_key"]} for r in rows]}
# What goes out after an approval (the FPS, the pension file, the EPS) declares its use cases on the same brick; imported last so every helper above exists.
from brick_payroll.application import filing as file_out # noqa: E402,F401
from brick_payroll.application import ytd # noqa: E402,F401 (the year-to-date load: declares its use cases on the brick)