dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
src/brick_timesheets/infrastructure/tables.py
"""The tables, and one repository per concern. Every table is prefixed `ts_`, has a natural primary key (never an autoincrement), money is
NUMERIC(18,2) and hours and rates 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 claim, a line approval, a billing claim, a dated rate or setting) 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 rows
that move FORWARD (a sheet's status, an outbox row's status) or are edited while the week is open (a line, an entry) are written with the keyed
upsert. Statements per request do not grow with the rows around them: a sheet loads in four queries however many entries it has, entries are
written in ONE statement, and every list is one query.
"""
import json
from datetime import date, datetime, timedelta, 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_timesheets.domain import sheet as sh
from brick_timesheets.domain.errors import RuleBroken
from brick_timesheets.domain.money import hours4, money
from brick_timesheets.domain.values import DEFAULT_ACTIVITIES, DEFAULT_RATE_CURRENCY, DEFAULT_RATE_FROM, DEFAULT_RATES, SETTING_DEFS
md = MetaData()
MONEY = Numeric(18, 2)
HOURS = Numeric(19, 4)
ts_sheet = Table("ts_sheet", md,
Column("sheet_id", String, primary_key=True),
Column("revision", Integer, primary_key=True),
Column("person_id", BigInteger),
Column("company_code", String),
Column("week_start", Date),
Column("person_email", String),
Column("person_sub", String),
Column("status", String),
Column("expected_minutes", Integer),
Column("expected_json", String),
Column("home_cost_centre", String),
Column("approval_route", String),
Column("approver_person_id", BigInteger),
Column("approver_email", String),
Column("approver_sub", String),
Column("rerouted_from", String),
Column("currency", String),
Column("submitted_at", TIMESTAMP),
Column("submitted_by", String),
Column("submitted_sub", String),
Column("approved_by", String),
Column("approved_sub", String),
Column("approved_at", TIMESTAMP),
Column("approved_on", Date),
Column("journal_id", BigInteger),
Column("ended_by", String),
Column("ended_sub", String),
Column("ended_at", TIMESTAMP),
Column("reason", String),
Column("last_line_no", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
ts_line = Table("ts_line", md,
Column("sheet_id", String, primary_key=True),
Column("revision", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("kind", String),
Column("activity_code", String),
Column("project_id", String),
Column("item_id", String),
Column("cost_centre", String),
Column("billable", Boolean),
Column("bill_via", String),
Column("sales_order_id", String),
Column("so_line_no", Integer),
Column("product_id", String),
Column("rate_group", String),
Column("unit_cost", HOURS),
Column("absence_ref", String),
Column("description", String),
Column("lead_person_id", BigInteger),
Column("billing", String),
Column("stale", Boolean),
Column("stale_reason", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
ts_entry = Table("ts_entry", md,
Column("sheet_id", String, primary_key=True),
Column("revision", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("work_date", Date, primary_key=True),
Column("hours", HOURS),
Column("note", String),
Column("unit_cost", HOURS),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# A project lead's approval of one line (route project_lead_then_manager): append-only, a plain INSERT keyed by the line.
ts_line_approval = Table("ts_line_approval", md,
Column("sheet_id", String, primary_key=True),
Column("revision", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("approver_person_id", BigInteger),
Column("decided_by", String),
Column("decided_sub", String),
Column("decided_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
ts_activity = Table("ts_activity", md,
Column("company_code", String, primary_key=True),
Column("activity_code", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("name", String),
Column("billable_default", Boolean),
Column("internal_only", Boolean),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
ts_cost_rate = Table("ts_cost_rate", md,
Column("company_code", String, primary_key=True),
Column("rate_group", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("rate_amount", HOURS),
Column("currency", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
ts_person_rate_group = Table("ts_person_rate_group", md,
Column("person_id", BigInteger, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("rate_group", String),
Column("set_by", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# The row a billable line (or its correction) races for, so it is told to the order once. purpose: delivery (slice A) | ar_run (slice B).
# quantity is signed: a correction is negative. Written in the transaction that approves the revision, with its outbox row.
ts_billing_claim = Table("ts_billing_claim", md,
Column("sheet_id", String, primary_key=True),
Column("revision", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True),
Column("purpose", String, primary_key=True),
Column("run_id", String),
Column("delivery_ref", String),
Column("sales_order_id", String),
Column("so_line_no", Integer),
Column("quantity", HOURS),
Column("unit_price", HOURS),
Column("priced_from", String),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("product_id", String), # 00003: the time product an ar_run claim bills (a correction nets the same product)
Column("tax_category", String)) # 00003: that product's tax category
# One month's billable hours of one project, billed through ar (TS-13): draft -> released | held -> released | cancelled. lines_json is the invoice's lines
# as the run priced them (what a release sends, never recomputed); the claims of its hours are ts_billing_claim rows (purpose ar_run, run_id).
ts_billing_run = Table("ts_billing_run", md,
Column("run_id", String, primary_key=True),
Column("project_id", String),
Column("period_id", String),
Column("run_no", Integer),
Column("company_code", String),
Column("bp_id", BigInteger),
Column("currency", String),
Column("status", String),
Column("source_id", String),
Column("lines_json", String),
Column("total_amount", MONEY),
Column("claim_count", Integer),
Column("invoice_id", String),
Column("reason", String),
Column("created_by", String),
Column("released_by", String),
Column("released_sub", String),
Column("released_at", TIMESTAMP),
Column("released_on", Date),
Column("ended_by", String),
Column("ended_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# One guard row per operation on a person's week: the operation takes the claim in ITS transaction, so two racers cannot both succeed and a
# loser is a 409. scopes: open, submit, approval, close (return), reopen.
ts_claim = Table("ts_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))
# A side effect waiting for a neighbour, written in the SAME transaction as the state change that needs it and sent only after that
# commit. target: gl | sales-order | ar | notify (a reminder).
ts_outbox = Table("ts_outbox", md,
Column("target", String, primary_key=True),
Column("ref", String, primary_key=True),
Column("sheet_id", String),
Column("payload_json", String),
Column("status", String),
Column("attempts", Integer),
Column("last_error", String),
Column("sent_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
ts_settings = Table("ts_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))
# One Odoo migration run (TS-22): 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.
ts_migration_run = Table("ts_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 re-load matches and what a later brick resolves through.
ts_migration_map = Table("ts_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))
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, at: datetime) -> None:
"""A FACT row (or rows, in ONE statement): a plain INSERT - a second writer of the same key is rejected (23505), never overwritten."""
rows = row if isinstance(row, list) else [row]
if rows:
conn.execute(insert(table), [{**r, "created_at": _utc(at), "updated_at": _utc(at)} for r in rows])
def _all(conn, table: Table, cond=None, order=()) -> list:
q = select(table)
if cond is not None:
q = q.where(cond)
return [dict(m) for m in conn.execute(q.order_by(*order)).mappings()]
def _dec(v, f=money) -> Optional[Decimal]:
return None if v is None else f(v)
ROOT_COLS = tuple(c.name for c in ts_sheet.columns if c.name not in ("created_at", "updated_at"))
class SheetRepository:
"""A sheet revision whole: the root, its lines, their entries and the line approvals - four queries however many entries."""
def __init__(self, conn):
self.conn = conn
def get(self, sheet_id: str, revision: Optional[int] = None) -> Optional[sh.Sheet]:
q = select(ts_sheet).where(ts_sheet.c.sheet_id == sheet_id)
q = q.where(ts_sheet.c.revision == revision) if revision is not None else q.order_by(ts_sheet.c.revision.desc()).limit(1)
root = self.conn.execute(q).mappings().first()
if root is None:
return None
rev = root["revision"]
key = (ts_line.c.sheet_id == sheet_id) & (ts_line.c.revision == rev)
lines = _all(self.conn, ts_line, key, (ts_line.c.line_no,))
entries = _all(self.conn, ts_entry, (ts_entry.c.sheet_id == sheet_id) & (ts_entry.c.revision == rev), (ts_entry.c.line_no, ts_entry.c.work_date))
approvals = _all(self.conn, ts_line_approval, (ts_line_approval.c.sheet_id == sheet_id) & (ts_line_approval.c.revision == rev),
(ts_line_approval.c.line_no,))
return self._make(dict(root), lines, entries, approvals)
@staticmethod
def _make(root: dict, lines: list, entries: list, approvals: list) -> sh.Sheet:
by_line: dict = {}
for e in entries:
by_line.setdefault(e["line_no"], []).append(sh.Entry(e["work_date"], hours4(e["hours"]), e["note"] or "", _dec(e["unit_cost"], hours4)))
s = sh.Sheet(
sheet_id=root["sheet_id"], person_id=root["person_id"], company_code=root["company_code"], week_start=root["week_start"],
person=sh.Identity(root["person_email"] or "", root["person_sub"] or ""), revision=root["revision"], status=root["status"],
expected=json.loads(root["expected_json"] or "{}"), home_cost_centre=root["home_cost_centre"] or "",
approval_route=root["approval_route"] or "", approver_person_id=root["approver_person_id"],
approver=sh.Identity(root["approver_email"] or "", root["approver_sub"] or "") if (root["approver_email"] or root["approver_sub"]) else None,
rerouted_from=root["rerouted_from"], currency=root["currency"] or "", submitted_at=root["submitted_at"],
approved_by=root["approved_by"] or "", approved_sub=root["approved_sub"] or "", approved_at=root["approved_at"],
approved_on=root["approved_on"], journal_id=root["journal_id"], ended_by=root["ended_by"] or "", ended_sub=root["ended_sub"] or "",
ended_at=root["ended_at"], reason=root["reason"] or "", last_line_no=root["last_line_no"] or 0,
lines=[sh.Line(r["line_no"], r["kind"], r["activity_code"] or "", r["project_id"] or "", r["item_id"] or "", r["cost_centre"] or "",
bool(r["billable"]), r["bill_via"] or "none", r["sales_order_id"] or "", r["so_line_no"], r["product_id"] or "",
r["rate_group"] or "", _dec(r["unit_cost"], hours4), r["absence_ref"] or "", r["description"] or "", r["lead_person_id"],
r["billing"] or "", bool(r["stale"]), r["stale_reason"] or "", by_line.get(r["line_no"], [])) for r in lines],
approvals=[sh.LineApproval(a["line_no"], a["approver_person_id"], a["decided_by"] or "", a["decided_sub"] or "", a["decided_at"])
for a in approvals])
return s
def _root_row(self, s: sh.Sheet) -> dict:
return {"sheet_id": s.sheet_id, "revision": s.revision, "person_id": s.person_id, "company_code": s.company_code, "week_start": s.week_start,
"person_email": s.person.email, "person_sub": s.person.sub, "status": s.status, "expected_minutes": sum(s.expected.values()),
"expected_json": json.dumps(s.expected, sort_keys=True), "home_cost_centre": s.home_cost_centre or None,
"approval_route": s.approval_route or None, "approver_person_id": s.approver_person_id,
"approver_email": s.approver.email if s.approver else None, "approver_sub": s.approver.sub if s.approver else None,
"rerouted_from": s.rerouted_from, "currency": s.currency or None, "submitted_at": _utc(s.submitted_at),
"submitted_by": s.person.email if s.submitted_at else None, "submitted_sub": s.person.sub if s.submitted_at else None,
"approved_by": s.approved_by or None, "approved_sub": s.approved_sub or None, "approved_at": _utc(s.approved_at),
"approved_on": s.approved_on, "journal_id": s.journal_id, "ended_by": s.ended_by or None, "ended_sub": s.ended_sub or None,
"ended_at": _utc(s.ended_at), "reason": s.reason or None, "last_line_no": s.last_line_no}
def create(self, s: sh.Sheet, at: datetime) -> None:
"""A new revision: its root (a plain INSERT: the key is claimed once), its lines and entries - three statements however many."""
_plain(self.conn, ts_sheet, self._root_row(s), at)
self.add_lines(s.sheet_id, s.revision, s.lines, at)
def save(self, s: sh.Sheet, at: datetime) -> None:
"""The root moves forward (status, approver, approval, journal): a keyed upsert of the whole row."""
upsert(self.conn, ts_sheet, [{**self._root_row(s), "updated_at": _utc(at)}], ("sheet_id", "revision"))
@staticmethod
def _line_row(sheet_id: str, revision: int, ln: sh.Line) -> dict:
return {"sheet_id": sheet_id, "revision": revision, "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, "unit_cost": ln.unit_cost, "absence_ref": ln.absence_ref or None,
"description": ln.description or None, "lead_person_id": ln.lead_person_id, "billing": ln.billing or None, "stale": ln.stale,
"stale_reason": ln.stale_reason or None}
@staticmethod
def _entry_row(sheet_id: str, revision: int, line_no: int, e: sh.Entry) -> dict:
return {"sheet_id": sheet_id, "revision": revision, "line_no": line_no, "work_date": e.work_date, "hours": e.hours,
"note": e.note or None, "unit_cost": e.unit_cost}
def add_lines(self, sheet_id: str, revision: int, lines: list, at: datetime) -> None:
"""New lines and their entries: ONE statement each, however many."""
_plain(self.conn, ts_line, [self._line_row(sheet_id, revision, ln) for ln in lines], at)
_plain(self.conn, ts_entry, [self._entry_row(sheet_id, revision, ln.line_no, e) for ln in lines for e in ln.entries], at)
def rewrite_lines(self, sheet_id: str, revision: int, lines: list, at: datetime) -> None:
"""Lines (and entries) that already exist, written again whole: the terms and rates a submission copies onto them."""
if lines:
upsert(self.conn, ts_line, [{**self._line_row(sheet_id, revision, ln), "updated_at": _utc(at)} for ln in lines], ("sheet_id", "revision", "line_no"))
entries = [{**self._entry_row(sheet_id, revision, ln.line_no, e), "updated_at": _utc(at)} for ln in lines for e in ln.entries]
if entries:
upsert(self.conn, ts_entry, entries, ("sheet_id", "revision", "line_no", "work_date"))
def remove_lines(self, sheet_id: str, revision: int, line_nos: list) -> None:
if not line_nos:
return
for t in (ts_entry, ts_line):
self.conn.execute(delete(t).where((t.c.sheet_id == sheet_id) & (t.c.revision == revision) & t.c.line_no.in_(line_nos)))
def set_entries(self, sheet_id: str, revision: int, line_no: int, put: list, clear: list, at: datetime) -> None:
"""Hours of one line: the days with hours are written (keyed upsert), the days set to zero are removed - two statements at most."""
if put:
upsert(self.conn, ts_entry, [{**self._entry_row(sheet_id, revision, line_no, e), "updated_at": _utc(at)} for e in put],
("sheet_id", "revision", "line_no", "work_date"))
if clear:
self.conn.execute(delete(ts_entry).where((ts_entry.c.sheet_id == sheet_id) & (ts_entry.c.revision == revision)
& (ts_entry.c.line_no == line_no) & ts_entry.c.work_date.in_(clear)))
def add_approvals(self, sheet_id: str, revision: int, rows: list, at: datetime) -> None:
_plain(self.conn, ts_line_approval, [{"sheet_id": sheet_id, "revision": revision, "line_no": a.line_no, "approver_person_id": a.approver_person_id,
"decided_by": a.decided_by, "decided_sub": a.decided_sub, "decided_at": _utc(a.decided_at)} for a in rows], at)
# ── lists: one query each ────────────────────────────────────────────────────────────────
def summaries(self, *, mine: Optional[sh.Identity] = None, waiting_for: Optional[sh.Identity] = None, lead: Optional[int] = None,
everything: bool = False, limit: int = 200) -> list:
"""The caller's weeks, the weeks waiting for them, the weeks with lines waiting for their lead approval, or (everything) every week:
the LATEST revision of each, root rows only."""
latest = select(ts_sheet.c.sheet_id, func.max(ts_sheet.c.revision).label("rev")).group_by(ts_sheet.c.sheet_id).subquery()
q = select(ts_sheet).join(latest, (ts_sheet.c.sheet_id == latest.c.sheet_id) & (ts_sheet.c.revision == latest.c.rev))
conds = []
if everything:
conds.append(ts_sheet.c.sheet_id.isnot(None))
if mine is not None:
own = [ts_sheet.c.person_sub == mine.sub] if mine.sub else []
own += [ts_sheet.c.person_email == mine.email.lower()] if mine.email else []
conds.append(or_(*own) if own else ts_sheet.c.sheet_id.is_(None))
if waiting_for is not None:
w = [ts_sheet.c.approver_sub == waiting_for.sub] if waiting_for.sub else []
w += [ts_sheet.c.approver_email == waiting_for.email.lower()] if waiting_for.email else []
conds.append((ts_sheet.c.status == "submitted") & (or_(*w) if w else ts_sheet.c.sheet_id.is_(None)))
if lead is not None:
led = select(ts_line.c.sheet_id, ts_line.c.revision).where(ts_line.c.lead_person_id == lead).distinct().subquery()
q = q.join(led, (ts_sheet.c.sheet_id == led.c.sheet_id) & (ts_sheet.c.revision == led.c.revision), isouter=True)
conds.append((ts_sheet.c.status == "submitted") & led.c.sheet_id.isnot(None))
if not conds:
return []
return [dict(r) for r in self.conn.execute(q.where(or_(*conds)).order_by(ts_sheet.c.sheet_id).limit(limit)).mappings()]
def revisions(self, sheet_id: str) -> list:
"""[(revision, status)] of every revision of the week, oldest first: how a returned or corrected week reads."""
rows = self.conn.execute(select(ts_sheet.c.revision, ts_sheet.c.status).where(ts_sheet.c.sheet_id == sheet_id).order_by(ts_sheet.c.revision))
return [(r["revision"], r["status"]) for r in rows.mappings()]
def prior_approved(self, sheet_id: str, before: int) -> Optional[int]:
"""The latest revision of this week before `before` that was approved (its effects are what a correction moves by the difference of)."""
return self.conn.execute(select(func.max(ts_sheet.c.revision)).where(
(ts_sheet.c.sheet_id == sheet_id) & (ts_sheet.c.revision < before) & ts_sheet.c.approved_at.isnot(None))).scalar()
def approved_roots(self, company_code: str, start: date, end: date, person_id: Optional[int]) -> list:
"""Every approved revision of the weeks that touch [start, end] in this company: ONE query."""
q = select(ts_sheet).where((ts_sheet.c.company_code == company_code) & ts_sheet.c.approved_at.isnot(None)
& (ts_sheet.c.week_start >= start - timedelta(days=6)) & (ts_sheet.c.week_start <= end))
if person_id is not None:
q = q.where(ts_sheet.c.person_id == person_id)
return [dict(r) for r in self.conn.execute(q.order_by(ts_sheet.c.person_id, ts_sheet.c.sheet_id, ts_sheet.c.revision)).mappings()]
def approved_ar_lines(self, project_id: str) -> list:
"""The LATEST approved revision of every week with a billable ar line on this project, whole (lines and entries): four queries however many weeks."""
q = (select(ts_line.c.sheet_id).distinct().select_from(ts_line.join(ts_sheet, (ts_line.c.sheet_id == ts_sheet.c.sheet_id) & (ts_line.c.revision == ts_sheet.c.revision)))
.where(ts_sheet.c.approved_at.isnot(None) & (ts_line.c.project_id == project_id) & (ts_line.c.bill_via == "ar")))
ids = sorted(r[0] for r in self.conn.execute(q))
if not ids:
return []
roots = _all(self.conn, ts_sheet, ts_sheet.c.sheet_id.in_(ids) & ts_sheet.c.approved_at.isnot(None), (ts_sheet.c.sheet_id, ts_sheet.c.revision))
latest = {r["sheet_id"]: r for r in roots}
lines = _all(self.conn, ts_line, ts_line.c.sheet_id.in_(ids) & (ts_line.c.project_id == project_id), (ts_line.c.sheet_id, ts_line.c.revision, ts_line.c.line_no))
entries = _all(self.conn, ts_entry, ts_entry.c.sheet_id.in_(ids), (ts_entry.c.sheet_id, ts_entry.c.revision, ts_entry.c.line_no, ts_entry.c.work_date))
out = []
for sid, root in latest.items():
mine = [ln for ln in lines if ln["sheet_id"] == sid and ln["revision"] == root["revision"]]
ents = [e for e in entries if e["sheet_id"] == sid and e["revision"] == root["revision"]]
out.append(self._make(root, mine, ents, []))
return out
def project_rows(self, project_id: str, start: date, end: date) -> tuple:
"""(one row per entry of the project's lines in the dates, {sheet_id: the standing approved revision}): TWO queries however many weeks. Only this
project's rows, never another's."""
q = (select(ts_entry.c.sheet_id, ts_entry.c.revision, ts_entry.c.line_no, ts_entry.c.hours, ts_entry.c.unit_cost, ts_line.c.item_id, ts_line.c.billable,
ts_line.c.lead_person_id, ts_sheet.c.status, ts_sheet.c.approved_at, ts_sheet.c.person_id, ts_sheet.c.currency)
.select_from(ts_entry.join(ts_line, (ts_entry.c.sheet_id == ts_line.c.sheet_id) & (ts_entry.c.revision == ts_line.c.revision) & (ts_entry.c.line_no == ts_line.c.line_no))
.join(ts_sheet, (ts_entry.c.sheet_id == ts_sheet.c.sheet_id) & (ts_entry.c.revision == ts_sheet.c.revision)))
.where((ts_line.c.project_id == project_id) & (ts_line.c.kind == "project") & (ts_entry.c.work_date >= start) & (ts_entry.c.work_date <= end)
& ts_sheet.c.status.in_(("submitted", "approved", "superseded"))))
rows = [dict(r) for r in self.conn.execute(q).mappings()]
ids = sorted({r["sheet_id"] for r in rows})
standing: dict = {}
if ids:
q2 = (select(ts_sheet.c.sheet_id, func.max(ts_sheet.c.revision).label("rev")).where(ts_sheet.c.sheet_id.in_(ids) & ts_sheet.c.approved_at.isnot(None))
.group_by(ts_sheet.c.sheet_id))
standing = {r["sheet_id"]: r["rev"] for r in self.conn.execute(q2).mappings()}
return rows, standing
def waiting(self) -> list:
"""Every week still open or submitted (the chase reads them all): ONE query."""
return _all(self.conn, ts_sheet, ts_sheet.c.status.in_(("open", "submitted")), (ts_sheet.c.sheet_id, ts_sheet.c.revision))
def entries_of(self, sheet_ids: list) -> list:
"""The entries of those sheets (every revision) with their line's kind: ONE query."""
if not sheet_ids:
return []
q = (select(ts_entry.c.sheet_id, ts_entry.c.revision, ts_entry.c.line_no, ts_entry.c.work_date, ts_entry.c.hours, ts_line.c.kind)
.select_from(ts_entry.join(ts_line, (ts_entry.c.sheet_id == ts_line.c.sheet_id) & (ts_entry.c.revision == ts_line.c.revision)
& (ts_entry.c.line_no == ts_line.c.line_no)))
.where(ts_entry.c.sheet_id.in_(sorted(set(sheet_ids)))))
return [dict(r) for r in self.conn.execute(q).mappings()]
class ClaimRepository:
"""The one guard key per operation on a week: every operation takes its 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(ts_claim).where(
(ts_claim.c.claim_scope == scope) & (ts_claim.c.claim_key == key))).scalar() or 0
def once(self, scope: str, key: str, at: datetime, by: str = "", ref: str = "") -> None:
"""A key that can be claimed exactly ONCE (one submission of one revision, one approval): 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 timesheet again")
_plain(self.conn, ts_claim, {"claim_scope": scope, "claim_key": key, "n": 1, "ref": ref, "claimed_by": by, "claimed_at": _utc(at)}, at)
class OutboxRepository:
def __init__(self, conn):
self.conn = conn
def add_many(self, rows: list, at: datetime) -> None:
"""Pending rows, in ONE statement: (target, ref, sheet_id, payload)."""
_plain(self.conn, ts_outbox, [{"target": t, "ref": r, "sheet_id": sid, "payload_json": json.dumps(p, sort_keys=True), "status": "pending",
"attempts": 0} for t, r, sid, p in rows], at)
def get(self, target: str, ref: str) -> Optional[dict]:
row = self.conn.execute(select(ts_outbox).where((ts_outbox.c.target == target) & (ts_outbox.c.ref == ref))).mappings().first()
return dict(row) if row else None
def many(self, target: str, refs: list) -> dict:
"""ref -> row for a set of refs: one query per 500."""
got: dict = {}
refs = sorted(set(refs))
for i in range(0, len(refs), 500):
for r in _all(self.conn, ts_outbox, (ts_outbox.c.target == target) & ts_outbox.c.ref.in_(refs[i:i + 500])):
got[r["ref"]] = r
return got
def listed(self, target: str, status: Optional[str] = None, limit: int = 200) -> list:
cond = ts_outbox.c.target == target
if status:
cond &= ts_outbox.c.status == status
return [dict(r) for r in self.conn.execute(select(ts_outbox).where(cond).order_by(ts_outbox.c.created_at.desc(), ts_outbox.c.ref).limit(limit)).mappings()]
def pending(self, sheet_id: Optional[str] = None, targets: Optional[tuple] = None) -> list:
q = select(ts_outbox).where(ts_outbox.c.status == "pending")
if sheet_id:
q = q.where(ts_outbox.c.sheet_id == sheet_id)
if targets:
q = q.where(ts_outbox.c.target.in_(targets))
return [dict(r) for r in self.conn.execute(q.order_by(ts_outbox.c.created_at, ts_outbox.c.target, ts_outbox.c.ref)).mappings()]
def of_sheet(self, sheet_id: str) -> list:
return _all(self.conn, ts_outbox, ts_outbox.c.sheet_id == sheet_id, (ts_outbox.c.target, ts_outbox.c.ref))
def mark(self, row: dict, status: str, error: str, at: datetime) -> None:
keep = {k: row[k] for k in ("target", "ref", "sheet_id", "payload_json")}
upsert(self.conn, ts_outbox, [{**keep, "status": status, "attempts": (row["attempts"] or 0) + 1, "last_error": error or None,
"sent_at": _utc(at) if status == "sent" else None, "updated_at": _utc(at)}], ("target", "ref"))
class BillingClaimRepository:
def __init__(self, conn):
self.conn = conn
def of_sheet(self, sheet_id: str, purpose: str = "delivery") -> list:
"""What the order was told about this week so far, every revision: ONE query."""
return _all(self.conn, ts_billing_claim, (ts_billing_claim.c.sheet_id == sheet_id) & (ts_billing_claim.c.purpose == purpose),
(ts_billing_claim.c.revision, ts_billing_claim.c.line_no))
def add_many(self, rows: list, at: datetime) -> None:
_plain(self.conn, ts_billing_claim, [{**r, "claimed_at": _utc(at)} for r in rows], at)
def ar_billed(self, sheet_ids: list, project_id: str) -> dict:
"""(sheet_id, line_no) -> hours this project's ar runs already claimed, every revision (corrections are negative): ONE query."""
if not sheet_ids:
return {}
q = (select(ts_billing_claim.c.sheet_id, ts_billing_claim.c.line_no, func.sum(ts_billing_claim.c.quantity).label("q"))
.select_from(ts_billing_claim.join(ts_billing_run, ts_billing_claim.c.run_id == ts_billing_run.c.run_id))
.where((ts_billing_claim.c.purpose == "ar_run") & (ts_billing_run.c.project_id == project_id) & ts_billing_claim.c.sheet_id.in_(sorted(set(sheet_ids))))
.group_by(ts_billing_claim.c.sheet_id, ts_billing_claim.c.line_no))
return {(r["sheet_id"], r["line_no"]): hours4(r["q"] or 0) for r in self.conn.execute(q).mappings()}
def ar_first_price(self, sheet_ids: list, project_id: str) -> dict:
"""(sheet_id, line_no) -> (unit_price, priced_from) of the line's FIRST claim: what a correction is priced at. ONE query."""
if not sheet_ids:
return {}
q = (select(ts_billing_claim).select_from(ts_billing_claim.join(ts_billing_run, ts_billing_claim.c.run_id == ts_billing_run.c.run_id))
.where((ts_billing_claim.c.purpose == "ar_run") & (ts_billing_run.c.project_id == project_id) & ts_billing_claim.c.sheet_id.in_(sorted(set(sheet_ids))))
.order_by(ts_billing_claim.c.sheet_id, ts_billing_claim.c.line_no, ts_billing_claim.c.revision))
out: dict = {}
for r in self.conn.execute(q).mappings():
out.setdefault((r["sheet_id"], r["line_no"]), (hours4(r["unit_price"]), r["priced_from"] or ""))
return out
def ar_released(self, sheet_ids: list, project_id: str) -> dict:
"""(sheet_id, line_no) -> hours that RELEASED ar runs billed (corrections are negative): ONE query."""
if not sheet_ids:
return {}
q = (select(ts_billing_claim.c.sheet_id, ts_billing_claim.c.line_no, func.sum(ts_billing_claim.c.quantity).label("q"))
.select_from(ts_billing_claim.join(ts_billing_run, ts_billing_claim.c.run_id == ts_billing_run.c.run_id))
.where((ts_billing_claim.c.purpose == "ar_run") & (ts_billing_run.c.project_id == project_id) & (ts_billing_run.c.status == "released")
& ts_billing_claim.c.sheet_id.in_(sorted(set(sheet_ids))))
.group_by(ts_billing_claim.c.sheet_id, ts_billing_claim.c.line_no))
return {(r["sheet_id"], r["line_no"]): hours4(r["q"] or 0) for r in self.conn.execute(q).mappings()}
def of_run(self, run_id: str) -> list:
return _all(self.conn, ts_billing_claim, (ts_billing_claim.c.purpose == "ar_run") & (ts_billing_claim.c.run_id == run_id),
(ts_billing_claim.c.sheet_id, ts_billing_claim.c.revision, ts_billing_claim.c.line_no))
def delete_run(self, run_id: str) -> None:
"""A cancelled run lets go of its hours: its claims are removed, so the lines fall into the next run."""
self.conn.execute(delete(ts_billing_claim).where((ts_billing_claim.c.purpose == "ar_run") & (ts_billing_claim.c.run_id == run_id)))
def people_of_run(self, run_id: str) -> list:
"""[(person_email, person_sub)] of everyone whose hours are in the run: ONE query."""
q = (select(ts_sheet.c.person_email, ts_sheet.c.person_sub).distinct()
.select_from(ts_billing_claim.join(ts_sheet, (ts_billing_claim.c.sheet_id == ts_sheet.c.sheet_id) & (ts_billing_claim.c.revision == ts_sheet.c.revision)))
.where((ts_billing_claim.c.purpose == "ar_run") & (ts_billing_claim.c.run_id == run_id)))
return [(r["person_email"] or "", r["person_sub"] or "") for r in self.conn.execute(q).mappings()]
def ar_projects(self) -> list:
"""The projects with an approved billable line billed through ar: ONE query."""
q = (select(ts_line.c.project_id).distinct()
.select_from(ts_line.join(ts_sheet, (ts_line.c.sheet_id == ts_sheet.c.sheet_id) & (ts_line.c.revision == ts_sheet.c.revision)))
.where(ts_sheet.c.approved_at.isnot(None) & (ts_line.c.bill_via == "ar") & (ts_line.c.billable == True)) # noqa: E712
.order_by(ts_line.c.project_id))
return [r[0] for r in self.conn.execute(q) if r[0]]
class BillingRunRepository:
def __init__(self, conn):
self.conn = conn
def get(self, run_id: str) -> Optional[dict]:
row = self.conn.execute(select(ts_billing_run).where(ts_billing_run.c.run_id == run_id)).mappings().first()
return dict(row) if row else None
def of_period(self, project_id: str, period_id: str) -> list:
return _all(self.conn, ts_billing_run, (ts_billing_run.c.project_id == project_id) & (ts_billing_run.c.period_id == period_id), (ts_billing_run.c.run_no,))
def listing(self, status: Optional[str] = None, project_id: Optional[str] = None) -> list:
cond = None
for col, v in ((ts_billing_run.c.status, status), (ts_billing_run.c.project_id, project_id)):
if v:
cond = (col == v) if cond is None else cond & (col == v)
return _all(self.conn, ts_billing_run, cond, (ts_billing_run.c.period_id, ts_billing_run.c.project_id, ts_billing_run.c.run_no))
def add(self, row: dict, at: datetime) -> None:
_plain(self.conn, ts_billing_run, {**row, "released_at": _utc(row.get("released_at")), "ended_at": _utc(row.get("ended_at"))}, at)
def save(self, row: dict, at: datetime) -> None:
"""A run moves forward (a status, an invoice id): the keyed upsert of the whole row."""
keep = {k: v for k, v in row.items() if k not in ("created_at", "updated_at")}
for k in ("released_at", "ended_at"):
keep[k] = _utc(keep.get(k))
upsert(self.conn, ts_billing_run, [{**keep, "updated_at": _utc(at)}], ("run_id",))
class MigrationRepository:
"""The Odoo slice's rows (TS-22): the run, the id map, the carried weeks in bulk, and what a reconciliation reads back."""
def __init__(self, conn):
self.conn = conn
def run(self, run_id: str) -> Optional[dict]:
row = self.conn.execute(select(ts_migration_run).where(ts_migration_run.c.run_id == run_id)).mappings().first()
return dict(row) if row else None
def runs(self) -> list:
return _all(self.conn, ts_migration_run, None, (ts_migration_run.c.run_id,))
def add_run(self, row: dict, at: datetime) -> None:
_plain(self.conn, ts_migration_run, row, at)
def save_run(self, row: dict, at: datetime) -> None:
upsert(self.conn, ts_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, ts_migration_map, (ts_migration_map.c.odoo_db == db) & (ts_migration_map.c.odoo_model == model)
& ts_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):
_plain(self.conn, ts_migration_map, rows[i:i + 500], at)
def of_run(self, run_id: str) -> list:
return _all(self.conn, ts_migration_map, ts_migration_map.c.run_id == run_id, (ts_migration_map.c.odoo_model, ts_migration_map.c.odoo_id))
def add_sheets(self, sheets: list, at: datetime) -> None:
"""Carried weeks in bulk: their roots, lines and entries, a handful of statements however many weeks (plain INSERTs: a key is claimed once)."""
roots = [SheetRepository._root_row(None, s) for s in sheets]
lines = [SheetRepository._line_row(s.sheet_id, s.revision, ln) for s in sheets for ln in s.lines]
entries = [SheetRepository._entry_row(s.sheet_id, s.revision, ln.line_no, e) for s in sheets for ln in s.lines for e in ln.entries]
for rows, table in ((roots, ts_sheet), (lines, ts_line), (entries, ts_entry)):
for i in range(0, len(rows), 200):
_plain(self.conn, table, rows[i:i + 200], at)
def claims_of(self, sheet_ids: list) -> list:
out: list = []
ids = sorted(set(sheet_ids))
for i in range(0, len(ids), 400):
out += _all(self.conn, ts_billing_claim, ts_billing_claim.c.sheet_id.in_(ids[i:i + 400]), (ts_billing_claim.c.sheet_id, ts_billing_claim.c.line_no))
return out
def hours_by_project(self, sheet_ids: list) -> dict:
"""project_id -> hours of these sheets' lines (revision 1 is the only revision a run writes)."""
out: dict = {}
ids = sorted(set(sheet_ids))
for i in range(0, len(ids), 400):
q = (select(ts_line.c.project_id, func.sum(ts_entry.c.hours).label("h"))
.select_from(ts_entry.join(ts_line, (ts_entry.c.sheet_id == ts_line.c.sheet_id) & (ts_entry.c.revision == ts_line.c.revision)
& (ts_entry.c.line_no == ts_line.c.line_no)))
.where(ts_entry.c.sheet_id.in_(ids[i:i + 400])).group_by(ts_line.c.project_id))
for r in self.conn.execute(q).mappings():
out[r["project_id"]] = out.get(r["project_id"], Decimal(0)) + hours4(r["h"] or 0)
return out
def sheets_of(self, sheet_ids: list) -> list:
"""Every revision of these weeks (root rows): a run's own sheets are revision 1; a later revision means a correction happened since."""
out: list = []
ids = sorted(set(sheet_ids))
for i in range(0, len(ids), 400):
out += _all(self.conn, ts_sheet, ts_sheet.c.sheet_id.in_(ids[i:i + 400]), (ts_sheet.c.sheet_id, ts_sheet.c.revision))
return out
def historic_runs(self, run_id: str) -> list:
return _all(self.conn, ts_billing_run, ts_billing_run.c.period_id == run_id, (ts_billing_run.c.run_id,))
def delete_run(self, run_id: str, sheet_ids: list) -> None:
"""Take a discarded run's rows back out: its weeks, their lines and entries, its claims and historic billing runs, its map. Only ever for a run
nothing has been done to since the load."""
ids = sorted(set(sheet_ids))
for i in range(0, len(ids), 400):
chunk = ids[i:i + 400]
for t in (ts_entry, ts_line, ts_billing_claim, ts_sheet):
self.conn.execute(delete(t).where(t.c.sheet_id.in_(chunk)))
self.conn.execute(delete(ts_billing_run).where(ts_billing_run.c.period_id == run_id))
self.conn.execute(delete(ts_migration_map).where(ts_migration_map.c.run_id == run_id))
class ActivityRepository:
def __init__(self, conn):
self.conn = conn
def in_force(self, company_code: str, codes: set, on: date) -> dict:
"""code -> the activity row in force on `on`: the company's own rows over the '*' default, then the latest valid_from on or before it."""
if not codes:
return {}
rows = _all(self.conn, ts_activity, ts_activity.c.company_code.in_((company_code, "*")) & ts_activity.c.activity_code.in_(sorted(codes))
& (ts_activity.c.valid_from <= on), (ts_activity.c.valid_from,))
out: dict = {}
for code in codes:
mine = [r for r in rows if r["activity_code"] == code]
own = [r for r in mine if r["company_code"] != "*"]
pick = (own or mine)[-1:]
if pick:
out[code] = pick[0]
return out
def put(self, company_code: str, code: str, name: str, billable: bool, internal_only: bool, valid_from: date, at: datetime) -> None:
_plain(self.conn, ts_activity, {"company_code": company_code, "activity_code": code, "valid_from": valid_from, "name": name,
"billable_default": billable, "internal_only": internal_only}, at)
def listing(self, company_code: str) -> list:
return _all(self.conn, ts_activity, ts_activity.c.company_code.in_((company_code, "*")), (ts_activity.c.activity_code, ts_activity.c.valid_from))
class RateRepository:
"""Standard cost rates per rate group, and which group a person is costed at - both dated rows, never edited."""
def __init__(self, conn):
self.conn = conn
def person_groups(self, person_id: int) -> list:
return _all(self.conn, ts_person_rate_group, ts_person_rate_group.c.person_id == person_id, (ts_person_rate_group.c.valid_from,))
def rates(self, company_code: str, groups: set) -> list:
if not groups:
return []
return _all(self.conn, ts_cost_rate, ts_cost_rate.c.company_code.in_((company_code, "*")) & ts_cost_rate.c.rate_group.in_(sorted(groups)),
(ts_cost_rate.c.valid_from,))
def listing(self, company_code: str) -> list:
return _all(self.conn, ts_cost_rate, ts_cost_rate.c.company_code.in_((company_code, "*")), (ts_cost_rate.c.rate_group, ts_cost_rate.c.valid_from))
def put_rate(self, company_code: str, group: str, amount: Decimal, currency: str, valid_from: date, by: str, at: datetime) -> None:
_plain(self.conn, ts_cost_rate, {"company_code": company_code, "rate_group": group, "valid_from": valid_from, "rate_amount": amount,
"currency": currency, "set_by": by}, at)
def put_person(self, person_id: int, group: str, valid_from: date, by: str, at: datetime) -> None:
_plain(self.conn, ts_person_rate_group, {"person_id": person_id, "valid_from": valid_from, "rate_group": group, "set_by": by}, at)
def has_rate_rows(self, company_code: str, group: str) -> bool:
return bool(self.conn.execute(select(func.count()).select_from(ts_cost_rate).where(
(ts_cost_rate.c.company_code == company_code) & (ts_cost_rate.c.rate_group == group))).scalar())
def has_person_rows(self, person_id: int) -> bool:
return bool(self.conn.execute(select(func.count()).select_from(ts_person_rate_group).where(ts_person_rate_group.c.person_id == person_id)).scalar())
def rate_group_on(groups: list, day: date) -> Optional[str]:
"""The person's rate group on a date: the latest row dated on or before it."""
pick = [g["rate_group"] for g in groups if g["valid_from"] <= day]
return pick[-1] if pick else None
def rate_on(rates: list, company_code: str, group: str, day: date) -> Optional[dict]:
"""The cost rate row in force: the company's own rows over the '*' default, the latest valid_from on or before the date."""
mine = [r for r in rates if r["rate_group"] == group and r["valid_from"] <= day]
own = [r for r in mine if r["company_code"] != "*"]
pick = (own or mine)[-1:]
return pick[0] if pick else None
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, ts_settings, None, (ts_settings.c.setting_key, ts_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 has(self, key: str) -> bool:
return any(r["setting_key"] == key for r in self.rows())
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,
which could put a row dated tomorrow in force at once."""
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, ts_settings, {"setting_key": key, "valid_from": valid_from, "value": str(value), "set_by": by}, at)
self._rows = None
class Repository:
"""Every repository on one connection (one unit of work)."""
def __init__(self, conn):
self.conn = conn
self.sheets = SheetRepository(conn)
self.claims = ClaimRepository(conn)
self.outbox = OutboxRepository(conn)
self.billing = BillingClaimRepository(conn)
self.runs = BillingRunRepository(conn)
self.migration = MigrationRepository(conn)
self.activities = ActivityRepository(conn)
self.rates = RateRepository(conn)
self.settings = SettingsRepository(conn)
def seed(self, at: Optional[datetime] = None) -> None:
"""The activities and placeholder cost rates every company starts with, once (company '*'). Idempotent; the same rows migration 00002 inserts."""
at = at or work.now()
have = {r["activity_code"] for r in _all(self.conn, ts_activity, ts_activity.c.company_code == "*")}
for code, name, billable, internal in DEFAULT_ACTIVITIES:
if code not in have:
self.activities.put("*", code, name, billable, internal, date(2000, 1, 1), at)
have_rates = {r["rate_group"] for r in _all(self.conn, ts_cost_rate, ts_cost_rate.c.company_code == "*")}
for group, amount in DEFAULT_RATES:
if group not in have_rates:
self.rates.put_rate("*", group, Decimal(amount), DEFAULT_RATE_CURRENCY, date.fromisoformat(DEFAULT_RATE_FROM), "migration", at)