dodil/erp-brick-time-off
DODIL ERP brick: time off
src/brick_time_off/application/settings.py
"""What the customer edits as rows (TOF-11, TOF-14): dated settings, leave types and policies - a new rule for next year is entered, not programmed.
Every row is DATED and a plain INSERT (history is never rewritten): a row for a key that already has one must start AFTER today, so what was in force on
any past day is what the ledger saw; the first row of a key may start today. The controls that decide how far a balance may go into the red or what a year end
carries (allow_negative_days, carry_over_max_days, overlap_mode, manager_records_sickness, business_timezone) are the OWNER's - tof:settings:controls, which
the HR user they control never holds. A policy for employees is refused below the statutory minimum (WTR 1998 regs. 13 and 13A: 5.6 weeks, 28 days for a
five-day week), counting the bank holidays only as far as the fewest any UK nation has.
GAPS (AGENTS.md): the spec names a permission tof:settings:write that is not in its catalog - non-control settings use tof:policy:write; the bank-holiday
count of the statutory check is a constant (8) because this brick keeps no calendar; one policy row's leave_year_start cannot differ from the policy it
replaces mid-year (the entitlement would jump) and is not checked."""
import json
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Annotated, Literal, Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from erp_platform import auth
from erp_platform.bricks import Refused
from pydantic import Field
from brick_time_off.application import use_cases as hub
from brick_time_off.application.inputs import Code, DateText, Note
from brick_time_off.domain.values import SETTING_DEFS
TOF = hub.TOF
CONTROLS, POLICY_WRITE, LEAVE_TYPE_WRITE = "tof:settings:controls", "tof:policy:write", "tof:leave_type:write"
CONTROL_KEYS = ("allow_negative_days", "carry_over_max_days", "overlap_mode", "manager_records_sickness", "business_timezone")
WEEKS_MINIMUM_DAYS = Decimal("28") # 5.6 weeks of a five-day week: WTR 1998 reg. 13 and 13A (the cap of 28 is the same figure)
MIN_BANK_HOLIDAYS = 8 # the fewest any UK nation has (England and Wales); a person's own nation is hr's, not known here
PAY_TREATMENTS = ("paid", "unpaid", "sick", "statutory_maternity", "statutory_paternity", "statutory_adoption", "statutory_shared_parental")
Value = Annotated[str, Field(max_length=64)]
MMDD = Annotated[str, Field(max_length=5)]
Amount = Annotated[str, Field(max_length=24)]
def _mmdd(v: str, what: str) -> str:
try:
month, dom = (int(x) for x in v.split("-"))
date(2024, month, dom) # 2024: 02-29 is a real day
except (ValueError, TypeError):
raise hub._bounds(f"{what} is a day of the year like 03-31") from None
return f"{month:02d}-{dom:02d}"
def _decimal(v: str, what: str, low: Decimal, high: Decimal, places: int = 4) -> Decimal:
try:
d = Decimal(str(v).strip())
except InvalidOperation:
raise hub._bounds(f"{what} {str(v)[:24]!r} is not a number") from None
if not d.is_finite() or not low <= d <= high:
raise hub._bounds(f"{what} is between {low} and {high}")
if d != d.quantize(Decimal(1).scaleb(-places)):
raise hub._bounds(f"{what} has at most {places} decimals")
return d
def _setting_value(key: str, raw: str) -> str:
"""A setting's value in its own type and range, or a 422 in words."""
raw = raw.strip()
if key in ("allow_negative_days", "carry_over_max_days"):
return str(_decimal(raw, key, Decimal(0), Decimal(365)))
if key == "overlap_threshold_pct":
return str(_decimal(raw, key, Decimal(0), Decimal(100)))
if key in ("max_days_ahead", "fit_note_after_days", "approval_reminder_days"):
high = {"max_days_ahead": 1100, "fit_note_after_days": 365, "approval_reminder_days": 60}[key]
return str(int(_decimal(raw, key, Decimal(0), Decimal(high), 0)))
if key == "carry_over_expires":
return _mmdd(raw, key)
if key in ("min_booking", "overlap_mode"):
allowed = {"min_booking": ("half_day", "full_day", "hour"), "overlap_mode": ("warn", "block")}[key]
if raw not in allowed:
raise hub._bounds(f"{key} is one of {', '.join(allowed)}")
return raw
if key in ("manager_records_sickness", "leaver_cancel_future"):
if raw.lower() not in ("true", "false"):
raise hub._bounds(f"{key} is true or false")
return raw.lower()
if key == "business_timezone":
try:
ZoneInfo(raw)
except (ZoneInfoNotFoundError, ValueError, OSError):
raise hub._bounds(f"{raw[:40]!r} is not a time zone like Europe/London") from None
return raw
raise hub._bounds(f"no such setting {key[:40]!r}")
def _dated(repo, rows_exist: bool, valid_from: date, today: date) -> None:
"""A row for something that already has one starts after today; the first row may start today (never in the past)."""
if valid_from < today or (rows_exist and valid_from <= today):
raise Refused(422, f"{valid_from} is not after today ({today}): what was in force on a past day is never rewritten", gate="bounds",
fix="date it for a day to come" if rows_exist else "date it today or later")
@TOF.use_case(kind="command", route="POST /settings", status=201)
def change_time_off_setting(caller: dict, key: Code, value: Value, valid_from: DateText) -> dict:
"""Set how time off is booked, approved and chased from a date (a new dated row; nothing is overwritten). The controls on balances - allow_negative_days,
carry_over_max_days, overlap_mode, manager_records_sickness, business_timezone - are the owner's alone (tof:settings:controls, 403 for HR); every other
setting is HR's (tof:policy:write). The first row of a key may start today; a later one must start after today, so the year end that already ran still
reads the cap that was in force."""
if key not in SETTING_DEFS:
raise hub._bounds(f"no such setting {key[:40]!r}: one of {', '.join(sorted(SETTING_DEFS))}")
if key in CONTROL_KEYS: # the owner may set any setting; HR every one but the controls
hub._need(caller, CONTROLS)
elif not auth.has_permission(caller, CONTROLS):
hub._need(caller, POLICY_WRITE)
start, clean = hub._date(valid_from, "valid_from"), _setting_value(key, value)
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
have = [r for r in repo.settings.rows() if r["setting_key"] == key]
if any(r["valid_from"] == start for r in have):
raise Refused(409, f"{key} already has a row from {start}", gate="dated_once", fix="date the new value for another day")
_dated(repo, bool(have), start, today)
repo.settings.put(key, clean, start, hub._who(caller), hub._now())
return {"key": key, "value": clean, "valid_from": start.isoformat(), "set_by": hub._who(caller), "controlled_by_owner": key in CONTROL_KEYS}
@TOF.use_case(kind="query", route="GET /settings")
def list_time_off_settings(caller: dict) -> dict:
"""Every setting with the value in force today, its default, who may change it and the dated rows behind it (HR and the owner)."""
if not (auth.has_permission(caller, POLICY_WRITE) or auth.has_permission(caller, CONTROLS) or auth.has_permission(caller, hub.READ_ALL)):
raise Refused(403, f"missing permission: {POLICY_WRITE}")
with hub.repository() as repo:
today = hub._today(repo)
return {"on": today.isoformat(), "settings": [
{"key": k, "value": repo.settings.on(k, today), "default": d[0], "changes": d[1], "owner_only": k in CONTROL_KEYS,
"rows": [{"valid_from": r["valid_from"].isoformat(), "value": r["value"], "set_by": r["set_by"]} for r in repo.settings.rows() if r["setting_key"] == k]}
for k, d in sorted(SETTING_DEFS.items())]}
# ── leave types ──────────────────────────────────────────────────────────────────────────────
def _type_view(r: dict) -> dict:
return {"company_code": r["company_code"], "leave_type_code": r["leave_type_code"], "valid_from": r["valid_from"].isoformat(), "name": r["name"],
"pay_treatment": r["pay_treatment"], "recorded_or_requested": r["recorded_or_requested"], "counts_against": r["counts_against"],
"evidence_rule": r["evidence_rule"], "max_days_per_year": r["max_days_per_year"]}
@TOF.use_case(kind="command", route="POST /leave-types", permission=LEAVE_TYPE_WRITE, status=201)
def put_time_off_leave_type(caller: dict, company_code: Code, leave_type_code: Code, name: Note, pay_treatment: Literal[PAY_TREATMENTS],
valid_from: DateText, recorded_or_requested: Literal["requested", "recorded"] = "requested", counts_against: Code = "none",
evidence_rule: Literal["none", "fit_note", "matb1"] = "none", max_days_per_year: Optional[Annotated[int, Field(ge=1, le=366)]] = None) -> dict:
"""Enter a kind of time off for a company from a date (HR): its pay treatment is one of paid | unpaid | sick | statutory_maternity | statutory_paternity |
statutory_adoption | statutory_shared_parental - and only payroll turns it into money. counts_against names the leave type whose policy it spends ('none' for
a type with no balance, like unpaid leave). A new row for a type the company already has starts after today. The company's own row wins over the default."""
if not name.strip():
raise Refused(422, "a leave type has a name", gate="bounds")
start = hub._date(valid_from, "valid_from")
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
own = repo.catalog.leave_type(company_code, leave_type_code, date(2100, 1, 1))
has_own = own is not None and own["company_code"] == company_code
if has_own and own["valid_from"] == start:
raise Refused(409, f"{leave_type_code} already has a row from {start}", gate="dated_once")
_dated(repo, has_own, start, today)
if counts_against != "none" and repo.catalog.leave_type(company_code, counts_against, start) is None and counts_against != leave_type_code:
raise Refused(409, f"counts_against {counts_against} is not a leave type of {company_code}", gate="leave_type")
row = {"company_code": company_code, "leave_type_code": leave_type_code, "valid_from": start, "name": name.strip(), "pay_treatment": pay_treatment,
"recorded_or_requested": recorded_or_requested, "counts_against": counts_against, "evidence_rule": evidence_rule, "max_days_per_year": max_days_per_year}
repo.catalog.add_leave_type(row, hub._now())
return _type_view(row)
@TOF.use_case(kind="query", route="GET /leave-types")
def list_time_off_leave_types(caller: dict, company_code: Code, on_date: Optional[DateText] = None) -> dict:
"""The kinds of time off in force for a company on a date (default today): the company's own rows over the defaults every company starts with."""
hub._need(caller, hub.READ_OWN)
when = hub._date(on_date, "on_date") if on_date else None
with hub.repository() as repo:
on = when or hub._today(repo)
codes = sorted({r["leave_type_code"] for r in repo.catalog.type_rows(company_code)})
return {"company_code": company_code, "on": on.isoformat(),
"leave_types": [_type_view(t) for c in codes if (t := repo.catalog.leave_type(company_code, c, on)) is not None]}
# ── policies ─────────────────────────────────────────────────────────────────────────────────
def statutory_shortfall(unit: str, annual_days: Decimal, bank_holidays: str) -> Optional[str]:
"""None when an annual-leave policy gives at least 5.6 weeks, else the words: the days plus the fewest bank holidays a UK nation has (when they are on
top), against 28 days for a five-day week. `annual_days` is for a FULL-TIME person; a part-timer's share is derived by FTE at read, never below pro rata."""
counted = annual_days + (MIN_BANK_HOLIDAYS if bank_holidays == "on_top" else 0)
if counted < WEEKS_MINIMUM_DAYS:
what = f"{annual_days.normalize():f} days" + (f" plus {MIN_BANK_HOLIDAYS} bank holidays" if bank_holidays == "on_top" else " with the bank holidays inside")
return f"{what} is below 5.6 weeks (28 days for a five-day week, WTR reg. 13 and 13A)"
return None
@TOF.use_case(kind="command", route="POST /policies", permission=POLICY_WRITE, status=201)
def put_time_off_policy(caller: dict, company_code: Code, policy_code: Code, leave_type_code: Code, unit: Literal["days", "hours"], annual_days: Amount,
bank_holidays: Literal["included", "on_top"], valid_from: DateText, rounding: Literal["half_day_up", "none"] = "half_day_up",
fte_min: Optional[Amount] = None, fte_below: Optional[Amount] = None, carry_over_max: Optional[Amount] = None,
carry_over_expires: MMDD = "03-31", leave_year_start: MMDD = "01-01") -> dict:
"""Enter a policy - how much of a leave type a group of people gets - from a date (HR), a new dated row: annual_days for a FULL-TIME person (prorated by
employment days and FTE on read), whether bank holidays are on top or inside, the rounding, the carry-over cap (days) and expiry (MM-DD) and the leave year's
start (MM-DD). `fte_min` / `fte_below` say who it applies to (a policy picks by FTE on the first day of the leave year). A policy of ANNUAL leave below the
statutory minimum is refused 422 {gate: statutory_minimum}: 18 days plus 8 bank holidays is below 5.6 weeks. A new row for a policy the company already
has starts after today."""
start = hub._date(valid_from, "valid_from")
days = _decimal(annual_days, "annual_days", Decimal(0), Decimal(366))
cap = _decimal(carry_over_max, "carry_over_max", Decimal(0), Decimal(366)) if carry_over_max is not None else None
applies = {}
for k, v in (("fte_min", fte_min), ("fte_below", fte_below)):
if v is not None:
applies[k] = f"{_decimal(v, k, Decimal(0), Decimal(1)):.4f}"
expires, year_start = _mmdd(carry_over_expires, "carry_over_expires"), _mmdd(leave_year_start, "leave_year_start")
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
lt = repo.catalog.leave_type(company_code, leave_type_code, start)
if lt is None:
raise Refused(409, f"{leave_type_code} is not a leave type of {company_code} on {start}", gate="leave_type")
if lt["counts_against"] in (None, "none"):
raise Refused(409, f"{leave_type_code} has no balance, so no policy counts against it", gate="leave_type", fix="enter the leave type with counts_against first")
if leave_type_code == "ANNUAL":
short = statutory_shortfall(unit, days, bank_holidays)
if short:
raise Refused(422, short, gate="statutory_minimum", fix="give at least 28 days counting the bank holidays")
mine = repo.catalog.policy_rows(company_code, policy_code)
if any(r["valid_from"] == start for r in mine):
raise Refused(409, f"{policy_code} already has a row from {start}", gate="dated_once")
_dated(repo, bool(mine), start, today)
row = {"company_code": company_code, "policy_code": policy_code, "valid_from": start, "leave_type_code": leave_type_code,
"applies_json": json.dumps(applies, sort_keys=True), "unit": unit, "annual_days": days, "bank_holidays": bank_holidays, "accrual": "front_loaded",
"accrual_pct": None, "leave_year_start": year_start, "rounding": rounding, "carry_over_max": cap, "carry_over_expires": expires}
repo.catalog.add_policy(row, hub._now())
return _policy_view(row)
def _policy_view(r: dict) -> dict:
return {"company_code": r["company_code"], "policy_code": r["policy_code"], "valid_from": r["valid_from"].isoformat(), "leave_type_code": r["leave_type_code"],
"applies": json.loads(r["applies_json"] or "{}"), "unit": r["unit"], "annual_days": format(Decimal(str(r["annual_days"])).normalize(), "f"),
"bank_holidays": r["bank_holidays"], "rounding": r["rounding"], "carry_over_max": None if r["carry_over_max"] is None else str(r["carry_over_max"]),
"carry_over_expires": r["carry_over_expires"], "leave_year_start": r["leave_year_start"]}
@TOF.use_case(kind="query", route="GET /policies")
def list_time_off_policies(caller: dict, company_code: Code, on_date: Optional[DateText] = None) -> dict:
"""The policies in force for a company on a date (default today), per leave type: the company's own rows over the defaults (HR and the owner)."""
if not (auth.has_permission(caller, POLICY_WRITE) or auth.has_permission(caller, hub.READ_ALL)):
raise Refused(403, f"missing permission: {POLICY_WRITE}")
when = hub._date(on_date, "on_date") if on_date else None
with hub.repository() as repo:
on = when or hub._today(repo)
return {"company_code": company_code, "on": on.isoformat(),
"policies": [_policy_view(p) for code in sorted({t["leave_type_code"] for t in repo.catalog.type_rows(company_code)}) for p in repo.catalog.policies(company_code, code, on)]}