dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
src/brick_timesheets/application/use_cases.py
"""The use cases - what a person (or their agent) can do with a timesheet.
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 (an employee sees their own weeks; the approver the weeks routed to them; a project lead the lines on projects they
lead; finance and the owner every week). 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_stories.py has one test per story):
- 'today' is the date in business_timezone (SettingsRepository.timezone), never the UTC date
- nobody approves, returns or reopens their own week, under either login (email OR sub); the approver is fixed at submission
- one claim key per operation on a week revision (ts_claim): the loser of a race is a 409, never a second effect
- a neighbour is called only AFTER the local commit, from a pending ts_outbox row written in the transaction that made the state change; a
crash after commit is sent on the next drain, a rolled-back approval sends nothing
- state only moves forward; a returned week continues as revision n+1, an approved one is superseded by n+1 and never edited; every downstream
effect (cost journal, deliveries) moves by the DIFFERENCE between revisions
- no read-your-writes inside an open transaction (DataK3): every use case reads committed rows, writes, and reports from what it holds
"""
import json
from contextlib import contextmanager
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional
from zoneinfo import ZoneInfo
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
from erp_platform.types import BigId
from sqlalchemy.exc import IntegrityError
from brick_timesheets.application.inputs import (
Code,
Currency,
Day,
Descr,
Entries,
Instant,
Rate,
Reason,
Ref,
Value,
Week,
day,
)
from brick_timesheets.domain import sheet as sh
from brick_timesheets.domain.errors import RuleBroken
from brick_timesheets.domain.money import hours4
from brick_timesheets.domain.values import MODES, OVERTIME_RULES, ROUTES, SETTING_DEFS
from brick_timesheets.infrastructure import seams, tables
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
READ_OWN, WRITE_OWN, SUBMIT = "ts:sheet:read_own", "ts:sheet:write_own", "ts:sheet:submit"
READ_TEAM, READ_ALL, APPROVE, REOPEN = "ts:sheet:read_team", "ts:sheet:read_all", "ts:sheet:approve", "ts:sheet:reopen"
LINE_APPROVE, COST_RATE, ACTIVITY, CONTROLS = "ts:line:approve_project", "ts:cost_rate:write", "ts:activity:write", "ts:settings:controls"
# machine: held only by a service client, and by an EXACT grant - no wildcard reaches them
POST_AUTO, DELIVER, READ_APPROVED = "ts:cost:post_auto", "ts:delivery:report", "ts:hours:read_approved"
MACHINE = (POST_AUTO, DELIVER, READ_APPROVED, "ts:sheet:open_auto", "ts:billing_run:create", "ts:reminder:tick", "ts:migration:load",
"ts:migration:reconcile")
TS = Brick("timesheets", version="0.2.0", namespace="ts", ids_as_strings=True, machine_permissions=MACHINE, instructions=(
"Timesheets: a person records one ISO week at a time. Open the week (open_timesheet): hr's contracted hours and time-off's approved absence are "
"already in it. Add a line for each project item or internal cost centre (add_time_line) and book hours (set_time_entries), then submit "
"(submit_timesheet): every gate that is not met is named - each working day's contracted hours covered, every project bookable on its day, "
"internal cost centres open, a cost rate for the person. The approver is the person's manager from hr on the week's Monday, fixed there; "
"nobody approves their own week under either login. The approver approves (approve_timesheet) or returns it with a question "
"(return_timesheet, which continues the week as a new revision). Approval posts ONE labour journal (the project's cost centre debited, the "
"person's home cost centre credited, at the standard cost rate) and reports billable hours on sales-order projects to the order's service "
"line. Finance corrects an approved week by reopening it (reopen_timesheet): the new revision, once approved, reverses the old journal and "
"moves the order by the difference. Hours are exact strings in steps of the time increment. Finance bills a contract or manual project's month: the system "
"gathers its approved billable hours into a draft run (create_time_billing_run, priced from the product price list, each line claimed once), finance "
"releases it (release_time_billing_run) and ar receives ONE draft invoice; a correction is billed as its difference in the next run. Project leads and "
"finance read a project's hours and cost (get_project_hours). The owner moves off Odoo with load_ts_from_odoo, reconcile_ts_migration and "
"sign_ts_migration."))
# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.
TS.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. On DataK3 a racer's duplicate key (two approvals of one revision, two opens of one week) can surface at COMMIT, after the
use case has returned - so it is caught here, around the commit, and answered as the retry-safe 409 a conflict seen mid-use-case gets."""
from brick_timesheets.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 timesheet again and retry") from None
repository = _repository_default # tests and a composition swap in their own
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words, through both doors; a silent neighbour is its own status."""
try:
yield
except RuleBroken as broken:
raise Refused(409, str(broken)) 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 timesheet 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) -> sh.Identity:
"""Both of the caller's names: the same person under either is the same person."""
return sh.Identity((caller.get("email") or "").strip().lower(), (caller.get("sub") or "").strip())
def _now() -> datetime:
return work.now()
def _today(repo) -> date:
"""Business today: the date NOW (work.now()) in business_timezone - read from ONE place - never the UTC date."""
return _now().astimezone(ZoneInfo(repo.settings.timezone())).date()
def _need(caller: dict, perm: str) -> None:
if not auth.has_permission(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 _cur(v: str) -> str:
v = (v or "").strip().upper()
if len(v) != 3 or not v.isalpha() or not v.isascii():
raise _bounds(f"{v or 'a blank currency'!r} is not a currency code: use three letters such as GBP")
return v
def _date(v: Optional[str], what: str) -> date:
try:
d = day(v)
except ValueError:
raise _bounds(f"{what} is a date like 2026-11-01") 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 _week(v: str) -> date:
try:
monday = sh.parse_week(v)
except ValueError as e:
raise _bounds(str(e)) from None
if not date(2000, 1, 3) <= monday <= date(2099, 12, 28):
raise _bounds(f"week {v} is outside 2000 .. 2099")
return monday
def _h(v) -> str:
"""Hours as the people of the stories write them: two places, four when the step needs them."""
s = f"{hours4(v):.4f}"
return s[:-2] if s.endswith("00") else s
def _m(v) -> Optional[str]:
return None if v is None else f"{Decimal(v):.2f}"
def _inc(repo, on: date) -> int:
return int(repo.settings.on("time_increment_minutes", on))
# ── visibility: which rows ────────────────────────────────────────────────────────────────────
def _full_scope(caller: dict, s: sh.Sheet) -> bool:
me = _me(caller)
if auth.has_permission(caller, READ_ALL):
return True
if auth.has_permission(caller, READ_OWN) and me.same_as(s.person):
return True
return bool(auth.has_permission(caller, READ_TEAM) and s.approver is not None and me.same_as(s.approver))
def _person_of(caller: dict) -> int:
"""The caller's person in hr (by their sign-in)."""
p = seams.person_by_sub(_me(caller).sub, _who(caller)) if _me(caller).sub else None
if p is None:
raise Refused(409, "hr holds no person with your sign-in", gate="person", fix="ask HR to link your login to your person record")
return int(p["person_id"])
def _load(repo, caller: dict, sheet_id: str, revision: Optional[int] = None) -> tuple:
"""(the whole sheet, the project lead's person_id or None): a row the caller may not see is a 404. A project lead sees a submitted week's lines on
projects they lead and nothing else of it."""
s = repo.sheets.get(sheet_id, revision)
if s is None:
raise Refused(404, f"no timesheet {sheet_id}")
if _full_scope(caller, s):
return s, None
if auth.has_permission(caller, LINE_APPROVE) and s.status != "open":
pid = _person_of(caller)
if any(ln.lead_person_id == pid for ln in s.lines):
return s, pid
raise Refused(404, f"no timesheet {sheet_id}")
def _own(repo, caller: dict, sheet_id: str) -> sh.Sheet:
"""A week of the caller's own (and only theirs): another person's, even finance's, is a 404."""
s = repo.sheets.get(sheet_id)
if s is None or not _me(caller).same_as(s.person):
raise Refused(404, f"no timesheet {sheet_id}")
return s
# ── views ───────────────────────────────────────────────────────────────────────────────────
def _entry_view(e: sh.Entry) -> dict:
return {"work_date": e.work_date.isoformat(), "hours": _h(e.hours), "note": e.note or None}
def _line_view(ln: sh.Line) -> dict:
return {"line_no": ln.line_no, "kind": ln.kind, "activity_code": ln.activity_code or None, "project_id": ln.project_id or None,
"item_id": ln.item_id or None, "cost_centre": ln.cost_centre or None, "billable": ln.billable, "bill_via": ln.bill_via,
"sales_order_id": ln.sales_order_id or None, "so_line_no": ln.so_line_no, "product_id": ln.product_id or None,
"rate_group": ln.rate_group or None, "absence_ref": ln.absence_ref or None, "description": ln.description or None,
"lead_person_id": None if ln.lead_person_id is None else str(ln.lead_person_id), "hours": _h(ln.hours()),
"stale": ln.stale, "stale_reason": ln.stale_reason or None, "entries": [_entry_view(e) for e in ln.entries]}
def _view(repo, s: sh.Sheet, *, lead: Optional[int] = None, full: bool = True, readiness: Optional[list] = None, stale: Optional[dict] = None) -> dict:
lines = s.lines if lead is None else [ln for ln in s.lines if ln.lead_person_id == lead]
booked = s.booked()
by_project: dict = {}
for ln in lines:
if ln.kind == "project":
by_project[ln.project_id] = by_project.get(ln.project_id, Decimal(0)) + ln.hours()
out = {"sheet_id": s.sheet_id, "revision": s.revision, "person_id": str(s.person_id), "company_code": s.company_code, "week": sh.week_id(s.week_start),
"week_start": s.week_start.isoformat(), "status": s.status,
"revisions": [{"revision": n, "status": st} for n, st in sorted({**dict(repo.sheets.revisions(s.sheet_id)), s.revision: s.status}.items())],
"lines": [_line_view(ln) for ln in lines],
"project_hours": {p: _h(h) for p, h in sorted(by_project.items())}}
if lead is not None:
out["restricted"] = "the lines on projects you lead"
return out
out.update({"expected_hours": _h(s.expected_hours()), "booked_hours": _h(s.total()), "worked_hours": _h(s.total(("project", "internal"))),
"absence_hours": _h(s.total(("absence",))), "home_cost_centre": s.home_cost_centre or None,
"days": [{"date": d.isoformat(), "expected_hours": _h(Decimal(s.expected.get(d.isoformat(), 0)) / 60),
"booked_hours": _h(booked.get(d, Decimal(0)))} for d in s.days()],
"approval_route": s.approval_route or None,
"approver": None if s.approver is None else {"person_id": None if s.approver_person_id is None else str(s.approver_person_id),
"rerouted_from": s.rerouted_from},
"line_approvals": [{"line_no": a.line_no, "decided_by": a.decided_by} for a in s.approvals],
"approved_on": s.approved_on.isoformat() if s.approved_on else None, "approved_by": s.approved_by or None,
"journal_id": None if s.journal_id is None else str(s.journal_id), "reason": s.reason or None})
if readiness is not None:
out["readiness"] = readiness
if stale:
out["stale"] = {str(n): why for n, why in sorted(stale.items())}
if full and s.approved_at is not None:
box = repo.outbox.of_sheet(s.sheet_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 _summary(row: dict) -> dict:
return {"sheet_id": row["sheet_id"], "week": sh.week_id(row["week_start"]), "person_id": str(row["person_id"]), "revision": row["revision"],
"status": row["status"], "company_code": row["company_code"], "expected_hours": _h(Decimal(row["expected_minutes"] or 0) / 60)}
def _local_readiness(repo, s: sh.Sheet) -> list:
max_day = Decimal(repo.settings.on("max_day_hours", s.week_start))
return sh.readiness(gaps=sh.coverage_gaps(s), mode=repo.settings.on("missing_hours_mode", s.week_start), notes=sh.long_days(s, max_day), bookable=[],
closed=[], rates=[], stale={})[:2]
# ── neighbours' facts a week is checked against ─────────────────────────────────────────────────
def _first_day(ln: sh.Line, fallback: date) -> date:
return min((e.work_date for e in ln.entries), default=fallback)
def _absence_lines(s: sh.Sheet, absences: list, first_no: int) -> list:
"""time-off's approved absences of the week as read-only lines: one line per absence, hours per day (minutes / 60)."""
lines, n = [], first_no
for a in sorted(absences, key=lambda x: x.get("absence_ref") or ""):
per: dict = {}
for d in a.get("days") or []:
try:
when = date.fromisoformat(d["date"])
except (KeyError, ValueError):
continue
if s.week_start <= when <= s.week_start + timedelta(days=6):
per[when] = per.get(when, Decimal(0)) + Decimal(d.get("minutes") or 0) / 60
if per:
lines.append(sh.Line(n, "absence", activity_code=a.get("leave_type_code") or "ABSENCE", absence_ref=a.get("absence_ref") or "",
entries=[sh.Entry(w, hours4(h)) for w, h in sorted(per.items())]))
n += 1
return lines
def _absence_days(absences: list, s: sh.Sheet) -> dict:
out: dict = {}
for ln in _absence_lines(s, absences, 1):
for e in ln.entries:
out[e.work_date.isoformat()] = out.get(e.work_date.isoformat(), Decimal(0)) + e.hours
return out
def _terms(cache: dict, project_id: str, on: date, actor: str) -> Optional[dict]:
"""A project's time terms on a day, asked once per request (one call per distinct project and day, however many lines)."""
if (project_id, on) not in cache:
cache[(project_id, on)] = seams.time_terms(project_id, on, actor)
return cache[(project_id, on)]
def _project_problems(s: sh.Sheet, actor: str, cache: Optional[dict] = None) -> tuple:
"""For every project line: why it is not bookable on one of its days (projects' bookable list, one call per distinct DAY), and how its terms
differ from what the line copied (one call per distinct PROJECT). Returns ({line_no: why}, {project_id: what changed})."""
project_lines = [ln for ln in s.lines if ln.kind == "project" and ln.entries]
if not project_lines:
return {}, {}
problems: dict = {}
days = sorted({e.work_date for ln in project_lines for e in ln.entries})
listing = {d: {p["project_id"]: p for p in seams.bookable(s.person_id, d, actor)} for d in days}
for ln in project_lines:
for e in ln.entries:
p = listing[e.work_date].get(ln.project_id)
if p is None:
problems[ln.line_no] = f"{ln.project_id} is not bookable on {e.work_date}: it is closed, cancelled or not yours to book to"
break
if ln.item_id:
item = next((i for i in p.get("items") or [] if i.get("item_id") == ln.item_id), None)
if item is None or item.get("state") in ("done", "closed", "cancelled"):
problems[ln.line_no] = f"{ln.project_id} item {ln.item_id} is not open on {e.work_date}"
break
changed: dict = {}
for pid in sorted({ln.project_id for ln in project_lines}):
mine = [ln for ln in project_lines if ln.project_id == pid]
t = _terms({} if cache is None else cache, pid, min(_first_day(ln, s.week_start) for ln in mine), actor)
if t is None:
changed[pid] = f"{pid} no longer exists in projects"
continue
was = mine[0]
now = (t.get("billing") or "non_billable", t.get("sales_order_id") or "", t.get("cost_centre") or "", t.get("lead_person_id"))
had = (was.billing, was.sales_order_id, was.cost_centre, was.lead_person_id)
if t.get("state") not in (None, "active"):
changed[pid] = f"{pid} is {t.get('state')}"
elif now != had:
changed[pid] = f"the terms of {pid} changed since the line was booked"
return problems, changed
def _stale(s: sh.Sheet, actor: str) -> dict:
"""line_no -> why a line was built on facts that have since changed (TS-16). Derived each time from time-off, hr's pattern held on the sheet, and
projects; nothing is stored. Only a week still waiting for its approval has stale lines: approved revisions are never flagged, only corrected."""
if s.status != "submitted":
return {}
absences = _absence_days(seams.absences(s.company_code, s.week_start, s.week_start + timedelta(days=6), s.person_id, actor), s)
problems, changed = _project_problems(s, actor)
why = sh.stale_lines(s, absences, changed)
for n, p in problems.items():
why.setdefault(n, p)
return why
# ── reading ─────────────────────────────────────────────────────────────────────────────────
@TS.use_case(kind="query", route="GET /sheets")
def list_timesheets(caller: dict, scope: str = "mine") -> dict:
"""The weeks the caller may see (the latest revision of each). scope=mine: their own weeks. scope=waiting: the weeks waiting for THEM to approve
(a manager's team; a project lead's weeks with lines on their projects). scope=all: every week (finance and the owner). Summaries only;
get_timesheet gives the lines and hours."""
if scope not in ("mine", "waiting", "all"):
raise _bounds(f"scope is mine, waiting or all, not {scope!r}")
with repository() as repo, _decide():
me = _me(caller)
if scope == "all":
_need(caller, READ_ALL)
rows = repo.sheets.summaries(everything=True)
elif scope == "waiting":
if not (auth.has_permission(caller, APPROVE) or auth.has_permission(caller, LINE_APPROVE)):
raise Refused(403, f"missing permission: {APPROVE} or {LINE_APPROVE}")
rows = repo.sheets.summaries(waiting_for=me) if auth.has_permission(caller, APPROVE) else []
if auth.has_permission(caller, LINE_APPROVE):
have = {r["sheet_id"] for r in rows}
rows += [r for r in repo.sheets.summaries(lead=_person_of(caller)) if r["sheet_id"] not in have]
else:
_need(caller, READ_OWN)
rows = repo.sheets.summaries(mine=me)
return {"timesheets": [_summary(r) for r in rows]}
@TS.use_case(kind="query", route="GET /sheets/{sheet_id}")
def get_timesheet(caller: dict, sheet_id: Ref, revision: Optional[int] = None) -> dict:
"""One week, whole: its lines and entries, hours per day against the contracted hours, the approver, the journal and deliveries it led to, and
(while it is open) the gates still unmet, or (while it waits for approval) the lines that went stale. The latest revision unless one is named.
A project lead sees only the lines on projects they lead. A week the caller may not see is a 404."""
with repository() as repo, _decide():
s, lead = _load(repo, caller, sheet_id, revision)
if lead is not None:
return _view(repo, s, lead=lead)
if s.status == "open" and _me(caller).same_as(s.person):
return _view(repo, s, readiness=_local_readiness(repo, s))
return _view(repo, s, stale=_stale(s, _who(caller)) if s.status == "submitted" else None)
@TS.use_case(kind="query", route="GET /activities")
def list_activities(caller: dict) -> dict:
"""The activities a person books hours to (development, meetings, administration ...) and whether each is billable by default."""
_need(caller, READ_OWN)
with repository() as repo, _decide():
person = seams.person_by_sub(_me(caller).sub, _who(caller)) if _me(caller).sub else None
rows = repo.activities.listing((person or {}).get("company_code") or "*")
return {"activities": [{"company_code": r["company_code"], "activity_code": r["activity_code"], "name": r["name"],
"billable_default": bool(r["billable_default"]), "internal_only": bool(r["internal_only"]),
"valid_from": r["valid_from"].isoformat()} for r in rows]}
# ── recording: open, add lines, book hours ───────────────────────────────────────────────────
@TS.use_case(kind="command", route="POST /sheets", permission=WRITE_OWN, status=201, idempotent=True)
def open_timesheet(caller: dict, week: Week) -> dict:
"""Open the caller's week (none -> open): sheet_id is <person_id>:<iso week>, e.g. 4410:2026-W47, revision 1. hr gives the person, the company,
the home cost centre and each day's contracted hours (a bank holiday expects 0); time-off's approved absence is already in the week as read-only
lines. Opening a week that is already open answers the same sheet. A week before the person started or after their last day does not open."""
monday = _week(week)
with repository() as repo, _decide():
me = _me(caller)
actor = _who(caller)
person = seams.person_by_sub(me.sub, actor) if me.sub else None
if person is None:
raise Refused(409, "hr holds no person with your sign-in, so you cannot record time", gate="person",
fix="ask HR to link your login to your person record")
pid = int(person["person_id"])
sheet_id = f"{pid}:{sh.week_id(monday)}"
have = repo.sheets.get(sheet_id)
if have is not None:
if not me.same_as(have.person):
raise Refused(404, f"no timesheet {sheet_id}")
return _view(repo, have, readiness=_local_readiness(repo, have) if have.status == "open" else None)
sunday = monday + timedelta(days=6)
pattern = {d["date"]: d for d in seams.work_pattern(pid, monday, sunday, actor)}
employed = sorted(k for k, d in pattern.items() if d.get("employed"))
if not employed:
raise Refused(409, f"You are not employed in week {sh.week_id(monday)} (hr has no engagement of yours on any of its days)", gate="employed",
fix="a week before you started or after your last day does not open")
emp = seams.employment(pid, date.fromisoformat(employed[0]), actor)
company = emp.get("company_code") or person.get("company_code")
if not company:
raise Refused(409, "hr gives no company for you in this week", gate="employed")
expected = {d.isoformat(): int((pattern.get(d.isoformat()) or {}).get("contracted_minutes") or 0) if (pattern.get(d.isoformat()) or {}).get("employed") else 0
for d in (monday + timedelta(days=n) for n in range(7))}
s = sh.Sheet(sheet_id, pid, company, monday, me, expected=expected, home_cost_centre=emp.get("cost_centre") or "")
s.lines = _absence_lines(s, seams.absences(company, monday, sunday, pid, actor), 1)
s.last_line_no = len(s.lines)
repo.claims.once("open", f"{sheet_id}:r1", _now(), actor)
repo.sheets.create(s, _now())
return _view(repo, s, readiness=_local_readiness(repo, s))
def _sync_absences(repo, s: sh.Sheet, actor: str) -> bool:
"""Make the week's absence lines what time-off holds now (a cancelled absence leaves an open week, a new one arrives). True when anything changed."""
want = seams.absences(s.company_code, s.week_start, s.week_start + timedelta(days=6), s.person_id, actor)
new = _absence_lines(s, want, s.next_line_no())
have = [ln for ln in s.lines if ln.kind == "absence"]
sig = lambda ls: sorted((ln.absence_ref, ln.activity_code, tuple((e.work_date, e.hours) for e in ln.entries)) for ln in ls) # noqa: E731
if sig(have) == sig(new):
return False
repo.sheets.remove_lines(s.sheet_id, s.revision, [ln.line_no for ln in have])
repo.sheets.add_lines(s.sheet_id, s.revision, new, _now())
s.lines = [ln for ln in s.lines if ln.kind != "absence"] + new
s.last_line_no = max(s.last_line_no, *(ln.line_no for ln in new)) if new else s.last_line_no
return True
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/refresh", permission=WRITE_OWN)
def refresh_timesheet(caller: dict, sheet_id: Ref) -> dict:
"""Re-read time-off for an OPEN week of the caller's: a newly approved absence appears as a read-only line, a cancelled one leaves. The week
is refreshed again by submit_timesheet."""
with repository() as repo, _decide():
s = _own(repo, caller, sheet_id)
s.ensure_open()
if _sync_absences(repo, s, _who(caller)):
repo.sheets.save(s, _now())
return _view(repo, s, readiness=_local_readiness(repo, s))
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/lines", permission=WRITE_OWN, status=201)
def add_time_line(caller: dict, sheet_id: Ref, activity_code: Code, project_id: Optional[Ref] = None, item_id: Optional[Ref] = None,
cost_centre: Optional[Code] = None, billable: Optional[bool] = None, description: Optional[Descr] = None) -> dict:
"""Add a line to an OPEN week: a project (project_id, optionally an item) or an internal cost centre (cost_centre), with the activity worked.
A project line copies the project's time terms now (projects): whether it is billable (only on time_and_materials, and only if you do not
untick it), how it bills, its cost centre and lead. billable defaults from the project's terms and the activity; you may untick it, never tick
it on a project that is not billable. Hours are booked with set_time_entries."""
if bool(project_id) == bool(cost_centre):
raise _bounds("a line is a project (project_id) or an internal cost centre (cost_centre), one of the two")
with repository() as repo, _decide():
s = _own(repo, caller, sheet_id)
s.ensure_open()
if len([ln for ln in s.lines if ln.kind != "absence"]) >= sh.LINES_MAX:
raise Refused(409, f"A week holds at most {sh.LINES_MAX} lines")
actor = _who(caller)
act = repo.activities.in_force(s.company_code, {activity_code}, s.week_start).get(activity_code)
if act is None:
raise Refused(409, f"No activity {activity_code} is in force for {s.company_code} in this week", gate="activity", fix="pick one of list_activities")
line = sh.Line(s.next_line_no(), "internal" if cost_centre else "project", activity_code=activity_code, description=(description or "").strip())
if cost_centre:
if billable:
raise Refused(409, "Time on an internal cost centre is never billable", gate="billable")
line.cost_centre = cost_centre
else:
if act["internal_only"]:
raise Refused(409, f"Activity {activity_code} is for internal cost centres only", gate="activity")
t = seams.time_terms(project_id, s.week_start, actor)
if t is None:
raise Refused(404, f"projects knows no project {project_id}")
if t.get("state") not in (None, "active"):
raise Refused(409, f"{project_id} is {t.get('state')}: time is booked to active projects", gate="bookable", fix="book to another project")
billing = t.get("billing") or "non_billable"
if billable and billing != "time_and_materials":
raise Refused(409, f"{project_id} is {billing.replace('_', ' ')}: time on it is never billable", gate="billable",
fix="leave billable off")
line.project_id, line.item_id, line.billing = project_id, item_id or "", billing
line.billable = (bool(act["billable_default"]) if billable is None else bool(billable)) and billing == "time_and_materials"
line.bill_via = (t.get("bill_via") or "none") if line.billable else "none"
line.sales_order_id = t.get("sales_order_id") or ""
line.cost_centre, line.lead_person_id = t.get("cost_centre") or "", t.get("lead_person_id")
repo.sheets.add_lines(s.sheet_id, s.revision, [line], _now())
s.last_line_no = line.line_no
repo.sheets.save(s, _now())
s.lines.append(line)
return _view(repo, s, readiness=_local_readiness(repo, s))
@TS.use_case(kind="command", route="DELETE /sheets/{sheet_id}/lines/{line_no}", permission=WRITE_OWN, destructive=True)
def remove_time_line(caller: dict, sheet_id: Ref, line_no: BigId) -> dict:
"""Remove a line (and its hours) from an OPEN week. Absence lines come from time-off and are not removed here."""
with repository() as repo, _decide():
s = _own(repo, caller, sheet_id)
s.ensure_open()
ln = s.line(int(line_no))
if ln is None:
raise Refused(404, f"timesheet {sheet_id} has no line {line_no}")
if ln.kind == "absence":
raise Refused(409, "An absence line comes from time-off: cancel the absence there", gate="absence_line")
repo.sheets.remove_lines(s.sheet_id, s.revision, [ln.line_no])
s.lines = [x for x in s.lines if x.line_no != ln.line_no]
return _view(repo, s, readiness=_local_readiness(repo, s))
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/lines/{line_no}/entries", permission=WRITE_OWN)
def set_time_entries(caller: dict, sheet_id: Ref, line_no: BigId, entries: Entries) -> dict:
"""Book hours on a line of an OPEN week: for each day (work_date inside the week) the hours as an exact string in steps of the time increment
(a quarter hour), at most 24.00 a day over all lines; "0" clears the day. A negative number, a day above 24.00 or a step the increment does
not allow is refused 422 {gate: bounds}. A note per day is optional (a day above max_day_hours needs one to submit)."""
with repository() as repo, _decide():
s = _own(repo, caller, sheet_id)
s.ensure_open()
ln = s.line(int(line_no))
if ln is None:
raise Refused(404, f"timesheet {sheet_id} has no line {line_no}")
if ln.kind == "absence":
raise Refused(409, "An absence line comes from time-off: it is read-only", gate="absence_line")
inc = _inc(repo, s.week_start)
put, clear, seen = [], [], set()
for e in entries:
when = _date(e.work_date, "work_date")
if not s.week_start <= when <= s.week_start + timedelta(days=6):
raise _bounds(f"{when} is outside week {sh.week_id(s.week_start)}")
if when in seen:
raise _bounds(f"{when} is given twice")
seen.add(when)
try:
hours = sh.parse_hours(e.hours, inc)
except ValueError as why:
raise _bounds(str(why)) from None
if hours == 0:
clear.append(when)
else:
put.append(sh.Entry(when, hours, (e.note or "").strip()))
keep = [x for x in ln.entries if x.work_date not in seen]
ln.entries = sorted(keep + put, key=lambda x: x.work_date)
try:
sh.check_day_totals(s, s.booked())
except ValueError as why:
raise _bounds(str(why)) from None
repo.sheets.set_entries(s.sheet_id, s.revision, ln.line_no, put, clear, _now())
return _view(repo, s, readiness=_local_readiness(repo, s))
# ── submitting: the gates, the rates and terms copied, the approver from hr, fixed ──────────────────
def _price(repo, s: sh.Sheet, actor: str, cache: dict) -> tuple:
"""Copy the cost rate onto every entry of a line that costs (the person's rate group on the work date, that group's rate on that date) and the
project's terms and time product onto every project line. Returns (rate problems, product problems, currency)."""
groups = repo.rates.person_groups(s.person_id)
costing = [ln for ln in s.lines if ln.costs() and ln.entries]
rates = repo.rates.rates(s.company_code, {g["rate_group"] for g in groups})
rate_problems, currencies = [], set()
for ln in costing:
posts = not (ln.kind == "internal" and ln.cost_centre == s.home_cost_centre)
for e in ln.entries:
g = tables.rate_group_on(groups, e.work_date)
r = tables.rate_on(rates, s.company_code, g, e.work_date) if g else None
if g is None or r is None:
e.unit_cost = None
if posts:
rate_problems.append(f"no cost rate for {'a rate group' if g is None else g} on {e.work_date} (line {ln.line_no})")
continue
e.unit_cost = hours4(r["rate_amount"])
if posts:
currencies.add(r["currency"])
first = tables.rate_group_on(groups, _first_day(ln, s.week_start))
ln.rate_group = first or ""
costs = {e.unit_cost for e in ln.entries}
ln.unit_cost = next(iter(costs)) if len(costs) == 1 else None
product_problems = []
for pid in sorted({ln.project_id for ln in costing if ln.kind == "project"}):
mine = [ln for ln in costing if ln.project_id == pid]
t = _terms(cache, pid, min(_first_day(ln, s.week_start) for ln in mine), actor)
if t is None:
product_problems.append(f"{pid} is unknown to projects")
continue
for ln in mine:
ln.billing = t.get("billing") or "non_billable"
ln.cost_centre, ln.lead_person_id, ln.sales_order_id = t.get("cost_centre") or "", t.get("lead_person_id"), t.get("sales_order_id") or ""
if ln.billing != "time_and_materials":
ln.billable, ln.bill_via = False, "none"
continue
ln.bill_via = (t.get("bill_via") or "none") if ln.billable else "none"
prod = next((p for p in t.get("products") or [] if p.get("rate_group") == ln.rate_group), None)
ln.product_id = (prod or {}).get("product_id") or ""
ln.so_line_no = (prod or {}).get("so_line_no")
if ln.billable and ln.bill_via == "sales_order" and ln.rate_group and (prod is None or ln.so_line_no is None):
product_problems.append(f"{pid} has no time product on its order for rate group {ln.rate_group or '(none)'} (line {ln.line_no})")
return rate_problems, product_problems, currencies
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/submit", permission=SUBMIT)
def submit_timesheet(caller: dict, sheet_id: Ref) -> dict:
"""Submit the caller's OPEN week (open -> submitted). Refused with every unmet gate named: each working day's contracted hours covered by time or
absence (missing_hours_mode block), a day above max_day_hours has a note, every project line bookable on its day (projects), every internal cost
centre open (gl), a cost rate for the person's rate group, no stale line. Absence is re-read from time-off; the project's terms and the cost rate
are copied onto the lines. The approver is the person's manager from hr on the week's Monday (a manager's own week goes one up) and the route in
force that Monday - both FIXED here - and the lines are frozen."""
with repository() as repo, _decide():
s = _own(repo, caller, sheet_id)
s.move("submit_timesheet")
actor = _who(caller)
monday = s.week_start
_sync_absences(repo, s, actor)
cache: dict = {}
rate_problems, product_problems, currencies = _price(repo, s, actor, cache)
if len(currencies) > 1:
rate_problems.append("the rates of this week are in more than one currency: " + ", ".join(sorted(currencies)))
problems, changed = _project_problems(s, actor, cache)
closed = []
for cc in sorted({ln.cost_centre for ln in s.lines if ln.kind == "internal"}):
if seams.cost_centre_open(cc, actor) is False:
closed.append(cc)
absences = _absence_days(seams.absences(s.company_code, monday, monday + timedelta(days=6), s.person_id, actor), s)
stale = sh.stale_lines(s, absences, {})
max_day = Decimal(repo.settings.on("max_day_hours", monday))
gates = sh.readiness(gaps=sh.coverage_gaps(s), mode=repo.settings.on("missing_hours_mode", monday), notes=sh.long_days(s, max_day),
bookable=[*problems.values(), *changed.values(), *product_problems], closed=closed, rates=rate_problems, stale=stale)
unmet = [g for g in gates if not g["met"]]
if unmet:
raise Refused(409, "Cannot submit: " + "; ".join(f"{g['gate']}: {g['reason']}" for g in unmet), gate=unmet[0]["gate"], fix=unmet[0]["fix"],
fields={"gates": gates})
s.approval_route = repo.settings.on("approval_route", monday)
s.approver_person_id, s.approver, s.rerouted_from = sh.derive_approver(s.person, lambda n: seams.manager(s.person_id, monday, n, actor))
s.currency = next(iter(currencies), "")
repo.claims.once("submit", f"{s.sheet_id}:r{s.revision}", _now(), actor)
s.submitted_at = _now()
repo.sheets.rewrite_lines(s.sheet_id, s.revision, [ln for ln in s.lines if ln.kind != "absence"], _now())
repo.sheets.save(s, _now())
return _view(repo, s, readiness=gates)
# ── deciding ────────────────────────────────────────────────────────────────────────────────
def _is_approver(caller: dict, s: sh.Sheet) -> bool:
return s.approver is not None and _me(caller).same_as(s.approver)
def _decider(repo, caller: dict, sheet_id: str, what: str) -> sh.Sheet:
"""The sheet and the proof the caller may decide it. Order: the row is theirs to see (404), the permission (403), that they ARE its approver (403),
then the rules (nobody decides their own week) as 409."""
s, lead = _load(repo, caller, sheet_id)
if lead is not None:
raise Refused(403, "You are not the approver of this week", gate="approver", fix="it waits for the person's manager")
_need(caller, APPROVE)
if _me(caller).same_as(s.person):
raise Refused(409, "This is your own week: it goes to your manager", gate="second_person", fix="it goes to your manager")
if s.status == "submitted" and not _is_approver(caller, s):
raise Refused(403, "You are not the approver of this week", gate="approver", fix="it waits for the person's manager")
sh.check_approver(s, _me(caller), what)
return s
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/approve-project-lines", permission=LINE_APPROVE)
def approve_project_lines(caller: dict, sheet_id: Ref) -> dict:
"""A project lead approves the lines of a SUBMITTED week that are booked to projects they lead (route project_lead_then_manager; the lead's own
lines go to their manager instead). Each line's approval is recorded once; the week goes to the manager when every project line is approved.
The lead sees only those lines. Never the lead's own week (email AND sub)."""
with repository() as repo, _decide():
s = repo.sheets.get(sheet_id)
if s is None or s.status == "open":
raise Refused(404, f"no timesheet {sheet_id}")
me = _me(caller)
pid = _person_of(caller)
mine = [ln for ln in s.lines if ln.kind == "project" and ln.lead_person_id == pid]
if not mine:
raise Refused(404, f"no timesheet {sheet_id}")
if me.same_as(s.person) or pid == s.person_id:
raise Refused(409, "This is your own week: your lines go to your manager", gate="second_person", fix="it goes to your manager")
if s.status != "submitted":
raise Refused(409, f"Timesheet {sheet_id} revision {s.revision} is {s.status}: only a submitted week's lines are approved")
if s.approval_route != "project_lead_then_manager":
raise Refused(409, "This week routes to the manager alone: project leads do not approve its lines", gate="route")
waiting = [ln for ln in s.lines_waiting() if ln.lead_person_id == pid]
if not waiting:
raise Refused(409, "Your lines of this week are already approved")
rows = [sh.LineApproval(ln.line_no, pid, me.email, me.sub, _now()) for ln in waiting]
repo.sheets.add_approvals(s.sheet_id, s.revision, rows, _now())
s.approvals += rows
return _view(repo, s, lead=pid)
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/return", permission=None)
def return_timesheet(caller: dict, sheet_id: Ref, reason: Reason) -> dict:
"""Send a SUBMITTED week back with a question (a reason is required). Revision n ends as returned (its question kept); the week continues as
an OPEN revision n+1 with the lines copied, and resubmitting starts again - the approvals of revision n never count for n+1. Decided by the
week's approver (ts:sheet:approve) or by a project lead of a line on it (ts:line:approve_project); never the person's own week."""
if not reason.strip():
raise _bounds("a reason is required to return a week")
with repository() as repo, _decide():
s, lead = _load(repo, caller, sheet_id)
me = _me(caller)
if me.same_as(s.person):
raise Refused(409, "This is your own week: it goes to your manager", gate="second_person", fix="it goes to your manager")
if lead is not None:
_need(caller, LINE_APPROVE)
else:
_need(caller, APPROVE)
if s.status == "submitted" and not _is_approver(caller, s):
raise Refused(403, "You are not the approver of this week", gate="approver", fix="it waits for the person's manager")
s.move("return_timesheet")
repo.claims.once("close", f"{s.sheet_id}:r{s.revision}", _now(), _who(caller))
_next_revision(repo, s, "returned", me, reason.strip())
return _view(repo, s, lead=None)
def _next_revision(repo, s: sh.Sheet, ended: str, me: sh.Identity, reason: str) -> None:
"""End revision n (returned | superseded) and open n+1 with the lines and entries copied; the approvals of n stay with n."""
s.status = ended
s.ended_by, s.ended_sub, s.ended_at, s.reason = me.email, me.sub, _now(), reason
repo.sheets.save(s, _now())
old = s.revision
s.lines = sh.copy_lines(s.lines)
s.revision, s.status, s.approvals = old + 1, "open", []
s.approver_person_id = s.approver = s.rerouted_from = s.submitted_at = None
s.approval_route = ""
s.approved_by = s.approved_sub = ""
s.approved_at = s.approved_on = s.journal_id = s.ended_at = None
s.ended_by = s.ended_sub = ""
repo.sheets.create(s, _now())
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/approve", permission=None)
def approve_timesheet(caller: dict, sheet_id: Ref) -> dict:
"""Approve a SUBMITTED week as a whole (submitted -> approved): for its approver (ts:sheet:approve), never the person's own week (matched on email
OR sub). Under the route project_lead_then_manager every project line must be approved by its lead first; a line built on facts that changed
(an absence now overlapping it, a project closed) is refused {gate: not_stale} until the person corrects it by a return. One transaction
records the approval, the labour journal (project cost centres debited, the person's home cost centre credited, hours x the standard rate) and the
deliveries to sales orders as PENDING rows, by the difference from the last approved revision; they are sent after the commit."""
with repository() as repo, _decide():
s = _decider(repo, caller, sheet_id, "approve")
if s.approval_route == "project_lead_then_manager" and s.lines_waiting():
waiting = s.lines_waiting()
raise Refused(409, "Lines wait for their project lead: " + "; ".join(f"{ln.project_id} waits for person {ln.lead_person_id}" for ln in waiting),
gate="project_lines_approved", fix="the lead approves them, or returns the week")
actor = _who(caller)
stale = _stale(s, actor)
if stale:
raise Refused(409, "Lines changed under the week: " + "; ".join(f"line {n}: {why}" for n, why in sorted(stale.items())), gate="not_stale",
fix="return the week for a correction", fields={"stale": {str(n): w for n, w in stale.items()}})
me = _me(caller)
now, today = _now(), _today(repo)
repo.claims.once("approval", f"{s.sheet_id}:r{s.revision}", now, actor)
s.move("approve_timesheet")
s.approved_by, s.approved_sub, s.approved_at, s.approved_on = me.email, me.sub, now, today
ref = f"ts:{s.sheet_id}:r{s.revision}"
rows = []
prev = repo.sheets.prior_approved(s.sheet_id, s.revision)
if prev is not None: # a correction: the journal of the revision it replaces is reversed first
old = repo.outbox.get("gl", f"ts:{s.sheet_id}:r{prev}:approve")
if old is not None:
was = json.loads(old["payload_json"])
rows.append(("gl", f"ts:{s.sheet_id}:r{prev}:reverse", s.sheet_id,
{"journal_id": seams.journal_id(f"ts:{s.sheet_id}:r{prev}", "reverse"), "source_module": "timesheets",
"source_document": f"ts:{s.sheet_id}:r{prev}", "event": "reverse", "company_code": s.company_code, "journal_date": today.isoformat(),
"currency": was["currency"], "memo": f"Timesheet {s.sheet_id} revision {prev} reversed"[:200],
"lines": sh.reversal_of(was["lines"])}))
lines = sh.journal_lines(s)
if lines:
body = {"journal_id": seams.journal_id(ref, "approve"), "source_module": "timesheets", "source_document": ref, "event": "approve",
"company_code": s.company_code, "journal_date": today.isoformat(), "currency": s.currency, "memo": f"Timesheet {s.sheet_id}"[:200],
"lines": lines}
s.journal_id = int(body["journal_id"])
rows.append(("gl", f"{ref}:approve", s.sheet_id, body))
delivered: dict = {}
last_ref: dict = {}
for c in repo.billing.of_sheet(s.sheet_id):
key = (c["line_no"], c["sales_order_id"], c["so_line_no"])
delivered[key] = delivered.get(key, Decimal(0)) + Decimal(c["quantity"])
last_ref[key] = c["delivery_ref"]
claims = []
for line_no, order, so_line, delta in sh.delivery_deltas(sh.delivery_wanted(s), delivered):
dref = f"{ref}:{line_no}"
body = {"line_no": so_line, "quantity": f"{delta:.4f}", "source": "timesheets", "delivery_ref": dref, "delivered_on": today.isoformat()}
if delta < 0 and (line_no, order, so_line) in last_ref:
body["corrects_ref"] = last_ref[(line_no, order, so_line)]
claims.append({"sheet_id": s.sheet_id, "revision": s.revision, "line_no": line_no, "purpose": "delivery", "delivery_ref": dref,
"sales_order_id": order, "so_line_no": so_line, "quantity": delta})
rows.append(("sales-order", dref, s.sheet_id, {"sales_order_id": order, "body": body}))
repo.billing.add_many(claims, now)
repo.outbox.add_many(rows, now)
repo.sheets.save(s, now)
_drain(sheet_id, actor=_who(caller))
with repository() as repo, _decide():
full = repo.sheets.get(sheet_id, s.revision)
return _view(repo, full)
@TS.use_case(kind="command", route="POST /sheets/{sheet_id}/reopen", permission=REOPEN)
def reopen_timesheet(caller: dict, sheet_id: Ref, reason: Reason) -> dict:
"""Finance reopens an APPROVED week for a correction (approved -> superseded; a reason is required). Revision n is never edited: revision n+1
opens with the lines copied, is submitted and approved on its own, and ONLY THEN do the downstream effects move - the journal of n is reversed
and the new one posted, the orders are told the difference. Never the person's own week, nor one the caller approved (email AND sub)."""
if not reason.strip():
raise _bounds("a reason is required to reopen a week")
with repository() as repo, _decide():
s, lead = _load(repo, caller, sheet_id)
me = _me(caller)
sh.check_reopener(s, me)
s.move("reopen_timesheet")
repo.claims.once("reopen", f"{s.sheet_id}:r{s.revision}", _now(), _who(caller))
_next_revision(repo, s, "superseded", me, reason.strip())
return _view(repo, s)
# ── sending: side effects only after a commit, from pending rows ────────────────────────────────────
def _deliver(row: dict, who: str) -> dict:
"""Send one pending row (after its commit): the journal to gl, or the delivery to the order. Both replay the same derived key, so a repeat is
their replay. A refusal (a closed period, hours beyond the ordered) stays on the pending row with the neighbour's gate and reason, never silently
dropped; no answer is retried on the next pass."""
if row["target"] == "notify":
return _deliver_notify(row, who)
body = json.loads(row["payload_json"])
if row["target"] == "gl":
out = seams.post_journal(body, who)
elif row["target"] == "ar":
out = seams.post_invoice(body, who)
else:
out = seams.post_delivery(body["sales_order_id"], body["body"], 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"}
if out["ok"]:
repo.outbox.mark(fresh, "sent", "", _now())
if row["target"] == "ar": # the run learns its invoice
run = repo.runs.get(row["sheet_id"])
if run is not None and not run["invoice_id"]:
repo.runs.save({**run, "invoice_id": str((out["data"] or {}).get("invoice_id") or "") or None}, _now())
return {"ref": row["ref"], "status": "sent"}
why = f"{out['gate'] + ': ' if out['gate'] else ''}{out['reason']}"
repo.outbox.mark(fresh, "pending", why, _now())
return {"ref": row["ref"], "status": "refused" if not out["retry"] else "retry", "reason": why}
def _deliver_notify(row: dict, who: str) -> dict:
"""Hand one reminder to the mail capability, after the commit that wrote it. With no mail provider it is a task for a person (status manual); a provider
that is down leaves it pending for the next pass."""
body = json.loads(row["payload_json"])
out = seams.send_mail(row["ref"], body["from_email"], body["to"], body["subject"], body["body_text"], 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"}
if out["ok"]:
repo.outbox.mark(fresh, "manual" if out["manual"] else "sent", out["reason"], _now())
return {"ref": row["ref"], "status": "manual" if out["manual"] else "sent"}
repo.outbox.mark(fresh, "pending", out["reason"], _now())
return {"ref": row["ref"], "status": "retry", "reason": out["reason"]}
def _drain(sheet_id: Optional[str] = None, targets: Optional[tuple] = None, actor: str = "timesheets") -> list:
"""Send what is pending, oldest first: after a commit for one week, or for everything (the worker's tick). One row failing never stops the rest;
a row that stays pending is not tried twice in one pass."""
with repository() as repo:
rows = repo.outbox.pending(sheet_id, targets)
done = []
for row in rows:
try:
done.append(_deliver(row, actor))
except Refused as lost:
done.append({"ref": row["ref"], "status": "lost a race", "reason": lost.reason})
return done
@TS.use_case(kind="command", route="POST /outbox/drain", permission=POST_AUTO, idempotent=True, tool=False)
def drain_ts_outbox(caller: dict) -> dict:
"""The worker's tick (no server-side scheduler): send every pending labour journal to gl. Safe to run at any time and from two workers: gl
replays the derived journal id. A journal gl refused (the period was closed) stays pending with gl's reason and goes out once the period opens."""
return {"sent": _drain(None, ("gl",), _who(caller))}
@TS.use_case(kind="command", route="POST /deliveries/report", permission=DELIVER, idempotent=True, tool=False)
def report_deliveries(caller: dict) -> dict:
"""The worker's tick for deliveries: send every pending delivery to its sales order. The order replays the same delivery_ref; hours beyond the
ordered stay pending with the order's reason, for the project lead to see."""
return {"sent": _drain(None, ("sales-order",), _who(caller))}
# ── settings the customer edits: cost rates, rate groups, activities, controls ───────────────────────
def _later(repo, when: date, today: date, first: bool, what: str) -> None:
"""Rows are dated and never edited: a change starts AFTER today. The first row of a key may start any day (it is the history)."""
if not first and when <= today:
raise Refused(422, f"{what} starts after today ({today}): the past is never edited", gate="bounds", fix="date it from tomorrow or later")
@TS.use_case(kind="command", route="POST /cost-rates", permission=COST_RATE, status=201)
def set_cost_rate(caller: dict, company_code: Code, rate_group: Code, rate_amount: Rate, currency: Currency, valid_from: Day) -> dict:
"""Finance sets the standard cost of an hour for a rate group from a date (a dated row; NUMERIC(19,4); never read from a salary). A line costs at
the rate in force on its WORK date. The first row of a group may start any day; a change must start after today - approved and posted lines
never change."""
when = _date(valid_from, "valid_from")
cur = _cur(currency)
with repository() as repo, _decide():
_later(repo, when, _today(repo), not repo.rates.has_rate_rows(company_code, rate_group.upper()), "a new cost rate")
repo.rates.put_rate(company_code, rate_group.upper(), rate_amount, cur, when, _who(caller), _now())
return {"company_code": company_code, "rate_group": rate_group.upper(), "rate_amount": f"{rate_amount:.4f}", "currency": cur, "valid_from": when.isoformat()}
@TS.use_case(kind="command", route="POST /person-rate-groups", permission=COST_RATE, status=201)
def set_person_rate_group(caller: dict, person_id: BigId, rate_group: Code, valid_from: Day) -> dict:
"""Finance sets which rate group a person is costed at from a date (a dated row). The first row of a person may start any day; a change must start
after today."""
when = _date(valid_from, "valid_from")
if int(person_id) < 1:
raise _bounds("a person id is a positive number")
with repository() as repo, _decide():
_later(repo, when, _today(repo), not repo.rates.has_person_rows(int(person_id)), "a new rate group")
repo.rates.put_person(int(person_id), rate_group.upper(), when, _who(caller), _now())
return {"person_id": str(person_id), "rate_group": rate_group.upper(), "valid_from": when.isoformat()}
@TS.use_case(kind="query", route="GET /cost-rates", permission=COST_RATE)
def list_cost_rates(caller: dict, company_code: Code) -> dict:
"""The dated standard cost rates in force for a company (its own rows and the '*' defaults). Visible to finance only."""
with repository() as repo:
return {"cost_rates": [{"company_code": r["company_code"], "rate_group": r["rate_group"], "rate_amount": f"{Decimal(r['rate_amount']):.4f}",
"currency": r["currency"], "valid_from": r["valid_from"].isoformat()} for r in repo.rates.listing(company_code)]}
@TS.use_case(kind="command", route="POST /activities", permission=ACTIVITY, status=201)
def set_activity(caller: dict, company_code: Code, activity_code: Code, name: Descr, billable_default: bool, valid_from: Day,
internal_only: bool = False) -> dict:
"""Finance defines an activity people book hours to (development, travel, admin) and whether it is billable by default, from a date. The first row
of a code may start any day; a change must start after today."""
when = _date(valid_from, "valid_from")
with repository() as repo, _decide():
first = activity_code.upper() not in {r["activity_code"] for r in repo.activities.listing(company_code) if r["company_code"] == company_code}
_later(repo, when, _today(repo), first, "a new activity")
repo.activities.put(company_code, activity_code.upper(), name, billable_default, internal_only, when, _now())
return {"company_code": company_code, "activity_code": activity_code.upper(), "valid_from": when.isoformat()}
CONTROL_KEYS = tuple(k for k, v in SETTING_DEFS.items() if v[2])
@TS.use_case(kind="command", route="POST /controls", permission=CONTROLS, status=201)
def set_ts_control(caller: dict, key: Code, value: Value, valid_from: Day) -> dict:
"""The owner sets a control from a date (a dated row): approval_route (manager | project_lead_then_manager), missing_hours_mode (block | warn) or
business_timezone. Held by the owner alone: finance cannot turn the missing-hours check off for the team it manages. The first row of a key may
start today; a change must start after today; sheets of weeks starting before it keep the earlier value."""
when = _date(valid_from, "valid_from")
if key not in CONTROL_KEYS:
raise _bounds(f"{key!r} is not a control: {', '.join(CONTROL_KEYS)}")
allowed = {"approval_route": ROUTES, "missing_hours_mode": MODES}.get(key)
if allowed and value not in allowed:
raise _bounds(f"{key} is one of {', '.join(allowed)}")
if key == "business_timezone":
try:
ZoneInfo(value)
except Exception:
raise _bounds(f"{value!r} is not a time zone like Europe/London") from None
with repository() as repo, _decide():
today = _today(repo)
if when < today:
raise _bounds(f"valid_from {when} is before today ({today})")
_later(repo, when, today, not repo.settings.has(key), f"a new {key}")
repo.settings.put(key, value, when, _who(caller), _now())
return {"key": key, "value": value, "valid_from": when.isoformat()}
# ── payroll and time-off read what was approved ───────────────────────────────────────────────────
@TS.use_case(kind="query", route="GET /approved-hours", permission=READ_APPROVED, tool=False, aliases={"date_from": "from", "date_to": "to"})
def get_approved_hours(caller: dict, company_code: Code, date_from: Day, date_to: Day, person_id: Optional[BigId] = None,
changed_since: Optional[Instant] = None) -> dict:
"""The hours of approved weeks for payroll and time-off: one row per person, day and approved revision {person_id, work_date, sheet_id, revision,
status approved | superseded, normal_hours, overtime_hours, absence_hours, changed_at}. Overtime follows the overtime_rule in force (default: hours
above the week's contracted hours). Only the company asked for; at most 93 days. With changed_since the superseded revisions appear too, so a
reader settles the difference. No rates, no cost. Held by payroll's and time-off's service clients alone."""
start, end = _date(date_from, "from"), _date(date_to, "to")
if end < start or (end - start).days > 93:
raise _bounds("from and to are at most 93 days apart")
since = None
if changed_since:
try:
since = datetime.fromisoformat(changed_since.strip().replace("Z", "+00:00"))
except ValueError:
raise _bounds("changed_since is an RFC 3339 instant like 2026-11-23T10:00:00Z") from None
since = since.astimezone(timezone.utc).replace(tzinfo=None) if since.tzinfo else since
with repository() as repo:
roots = repo.sheets.approved_roots(company_code, start, end, None if person_id is None else int(person_id))
entries = repo.sheets.entries_of([r["sheet_id"] for r in roots])
rule = repo.settings.on("overtime_rule", end)
if rule not in OVERTIME_RULES:
rule = "weekly_above_contracted"
by_sheet: dict = {}
for r in roots:
by_sheet.setdefault(r["sheet_id"], []).append(r)
rows = []
for sid, revs in by_sheet.items():
current = revs[-1]
for i, r in enumerate(revs):
superseded = r is not current
changed_at = revs[i + 1]["approved_at"] if superseded else r["approved_at"]
if superseded and since is None:
continue
if since is not None and changed_at <= since:
continue
worked: dict = {}
away: dict = {}
for e in entries:
if e["sheet_id"] == sid and e["revision"] == r["revision"]:
(away if e["kind"] == "absence" else worked)[e["work_date"]] = (away if e["kind"] == "absence" else worked).get(
e["work_date"], Decimal(0)) + Decimal(e["hours"])
split = sh.split_hours(worked, away, json.loads(r["expected_json"] or "{}"), rule)
for d in sorted(set(worked) | set(away)):
if start <= d <= end:
normal, over = split.get(d, (Decimal(0), Decimal(0)))
rows.append({"person_id": str(r["person_id"]), "work_date": d.isoformat(), "sheet_id": sid, "revision": r["revision"],
"status": "superseded" if superseded else "approved", "normal_hours": _h(normal), "overtime_hours": _h(over),
"absence_hours": _h(away.get(d, Decimal(0))), "changed_at": changed_at.replace(tzinfo=timezone.utc).isoformat()})
rows.sort(key=lambda x: (int(x["person_id"]), x["work_date"], x["revision"]))
return {"company_code": company_code, "from": start.isoformat(), "to": end.isoformat(),
"as_of": _now().astimezone(timezone.utc).isoformat(), "rows": rows}
# The billing run (TS-13/14) declares its use cases on the same brick; imported last so every helper above exists.
from brick_timesheets.application import billing, migration, reports # noqa: E402,F401