dodil/erp-brick-time-off
DODIL ERP brick: time off
src/brick_time_off/infrastructure/tables.py
"""The tables, and one repository per concern. Every table is prefixed `tof_`, has a natural primary key (never an autoincrement),
and a quantity is NUMERIC(19,4). schema.sql is written from the same column model; tests/test_schema.py keeps the two (and the
migrations) equal.
Rows that are FACTS (a day claim, a ledger entry, a claim key, a leaver notice) are written with a plain INSERT: a second writer of
the same key is rejected (23505), never silently overwritten - DataK3's ON CONFLICT DO UPDATE rewrites every column in VALUES. Only
the request row, which moves FORWARD, is written with the keyed upsert. Statements per request do not grow with the days of a
request or with the rows of other people: every list and every balance is a fixed number of queries.
"""
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, delete, func, insert, or_, select
from brick_time_off.domain import adjustment as adj
from brick_time_off.domain import request as rq
from brick_time_off.domain import sickness as sk
from brick_time_off.domain.calendar import Slot, q4
from brick_time_off.domain.errors import RuleBroken
from brick_time_off.domain.values import DEFAULT_LEAVE_TYPES, DEFAULT_POLICIES, SETTING_DEFS
md = MetaData()
QTY = Numeric(19, 4)
tof_request = Table("tof_request", md,
Column("request_id", String, primary_key=True),
Column("person_id", BigInteger),
Column("company_code", String),
Column("requester_email", String),
Column("requester_sub", String),
Column("leave_type_code", String),
Column("policy_code", String),
Column("leave_year", Integer),
Column("from_date", Date),
Column("from_part", String),
Column("to_date", Date),
Column("to_part", String),
Column("cost", QTY),
Column("unit", String),
Column("status", String),
Column("revision", Integer),
Column("approver_person_id", BigInteger),
Column("approver_email", String),
Column("approver_sub", String),
Column("approved_by", String),
Column("approved_sub", String),
Column("decided_at", TIMESTAMP),
Column("reason", String),
Column("note", String),
Column("overlap_ack", String),
Column("stale", Boolean),
Column("stale_cost", QTY),
Column("cancel_asked_at", TIMESTAMP),
Column("cancel_asked_by", String),
Column("cancel_refused_reason", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The row a half day of one person races for: a plain INSERT, so of two racing writes exactly one wins. Released by DELETE in the
# same transaction as the withdrawal, rejection or cancellation. quantity is what the half costs in the request's unit.
tof_day_claim = Table("tof_day_claim", md,
Column("person_id", BigInteger, primary_key=True),
Column("absence_date", Date, primary_key=True),
Column("part", String, primary_key=True),
Column("absence_ref", String),
Column("minutes", Integer),
Column("quantity", QTY),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# Append-only: an entry is never edited or deleted. quantity is SIGNED (a take is negative, its reversal positive).
tof_balance_entry = Table("tof_balance_entry", md,
Column("entry_ref", String, primary_key=True),
Column("person_id", BigInteger),
Column("leave_type_code", String),
Column("leave_year", Integer),
Column("kind", String),
Column("quantity", QTY),
Column("unit", String),
Column("on_date", Date),
Column("by_email", String),
Column("by_sub", String),
Column("reason", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One guard row per bounded operation on a person or a ledger: the operation takes the NEXT n for its key in its own transaction, so two
# racers cannot both succeed and a loser is a 409. scopes: series, decide, cancel, cancel_ask, leaver.
tof_claim = Table("tof_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),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
tof_leave_type = Table("tof_leave_type", md,
Column("company_code", String, primary_key=True),
Column("leave_type_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("name", String),
Column("pay_treatment", String),
Column("recorded_or_requested", String),
Column("counts_against", String),
Column("evidence_rule", String),
Column("max_days_per_year", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
tof_policy = Table("tof_policy", md,
Column("company_code", String, primary_key=True),
Column("policy_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("leave_type_code", String),
Column("applies_json", String),
Column("unit", String),
Column("annual_days", Numeric(9, 4)),
Column("bank_holidays", String),
Column("accrual", String),
Column("accrual_pct", Numeric(9, 4)),
Column("leave_year_start", String),
Column("rounding", String),
Column("carry_over_max", Numeric(9, 4)),
Column("carry_over_expires", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
tof_settings = Table("tof_settings", md,
Column("setting_key", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# hr told us a person left (POST /employees/{person_id}/leaver): one row per hr source_ref, and the answer given.
tof_leaver = Table("tof_leaver", md,
Column("source_ref", String, primary_key=True),
Column("person_id", BigInteger),
Column("n", Integer),
Column("last_day", Date),
Column("cancelled_json", String),
Column("balance_json", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One spell of sickness (TOF-07): recorded, not requested. reason_detail is SENSITIVE (only the sick person and tof:sickness:read_detail read it).
# qualifying_days, piw, linked_to and linked_group are the facts payroll needs for statutory sick pay; they are refreshed from hr when the spell is
# recorded, extended or closed. changed_at is when payroll last needs to look again.
tof_sickness = Table("tof_sickness", md,
Column("absence_id", String, primary_key=True),
Column("person_id", BigInteger),
Column("company_code", String),
Column("first_day", Date),
Column("expected_return", Date),
Column("last_day_off", Date),
Column("unit", String),
Column("status", String),
Column("revision", Integer),
Column("reason_detail", String),
Column("self_certified", Boolean),
Column("evidence_ok", Boolean),
Column("recorded_by", String),
Column("recorded_sub", String),
Column("person_email", String),
Column("person_sub", String),
Column("approver_person_id", BigInteger),
Column("approver_email", String),
Column("approver_sub", String),
Column("confirmed_by", String),
Column("confirmed_sub", String),
Column("void_reason", String),
Column("qualifying_days", Integer),
Column("piw", Boolean),
Column("linked_to", String),
Column("linked_group", String),
Column("changed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A fit note (or another document) backing an absence, stored as the object raw/time-off/<absence_ref>/<sha256>.<ext>. Never overwritten.
tof_evidence = Table("tof_evidence", md,
Column("object_key", String, primary_key=True),
Column("absence_ref", String),
Column("kind", String),
Column("covers_from", Date),
Column("covers_to", Date),
Column("sha256", String),
Column("content_type", String),
Column("size_bytes", Integer),
Column("uploaded_by", String),
Column("uploaded_sub", String),
Column("uploaded_at", TIMESTAMP),
Column("retain_until", Date),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A day the company closes (TOF-13): published once per company and date. It is a FACT about the company, not a booking per person: what a person's
# allowance pays for is DERIVED on read from these rows and hr's days, so a starter gets it with nobody recalculating and nothing needs a roster.
tof_closure_day = Table("tof_closure_day", md,
Column("company_code", String, primary_key=True),
Column("closure_date", Date, primary_key=True),
Column("name", String),
Column("leave_type_code", String),
Column("published_by", String),
Column("published_sub", String),
Column("published_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A manual adjustment (TOF-12): proposed by one person, decided by another; only an approved one is a ledger entry (entry_ref = adjustment_id).
tof_adjustment = Table("tof_adjustment", md,
Column("adjustment_id", String, primary_key=True),
Column("person_id", BigInteger),
Column("company_code", String),
Column("leave_type_code", String),
Column("leave_year", Integer),
Column("quantity", QTY),
Column("unit", String),
Column("on_date", Date),
Column("reason", String),
Column("proposed_by", String),
Column("proposed_sub", String),
Column("status", String),
Column("decided_by", String),
Column("decided_sub", String),
Column("decided_at", TIMESTAMP),
Column("decision_reason", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One Odoo migration run (TOF-23): loaded -> reconciled -> signed | discarded. checks_json holds the reconciliation (each check with Odoo's number, ours and
# the owner's explanation); a SIGNED run is frozen.
tof_migration_run = Table("tof_migration_run", md,
Column("run_id", String, primary_key=True),
Column("odoo_db", String),
Column("cutover_date", Date),
Column("company_code", String),
Column("status", String),
Column("doors_json", String),
Column("checks_json", String),
Column("summary_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),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# (Odoo db, Odoo model, Odoo id) -> our natural key (shared.yaml conventions.odoo_migration): what a later brick resolves through, and how a re-load matches.
tof_migration_map = Table("tof_migration_map", md,
Column("odoo_db", String, primary_key=True),
Column("odoo_model", String, primary_key=True),
Column("odoo_id", String, primary_key=True),
Column("ours", String),
Column("run_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The statutory facts of a family leave (TOF-09) - maternity, paternity, adoption, shared parental - that payroll reads to assess statutory pay (it computes
# the money): the expected week of childbirth and when the evidence arrived. The leave itself is a tof_request (recorded, approved by its recording). The
# keeping-in-touch days are tof_claim rows of scope kit_day (key request_id, n = 1..10, ref = the date).
tof_family_leave = Table("tof_family_leave", md,
Column("request_id", String, primary_key=True),
Column("person_id", BigInteger),
Column("company_code", String),
Column("leave_type_code", String),
Column("ewc", Date),
Column("evidence_received_on", Date),
Column("weeks", Integer),
Column("recorded_by", String),
Column("recorded_sub", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
class AlreadyDone(RuleBroken):
"""The one-shot claim was already taken: a 409 with these words."""
def _utc(dt: Optional[datetime]) -> Optional[datetime]:
"""Stored timestamps are naive UTC."""
if dt is None:
return None
return dt.astimezone(timezone.utc).replace(tzinfo=None) if dt.tzinfo else dt
def _plain(conn, table: Table, row: dict, at: datetime) -> None:
"""A FACT row: a plain INSERT - a second writer of the same key is rejected (23505), never overwritten."""
conn.execute(insert(table).values(**row, created_at=_utc(at), updated_at=_utc(at)))
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 _dec(v) -> Optional[Decimal]:
return None if v is None else q4(v)
class RequestRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _make(r: dict) -> rq.Request:
return rq.Request(
request_id=r["request_id"], person_id=r["person_id"], company_code=r["company_code"],
requester=rq.Identity(r["requester_email"] or "", r["requester_sub"] or ""), leave_type_code=r["leave_type_code"],
policy_code=r["policy_code"] or "", leave_year=r["leave_year"], from_date=r["from_date"], from_part=r["from_part"],
to_date=r["to_date"], to_part=r["to_part"], cost=q4(r["cost"]), unit=r["unit"], approver_person_id=r["approver_person_id"],
approver=rq.Identity(r["approver_email"] or "", r["approver_sub"] or ""), status=r["status"], revision=r["revision"],
approved_by=r["approved_by"] or "", approved_sub=r["approved_sub"] or "", decided_at=r["decided_at"], reason=r["reason"] or "",
note=r["note"] or "", overlap_ack=r["overlap_ack"] or "", stale=bool(r["stale"]), stale_cost=_dec(r["stale_cost"]),
cancel_asked_at=r["cancel_asked_at"], cancel_asked_by=r["cancel_asked_by"] or "",
cancel_refused_reason=r["cancel_refused_reason"] or "", created_at=r["created_at"])
@staticmethod
def _row(r: rq.Request) -> dict:
return {"request_id": r.request_id, "person_id": r.person_id, "company_code": r.company_code, "requester_email": r.requester.email,
"requester_sub": r.requester.sub, "leave_type_code": r.leave_type_code, "policy_code": r.policy_code or None,
"leave_year": r.leave_year, "from_date": r.from_date, "from_part": r.from_part, "to_date": r.to_date, "to_part": r.to_part,
"cost": r.cost, "unit": r.unit, "status": r.status, "revision": r.revision, "approver_person_id": r.approver_person_id,
"approver_email": r.approver.email or None, "approver_sub": r.approver.sub or None, "approved_by": r.approved_by or None,
"approved_sub": r.approved_sub or None, "decided_at": _utc(r.decided_at), "reason": r.reason or None, "note": r.note or None,
"overlap_ack": r.overlap_ack or None, "stale": r.stale, "stale_cost": r.stale_cost,
"cancel_asked_at": _utc(r.cancel_asked_at), "cancel_asked_by": r.cancel_asked_by or None,
"cancel_refused_reason": r.cancel_refused_reason or None}
def get(self, request_id: str) -> Optional[rq.Request]:
row = self.conn.execute(select(tof_request).where(tof_request.c.request_id == request_id)).mappings().first()
return self._make(dict(row)) if row else None
def create(self, r: rq.Request, at: datetime) -> None:
_plain(self.conn, tof_request, self._row(r), at)
def save(self, r: rq.Request, at: datetime) -> None:
"""The request moves forward (status, decision, ask): a keyed upsert of the whole row, creation stamp kept."""
upsert(self.conn, tof_request, [{**self._row(r), "created_at": _utc(r.created_at or at), "updated_at": _utc(at)}], ("request_id",))
# ── lists: one query each ────────────────────────────────────────────────────────────────
def summaries(self, *, mine: Optional[rq.Identity] = None, waiting_for: Optional[rq.Identity] = None, everything: bool = False,
person_id: Optional[int] = None, limit: int = 200) -> list:
conds = []
if everything:
conds.append(tof_request.c.request_id.isnot(None))
if mine is not None:
own = [tof_request.c.requester_sub == mine.sub] if mine.sub else []
own += [tof_request.c.requester_email == mine.email.lower()] if mine.email else []
conds.append(or_(*own) if own else tof_request.c.request_id.is_(None))
if waiting_for is not None:
w = [tof_request.c.approver_sub == waiting_for.sub] if waiting_for.sub else []
w += [tof_request.c.approver_email == waiting_for.email.lower()] if waiting_for.email else []
conds.append((tof_request.c.status == "requested") & (or_(*w) if w else tof_request.c.request_id.is_(None)))
if person_id is not None:
conds.append(tof_request.c.person_id == person_id)
if not conds:
return []
rows = self.conn.execute(select(tof_request).where(or_(*conds)).order_by(tof_request.c.request_id).limit(limit)).mappings()
return [dict(r) for r in rows]
def of_person_year(self, person_id: int, leave_type_code: str, leave_year: int) -> list:
"""{request_id: (status, ...)} source for the balance: the person's requests of one type and leave year, one query."""
return _all(self.conn, tof_request, (tof_request.c.person_id == person_id) & (tof_request.c.leave_type_code == leave_type_code)
& (tof_request.c.leave_year == leave_year), (tof_request.c.request_id,))
def of_company(self, company_code: str, first: date, last: date, person_id: Optional[int], changed_since: Optional[datetime]) -> list:
"""The company's approved requests touching [first, last] (a cancelled one only when payroll asks what changed since): ONE query."""
cond = (tof_request.c.company_code == company_code) & (tof_request.c.from_date <= last) & (tof_request.c.to_date >= first)
if person_id is not None:
cond &= tof_request.c.person_id == person_id
if changed_since is not None:
cond &= tof_request.c.status.in_(("approved", "cancelled")) & (tof_request.c.updated_at >= _utc(changed_since))
else:
cond &= tof_request.c.status == "approved"
return _all(self.conn, tof_request, cond, (tof_request.c.person_id, tof_request.c.request_id))
def open_of_person(self, person_id: int) -> list:
"""The person's requests that can still change: requested or approved."""
return _all(self.conn, tof_request, (tof_request.c.person_id == person_id) & tof_request.c.status.in_(("requested", "approved")),
(tof_request.c.request_id,))
class DayClaimRepository:
def __init__(self, conn):
self.conn = conn
def in_range(self, person_id: int, first: date, last: date) -> list:
"""Every half day the person already holds in [first, last]: ONE query however long the range."""
return _all(self.conn, tof_day_claim, (tof_day_claim.c.person_id == person_id) & (tof_day_claim.c.absence_date >= first)
& (tof_day_claim.c.absence_date <= last), (tof_day_claim.c.absence_date, tof_day_claim.c.part))
def of_refs(self, refs: list, first: date, last: date) -> list:
"""The halves held by these absences within [first, last]: ONE query."""
if not refs:
return []
return _all(self.conn, tof_day_claim, tof_day_claim.c.absence_ref.in_(refs) & (tof_day_claim.c.absence_date >= first)
& (tof_day_claim.c.absence_date <= last), (tof_day_claim.c.person_id, tof_day_claim.c.absence_date, tof_day_claim.c.part))
def add(self, person_id: int, ref: str, slots: list, at: datetime) -> None:
"""The halves of a request in ONE statement, plain INSERTs: a half somebody else holds is rejected at the key."""
if not slots:
return
self.conn.execute(insert(tof_day_claim), [{"person_id": person_id, "absence_date": s.date, "part": s.part, "absence_ref": ref,
"minutes": s.minutes, "quantity": s.quantity, "claimed_at": _utc(at),
"created_at": _utc(at), "updated_at": _utc(at)} for s in slots])
def release(self, person_id: int, ref: str, first: date, last: date) -> None:
"""Give the halves of one request back: ONE DELETE, in the same transaction as the move that frees them."""
self.conn.execute(delete(tof_day_claim).where((tof_day_claim.c.person_id == person_id) & (tof_day_claim.c.absence_ref == ref)
& (tof_day_claim.c.absence_date >= first) & (tof_day_claim.c.absence_date <= last)))
def release_after(self, person_id: int, ref: str, after: date, last: date) -> None:
self.conn.execute(delete(tof_day_claim).where((tof_day_claim.c.person_id == person_id) & (tof_day_claim.c.absence_ref == ref)
& (tof_day_claim.c.absence_date > after) & (tof_day_claim.c.absence_date <= last)))
@staticmethod
def slots(rows: list, ref: str) -> list:
return [Slot(r["absence_date"], r["part"], r["minutes"], q4(r["quantity"])) for r in rows if r["absence_ref"] == ref]
class EntryRepository:
"""The ledger: append-only."""
def __init__(self, conn):
self.conn = conn
def add(self, entry_ref: str, person_id: int, leave_type_code: str, leave_year: int, kind: str, quantity: Decimal, unit: str,
on_date: date, by_email: str, by_sub: str, reason: str, at: datetime) -> None:
_plain(self.conn, tof_balance_entry, {"entry_ref": entry_ref, "person_id": person_id, "leave_type_code": leave_type_code,
"leave_year": leave_year, "kind": kind, "quantity": quantity, "unit": unit, "on_date": on_date,
"by_email": by_email or None, "by_sub": by_sub or None, "reason": reason or None}, at)
def of_person_year(self, person_id: int, leave_type_code: str, leave_year: int) -> list:
return _all(self.conn, tof_balance_entry, (tof_balance_entry.c.person_id == person_id)
& (tof_balance_entry.c.leave_type_code == leave_type_code) & (tof_balance_entry.c.leave_year == leave_year),
(tof_balance_entry.c.on_date, tof_balance_entry.c.entry_ref))
def refs(self, refs: list) -> set:
"""Which of these entry_refs are already written: ONE query (the idempotency check of a job that must write nothing twice)."""
if not refs:
return set()
return {r[0] for r in self.conn.execute(select(tof_balance_entry.c.entry_ref).where(tof_balance_entry.c.entry_ref.in_(refs)))}
def of_kind(self, kind: str, leave_type_code: str, leave_year: int) -> list:
return _all(self.conn, tof_balance_entry, (tof_balance_entry.c.kind == kind) & (tof_balance_entry.c.leave_type_code == leave_type_code)
& (tof_balance_entry.c.leave_year == leave_year), (tof_balance_entry.c.person_id, tof_balance_entry.c.entry_ref))
def persons_known(self, leave_type_code: str, leave_year: int) -> list:
"""The people this brick has anything for in a leave year of a type (a request or an entry): the roster of a year-end pass that was not given one."""
a = select(tof_balance_entry.c.person_id).where((tof_balance_entry.c.leave_type_code == leave_type_code) & (tof_balance_entry.c.leave_year == leave_year))
b = select(tof_request.c.person_id).where((tof_request.c.leave_type_code == leave_type_code) & (tof_request.c.leave_year == leave_year))
return sorted({r[0] for r in self.conn.execute(a.union(b))})
class ClaimRepository:
"""The one guard key per bounded thing: every operation that touches it takes the next claim in ITS transaction."""
def __init__(self, conn):
self.conn = conn
def count(self, scope: str, key: str) -> int:
return self.conn.execute(select(func.count()).select_from(tof_claim).where(
(tof_claim.c.claim_scope == scope) & (tof_claim.c.claim_key == key))).scalar() or 0
def take(self, scope: str, key: str, at: datetime, by: str = "", ref: str = "") -> int:
"""The next n for a key that counts (a request number): n = committed claims + 1; a racer with the same n loses at the INSERT."""
n = self.count(scope, key) + 1
_plain(self.conn, tof_claim, {"claim_scope": scope, "claim_key": key, "n": n, "ref": ref, "claimed_by": by, "claimed_at": _utc(at)}, at)
return n
def once(self, scope: str, key: str, at: datetime, by: str = "", ref: str = "") -> None:
"""A key that can be claimed exactly ONCE (the decision of one revision): already claimed -> the operation was done a moment ago;
two racers both see none, both insert n=1, and 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 request again")
_plain(self.conn, tof_claim, {"claim_scope": scope, "claim_key": key, "n": 1, "ref": ref, "claimed_by": by, "claimed_at": _utc(at)}, at)
class CatalogRepository:
"""Leave types and policies: dated rows, the company's own over the '*' default, the latest valid_from on or before the date."""
def __init__(self, conn):
self.conn = conn
def leave_type(self, company_code: str, code: str, on: date) -> Optional[dict]:
rows = _all(self.conn, tof_leave_type, tof_leave_type.c.company_code.in_((company_code, "*")) & (tof_leave_type.c.leave_type_code == code)
& (tof_leave_type.c.valid_from <= on), (tof_leave_type.c.valid_from,))
own = [r for r in rows if r["company_code"] != "*"]
return (own or rows or [None])[-1]
def type_rows(self, company_code: str) -> list:
"""Every leave type row of the company and of the defaults, any date: ONE query."""
return _all(self.conn, tof_leave_type, tof_leave_type.c.company_code.in_((company_code, "*")), (tof_leave_type.c.leave_type_code, tof_leave_type.c.valid_from))
def add_leave_type(self, row: dict, at: datetime) -> None:
_plain(self.conn, tof_leave_type, row, at)
def policy_rows(self, company_code: str, policy_code: str) -> list:
return _all(self.conn, tof_policy, (tof_policy.c.company_code == company_code) & (tof_policy.c.policy_code == policy_code))
def add_policy(self, row: dict, at: datetime) -> None:
_plain(self.conn, tof_policy, row, at)
def policies(self, company_code: str, leave_type_code: str, on: date) -> list:
"""The policies of a leave type in force on a date (latest row of each policy_code; a company's own over '*'), in code order."""
rows = _all(self.conn, tof_policy, tof_policy.c.company_code.in_((company_code, "*")) & (tof_policy.c.leave_type_code == leave_type_code)
& (tof_policy.c.valid_from <= on), (tof_policy.c.valid_from,))
by_code: dict = {}
for r in rows:
cur = by_code.get(r["policy_code"])
if cur is None or (r["company_code"] != "*" or cur["company_code"] == "*"):
by_code[r["policy_code"]] = r
return [by_code[k] for k in sorted(by_code)]
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, tof_settings, None, (tof_settings.c.setting_key, tof_settings.c.valid_from))
return self._rows
def on(self, key: str, day: date) -> str:
"""The value in force on `day`: the latest row dated on or before it, else the default."""
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:
"""A zone row takes over once the business date, counted in the zone BEFORE it, has reached its valid_from - never by the UTC date."""
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, tof_settings, {"setting_key": key, "valid_from": valid_from, "value": str(value), "set_by": by}, at)
self._rows = None
class LeaverRepository:
def __init__(self, conn):
self.conn = conn
def get(self, source_ref: str) -> Optional[dict]:
row = self.conn.execute(select(tof_leaver).where(tof_leaver.c.source_ref == source_ref)).mappings().first()
return dict(row) if row else None
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, tof_leaver, row, at)
def set_balance(self, row: dict, balance: dict, at: datetime) -> None:
upsert(self.conn, tof_leaver, [{**{k: row[k] for k in ("source_ref", "person_id", "n", "last_day", "cancelled_json")},
"balance_json": json.dumps(balance, sort_keys=True), "created_at": row["created_at"],
"updated_at": _utc(at)}], ("source_ref",))
def last_day(self, person_id: int) -> Optional[date]:
"""The last day hr gave for this person: that of the highest source_ref number (a moved last day is n+1). One query."""
rows = self.conn.execute(select(tof_leaver.c.n, tof_leaver.c.last_day).where(tof_leaver.c.person_id == person_id)
.order_by(tof_leaver.c.n.desc()).limit(1)).first()
return rows[1] if rows else None
class SicknessRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _make(r: dict) -> sk.Sickness:
return sk.Sickness(
absence_id=r["absence_id"], person_id=r["person_id"], company_code=r["company_code"], first_day=r["first_day"],
expected_return=r["expected_return"], unit=r["unit"], status=r["status"], revision=r["revision"], last_day_off=r["last_day_off"],
reason_detail=r["reason_detail"] or "", self_certified=bool(r["self_certified"]), evidence_ok=bool(r["evidence_ok"]),
recorded_by=r["recorded_by"] or "", recorded_sub=r["recorded_sub"] or "", person=rq.Identity(r["person_email"] or "", r["person_sub"] or ""), approver_person_id=r["approver_person_id"],
approver=rq.Identity(r["approver_email"] or "", r["approver_sub"] or ""), confirmed_by=r["confirmed_by"] or "",
confirmed_sub=r["confirmed_sub"] or "", void_reason=r["void_reason"] or "", qualifying_days=r["qualifying_days"] or 0,
piw=bool(r["piw"]), linked_to=r["linked_to"] or "", linked_group=r["linked_group"] or "", changed_at=r["changed_at"],
created_at=r["created_at"])
@staticmethod
def _row(s: sk.Sickness) -> dict:
return {"absence_id": s.absence_id, "person_id": s.person_id, "company_code": s.company_code, "first_day": s.first_day,
"expected_return": s.expected_return, "last_day_off": s.last_day_off, "unit": s.unit, "status": s.status, "revision": s.revision,
"reason_detail": s.reason_detail or None, "self_certified": s.self_certified, "evidence_ok": s.evidence_ok,
"recorded_by": s.recorded_by or None, "recorded_sub": s.recorded_sub or None, "person_email": s.person.email or None, "person_sub": s.person.sub or None,
"approver_person_id": s.approver_person_id,
"approver_email": s.approver.email or None, "approver_sub": s.approver.sub or None, "confirmed_by": s.confirmed_by or None,
"confirmed_sub": s.confirmed_sub or None, "void_reason": s.void_reason or None, "qualifying_days": s.qualifying_days,
"piw": s.piw, "linked_to": s.linked_to or None, "linked_group": s.linked_group or None, "changed_at": _utc(s.changed_at)}
def get(self, absence_id: str) -> Optional[sk.Sickness]:
row = self.conn.execute(select(tof_sickness).where(tof_sickness.c.absence_id == absence_id)).mappings().first()
return self._make(dict(row)) if row else None
def create(self, s: sk.Sickness, at: datetime) -> None:
s.changed_at = at
_plain(self.conn, tof_sickness, self._row(s), at)
def save(self, s: sk.Sickness, at: datetime, *, touch: bool = True) -> None:
"""The spell moves forward: a keyed upsert of the whole row, creation stamp kept. `touch` stamps changed_at (what payroll's changed_since reads)."""
if touch:
s.changed_at = at
upsert(self.conn, tof_sickness, [{**self._row(s), "created_at": _utc(s.created_at or at), "updated_at": _utc(at)}], ("absence_id",))
def earlier_of_person(self, person_id: int, before: date) -> list:
"""[(absence_id, linked_group, first_day, end)] of the person's non-voided spells starting before `before`: one query."""
rows = _all(self.conn, tof_sickness, (tof_sickness.c.person_id == person_id) & (tof_sickness.c.status != "voided")
& (tof_sickness.c.first_day < before), (tof_sickness.c.first_day,))
return [(r["absence_id"], r["linked_group"], r["first_day"], self._make(r).end) for r in rows]
def summaries(self, *, person_id: Optional[int] = None, approver: Optional[rq.Identity] = None, everything: bool = False, limit: int = 200) -> list:
conds = []
if everything:
conds.append(tof_sickness.c.absence_id.isnot(None))
if person_id is not None:
conds.append(tof_sickness.c.person_id == person_id)
if approver is not None:
w = [tof_sickness.c.approver_sub == approver.sub] if approver.sub else []
w += [tof_sickness.c.approver_email == approver.email.lower()] if approver.email else []
conds.append(or_(*w) if w else tof_sickness.c.absence_id.is_(None))
if not conds:
return []
rows = self.conn.execute(select(tof_sickness).where(or_(*conds)).order_by(tof_sickness.c.absence_id).limit(limit)).mappings()
return [dict(r) for r in rows]
def of_company(self, company_code: str, first: date, last: date, changed_since: Optional[datetime]) -> list:
"""Every spell of the company that touches [first, last] (a voided one only when payroll asks what changed): ONE query."""
end = func.coalesce(tof_sickness.c.last_day_off, tof_sickness.c.expected_return)
cond = (tof_sickness.c.company_code == company_code) & (tof_sickness.c.first_day <= last) & (end >= first)
if changed_since is not None:
cond &= tof_sickness.c.changed_at >= _utc(changed_since)
else:
cond &= tof_sickness.c.status != "voided"
return _all(self.conn, tof_sickness, cond, (tof_sickness.c.person_id, tof_sickness.c.absence_id))
class EvidenceRepository:
def __init__(self, conn):
self.conn = conn
def get(self, object_key: str) -> Optional[dict]:
row = self.conn.execute(select(tof_evidence).where(tof_evidence.c.object_key == object_key)).mappings().first()
return dict(row) if row else None
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, tof_evidence, {**row, "uploaded_at": _utc(row["uploaded_at"])}, at)
def of_absence(self, absence_ref: str) -> list:
return _all(self.conn, tof_evidence, tof_evidence.c.absence_ref == absence_ref, (tof_evidence.c.covers_from, tof_evidence.c.object_key))
def of_absences(self, refs: list) -> list:
return _all(self.conn, tof_evidence, tof_evidence.c.absence_ref.in_(refs), (tof_evidence.c.absence_ref, tof_evidence.c.covers_from)) if refs else []
class ClosureRepository:
def __init__(self, conn):
self.conn = conn
def get(self, company_code: str, closure_date: date) -> Optional[dict]:
row = self.conn.execute(select(tof_closure_day).where((tof_closure_day.c.company_code == company_code)
& (tof_closure_day.c.closure_date == closure_date))).mappings().first()
return dict(row) if row else None
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, tof_closure_day, {**row, "published_at": _utc(row["published_at"])}, at)
def in_range(self, company_code: str, first: date, last: date, leave_type_code: Optional[str] = None) -> list:
"""The company's closure days in [first, last] (of one leave type when given): ONE query."""
cond = (tof_closure_day.c.company_code == company_code) & (tof_closure_day.c.closure_date >= first) & (tof_closure_day.c.closure_date <= last)
if leave_type_code:
cond &= tof_closure_day.c.leave_type_code == leave_type_code
return _all(self.conn, tof_closure_day, cond, (tof_closure_day.c.closure_date,))
class AdjustmentRepository:
def __init__(self, conn):
self.conn = conn
@staticmethod
def _make(r: dict) -> adj.Adjustment:
return adj.Adjustment(
adjustment_id=r["adjustment_id"], person_id=r["person_id"], company_code=r["company_code"], leave_type_code=r["leave_type_code"],
leave_year=r["leave_year"], quantity=q4(r["quantity"]), unit=r["unit"], on_date=r["on_date"], reason=r["reason"] or "",
proposed_by=rq.Identity(r["proposed_by"] or "", r["proposed_sub"] or ""), status=r["status"],
decided_by=rq.Identity(r["decided_by"] or "", r["decided_sub"] or ""), decided_at=r["decided_at"],
decision_reason=r["decision_reason"] or "", created_at=r["created_at"])
@staticmethod
def _row(a: adj.Adjustment) -> dict:
return {"adjustment_id": a.adjustment_id, "person_id": a.person_id, "company_code": a.company_code, "leave_type_code": a.leave_type_code,
"leave_year": a.leave_year, "quantity": a.quantity, "unit": a.unit, "on_date": a.on_date, "reason": a.reason,
"proposed_by": a.proposed_by.email or None, "proposed_sub": a.proposed_by.sub or None, "status": a.status,
"decided_by": a.decided_by.email or None, "decided_sub": a.decided_by.sub or None, "decided_at": _utc(a.decided_at),
"decision_reason": a.decision_reason or None}
def get(self, adjustment_id: str) -> Optional[adj.Adjustment]:
row = self.conn.execute(select(tof_adjustment).where(tof_adjustment.c.adjustment_id == adjustment_id)).mappings().first()
return self._make(dict(row)) if row else None
def create(self, a: adj.Adjustment, at: datetime) -> None:
_plain(self.conn, tof_adjustment, self._row(a), at)
def save(self, a: adj.Adjustment, at: datetime) -> None:
upsert(self.conn, tof_adjustment, [{**self._row(a), "created_at": _utc(a.created_at or at), "updated_at": _utc(at)}], ("adjustment_id",))
def listed(self, status: Optional[str], person_id: Optional[int], limit: int = 200) -> list:
cond = tof_adjustment.c.adjustment_id.isnot(None)
if status:
cond &= tof_adjustment.c.status == status
if person_id is not None:
cond &= tof_adjustment.c.person_id == person_id
return _all(self.conn, tof_adjustment, cond, (tof_adjustment.c.adjustment_id,))[:limit]
class MigrationRepository:
def __init__(self, conn):
self.conn = conn
def run(self, run_id: str) -> Optional[dict]:
row = self.conn.execute(select(tof_migration_run).where(tof_migration_run.c.run_id == run_id)).mappings().first()
return dict(row) if row else None
def runs(self) -> list:
return _all(self.conn, tof_migration_run, None, (tof_migration_run.c.run_id,))
def add_run(self, row: dict, at: datetime) -> None:
_plain(self.conn, tof_migration_run, row, at)
def save_run(self, row: dict, at: datetime) -> None:
upsert(self.conn, tof_migration_run, [{**{k: v for k, v in row.items() if k != "created_at"}, "updated_at": _utc(at)}], ("run_id",))
def mapped(self, db: str, model: str, ids: list) -> dict:
"""odoo_id -> map row for a set of Odoo ids: one query per 500."""
got: dict = {}
ids = sorted({str(i) for i in ids})
for i in range(0, len(ids), 500):
for r in _all(self.conn, tof_migration_map, (tof_migration_map.c.odoo_db == db) & (tof_migration_map.c.odoo_model == model)
& tof_migration_map.c.odoo_id.in_(ids[i:i + 500])):
got[r["odoo_id"]] = r
return got
def add_map(self, rows: list, at: datetime) -> None:
for i in range(0, len(rows), 500):
self.conn.execute(insert(tof_migration_map), [{**r, "created_at": _utc(at), "updated_at": _utc(at)} for r in rows[i:i + 500]])
def of_run(self, run_id: str, model: Optional[str] = None) -> list:
cond = tof_migration_map.c.run_id == run_id
if model:
cond &= tof_migration_map.c.odoo_model == model
return _all(self.conn, tof_migration_map, cond, (tof_migration_map.c.odoo_model, tof_migration_map.c.odoo_id))
def delete_run(self, run_id: str, request_ids: list, entry_refs: list) -> None:
"""Take a discarded run's rows back out: its requests, their half days and take entries, its opening entries and its map. ONLY for a run that was never
signed and whose requests are still as it loaded them (the use case checks); the one place a ledger entry is removed."""
for i in range(0, len(request_ids), 400):
chunk = request_ids[i:i + 400]
self.conn.execute(delete(tof_day_claim).where(tof_day_claim.c.absence_ref.in_(chunk)))
self.conn.execute(delete(tof_request).where(tof_request.c.request_id.in_(chunk)))
for i in range(0, len(entry_refs), 400):
self.conn.execute(delete(tof_balance_entry).where(tof_balance_entry.c.entry_ref.in_(entry_refs[i:i + 400])))
self.conn.execute(delete(tof_migration_map).where(tof_migration_map.c.run_id == run_id))
def add_requests(self, requests: list, at: datetime) -> None:
rows = [{**RequestRepository._row(r), "created_at": _utc(at), "updated_at": _utc(at)} for r in requests]
for i in range(0, len(rows), 200):
self.conn.execute(insert(tof_request), rows[i:i + 200])
def add_claims(self, rows: list, at: datetime) -> None:
rows = [{**r, "claimed_at": _utc(at), "created_at": _utc(at), "updated_at": _utc(at)} for r in rows]
for i in range(0, len(rows), 300):
self.conn.execute(insert(tof_day_claim), rows[i:i + 300])
def add_entries(self, rows: list, at: datetime) -> None:
rows = [{**r, "created_at": _utc(at), "updated_at": _utc(at)} for r in rows]
for i in range(0, len(rows), 300):
self.conn.execute(insert(tof_balance_entry), rows[i:i + 300])
class FamilyLeaveRepository:
def __init__(self, conn):
self.conn = conn
def get(self, request_id: str) -> Optional[dict]:
row = self.conn.execute(select(tof_family_leave).where(tof_family_leave.c.request_id == request_id)).mappings().first()
return dict(row) if row else None
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, tof_family_leave, row, at)
def of_requests(self, request_ids: list) -> dict:
if not request_ids:
return {}
return {r["request_id"]: r for r in _all(self.conn, tof_family_leave, tof_family_leave.c.request_id.in_(request_ids))}
def kit_days(self, request_ids: list) -> dict:
"""request_id -> the dates of its keeping-in-touch days: ONE query."""
out: dict = {}
if request_ids:
for r in _all(self.conn, tof_claim, (tof_claim.c.claim_scope == "kit_day") & tof_claim.c.claim_key.in_(request_ids), (tof_claim.c.claim_key, tof_claim.c.n)):
out.setdefault(r["claim_key"], []).append(r["ref"])
return out
class Repository:
"""Every repository on one connection (one unit of work)."""
def __init__(self, conn):
self.conn = conn
self.requests = RequestRepository(conn)
self.days = DayClaimRepository(conn)
self.entries = EntryRepository(conn)
self.claims = ClaimRepository(conn)
self.catalog = CatalogRepository(conn)
self.settings = SettingsRepository(conn)
self.leavers = LeaverRepository(conn)
self.sickness = SicknessRepository(conn)
self.evidence = EvidenceRepository(conn)
self.closures = ClosureRepository(conn)
self.migration = MigrationRepository(conn)
self.family = FamilyLeaveRepository(conn)
self.adjustments = AdjustmentRepository(conn)
def seed(self, at: Optional[datetime] = None) -> None:
"""The leave types and policies every company starts with, once (company '*'). Idempotent; the same rows migration 00002 inserts."""
at = at or work.now()
have_t = {r["leave_type_code"] for r in _all(self.conn, tof_leave_type, tof_leave_type.c.company_code == "*")}
for code, name, pay, rec, against, ev, mx in DEFAULT_LEAVE_TYPES:
if code not in have_t:
_plain(self.conn, tof_leave_type, {"company_code": "*", "leave_type_code": code, "valid_from": date(2000, 1, 1), "name": name,
"pay_treatment": pay, "recorded_or_requested": rec, "counts_against": against,
"evidence_rule": ev, "max_days_per_year": mx}, at)
have_p = {r["policy_code"] for r in _all(self.conn, tof_policy, tof_policy.c.company_code == "*")}
for code, ltype, unit, days, bh, rounding, applies in DEFAULT_POLICIES:
if code not in have_p:
_plain(self.conn, tof_policy, {"company_code": "*", "policy_code": code, "valid_from": date(2000, 1, 1), "leave_type_code": ltype,
"applies_json": applies, "unit": unit, "annual_days": Decimal(days), "bank_holidays": bh,
"accrual": "front_loaded", "accrual_pct": None, "leave_year_start": "01-01",
"rounding": rounding, "carry_over_max": Decimal("5"), "carry_over_expires": "03-31"}, at)