"""The tables, and one repository per aggregate that loads and saves it whole.
Every table is prefixed pj_, has a natural primary key (never SERIAL), and a later column is an append.
schema.sql is the same column model; tests/test_schema.py keeps the two equal.
History (pj_event) is appended by the save that caused it, in the same transaction, keyed on the version that
save wrote: a save retried after a lost race writes a new version and its own events, and nothing twice.
"""
import json
from datetime import datetime, timezone
from typing import Optional
from erp_platform.repository import Aggregate, Part, upsert
from sqlalchemy import TIMESTAMP, BigInteger, Column, Date, Integer, MetaData, String, Table, select
from brick_projects.domain.item import Item
from brick_projects.domain.project import Project
from brick_projects.domain.state import DEFAULT_STATES, State
from brick_projects.domain.values import SETTING_ROWS, Settings, settings_from_rows
md = MetaData()
pj_project = Table("pj_project", md,
Column("project_id", String, primary_key=True),
Column("bp_id", BigInteger),
Column("source_kind", String),
Column("source_id", String),
Column("quote_id", String),
Column("owner", String),
Column("starts_on", Date),
Column("ends_on", Date),
Column("reconciled_at", TIMESTAMP),
Column("reconciliation_json", String),
Column("state", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
PJ_PROJECT_COLUMNS = ("project_id", "bp_id", "source_kind", "source_id", "quote_id", "owner", "starts_on",
"ends_on", "reconciled_at", "reconciliation_json", "state")
PROJECT = Aggregate(Part(pj_project, ("project_id",)), version="version")
pj_item = Table("pj_item", md,
Column("item_id", String, primary_key=True),
Column("project_id", String),
Column("bp_id", BigInteger),
Column("source_kind", String),
Column("source_id", String),
Column("line_id", String),
Column("product_id", String),
Column("billing_kind", String),
Column("billing_plan", String),
Column("source", String),
Column("title", String),
Column("owner", String),
Column("due_on", Date),
Column("blocked_reason", String),
Column("blocked_since", Date),
Column("blocked_days", Integer),
Column("submitted_at", TIMESTAMP),
Column("submitted_by", String),
Column("revision", Integer),
Column("warned_at", TIMESTAMP),
Column("approved_at", TIMESTAMP),
Column("approved_by", String),
Column("approved_how", String),
Column("delivered_at", TIMESTAMP),
Column("evidence", String),
Column("evidence_kind", String),
Column("makegood_for", String),
Column("billing_status", String),
Column("delivery_status", String),
Column("state", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("changes_reason", String), # 00002: appended, so a bucket on 00001 gains it by ALTER
Column("work_ref", String), # 00003: likewise
Column("submitted_sub", String)) # 00004: likewise
PJ_ITEM_COLUMNS = ("item_id", "project_id", "bp_id", "source_kind", "source_id", "line_id", "product_id",
"billing_kind", "billing_plan", "source", "title", "owner", "due_on", "blocked_reason",
"blocked_since", "blocked_days", "submitted_at", "submitted_by", "revision", "warned_at", "approved_at",
"approved_by", "approved_how", "delivered_at", "evidence", "evidence_kind", "makegood_for",
"billing_status", "delivery_status", "state", "changes_reason", "work_ref", "submitted_sub")
ITEM = Aggregate(Part(pj_item, ("item_id",)), version="version")
pj_state = Table("pj_state", md,
Column("state_id", String, primary_key=True),
Column("name", String),
Column("position", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pj_event = Table("pj_event", md,
Column("event_id", String, primary_key=True),
Column("subject_id", String),
Column("kind", String),
Column("from_state", String),
Column("to_state", String),
Column("actor", String),
Column("detail", String),
Column("event_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
pj_setting = Table("pj_setting", md,
Column("setting_key", String, primary_key=True),
Column("value", String),
Column("changes", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
def _real_now() -> datetime:
"""Audit stamps use real time (BRICKS.md, Data); business decisions use work.now()."""
return datetime.now(timezone.utc).replace(tzinfo=None)
def _pick(obj, names: tuple) -> dict:
return {n: getattr(obj, n) for n in names}
def _make(cls, row: dict, names: tuple):
return cls(**{n: _plain(row[n]) for n in names})
def _plain(v):
"""The pg wire returns aware datetimes, SQLite naive: compare them as naive UTC (README)."""
if isinstance(v, datetime) and v.tzinfo is not None:
return v.astimezone(timezone.utc).replace(tzinfo=None)
return v
def _defaults(row: dict) -> dict:
"""A NULL integer reads as 0, a NULL text as '' — the dataclass defaults."""
for k in ("blocked_days", "revision", "bp_id"):
if k in row and row[k] is None:
row[k] = 0
for k in ("source_kind", "source_id", "quote_id", "owner", "reconciliation_json", "line_id", "product_id",
"billing_kind", "billing_plan", "title", "blocked_reason", "approved_by", "submitted_by", "approved_how",
"evidence", "evidence_kind", "makegood_for", "changes_reason", "work_ref", "submitted_sub"):
if k in row and row[k] is None:
row[k] = ""
for k, default in (("billing_status", "not_billed"), ("delivery_status", "not_reported")):
if k in row and row[k] is None:
row[k] = default
return row
def _seq(event_id: str) -> tuple:
"""<subject>@<version>#<n> → (version, n), so history reads in the order it was saved."""
try:
version, n = event_id.rsplit("@", 1)[1].split("#")
return int(version), int(n)
except (IndexError, ValueError):
return 0, 0
class _Events:
"""Appends an aggregate's recorded events, keyed <subject>@<version>#<n>."""
def __init__(self, conn):
self.conn = conn
def append(self, agg, version: int) -> None:
now = _real_now()
rows = [{"event_id": f"{e.subject_id}@{version}#{n}", "subject_id": e.subject_id, "kind": e.kind,
"from_state": e.from_state, "to_state": e.to_state, "actor": e.by, "detail": e.detail,
"event_at": now} for n, e in enumerate(agg.events, 1)]
if rows:
upsert(self.conn, pj_event, rows, ("event_id",))
agg.events.clear()
def of(self, subject_id: str) -> list:
stmt = select(pj_event).where(pj_event.c.subject_id == subject_id)
rows = [dict(r._mapping) for r in self.conn.execute(stmt)]
return [{k: _plain(r[k]) for k in ("event_id", "kind", "from_state", "to_state", "actor", "detail", "event_at")}
for r in sorted(rows, key=lambda r: (_plain(r["event_at"]) or datetime.min, _seq(r["event_id"])))]
class ProjectStore:
"""Loads and saves a project whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn, events: _Events):
self.conn, self.events = conn, events
def get(self, project_id: str) -> Optional[tuple]:
"""(Project, the version it was loaded at), or None."""
rows = PROJECT.load(self.conn, {"project_id": project_id})
if rows is None:
return None
return _make(Project, _defaults(dict(rows["pj_project"][0])), PJ_PROJECT_COLUMNS), PROJECT.version_of(rows)
def save(self, agg: Project, expected_version: int) -> int:
"""Writes version expected_version + 1 with the project's new history, or raises Conflict (409)."""
v = PROJECT.save(self.conn, {"pj_project": [_pick(agg, PJ_PROJECT_COLUMNS)]},
expected_version=expected_version)
self.events.append(agg, v)
return v
def all(self) -> list:
"""Every project with its version. The visibility rule filters in the use case."""
out = []
for r in self.conn.execute(select(pj_project)):
row = _defaults(dict(r._mapping))
out.append((_make(Project, row, PJ_PROJECT_COLUMNS), row["version"] or 0))
return sorted(out, key=lambda p: p[0].project_id)
def due_for_reconciliation(self, today) -> list:
stmt = select(pj_project.c.project_id).where(pj_project.c.state == "active", pj_project.c.ends_on < today)
return sorted(r[0] for r in self.conn.execute(stmt))
class ItemStore:
"""Loads and saves an item whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn, events: _Events):
self.conn, self.events = conn, events
def get(self, item_id: str) -> Optional[tuple]:
rows = ITEM.load(self.conn, {"item_id": item_id})
if rows is None:
return None
return _make(Item, _defaults(dict(rows["pj_item"][0])), PJ_ITEM_COLUMNS), ITEM.version_of(rows)
def save(self, agg: Item, expected_version: int) -> int:
v = ITEM.save(self.conn, {"pj_item": [_pick(agg, PJ_ITEM_COLUMNS)]}, expected_version=expected_version)
self.events.append(agg, v)
return v
def _rows(self, stmt) -> list:
out = []
for r in self.conn.execute(stmt):
row = _defaults(dict(r._mapping))
out.append((_make(Item, row, PJ_ITEM_COLUMNS), row["version"] or 0))
return sorted(out, key=lambda p: p[0].item_id)
def of_project(self, project_id: str) -> list:
return self._rows(select(pj_item).where(pj_item.c.project_id == project_id))
def of_source(self, source_kind: str, source_id: str) -> list:
return self._rows(select(pj_item).where(pj_item.c.source_kind == source_kind,
pj_item.c.source_id == source_id))
def all(self) -> list:
return self._rows(select(pj_item))
class StateStore:
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
"""Every item state: the customer's name where they set one, the reference name where not."""
named = {r._mapping["state_id"]: r._mapping for r in self.conn.execute(select(pj_state))}
out = [State(sid, (named.get(sid) or {}).get("name") or name,
pos if (named.get(sid) or {}).get("position") is None else named[sid]["position"])
for sid, name, pos in DEFAULT_STATES]
return sorted(out, key=lambda s: (s.position, s.state_id))
def get(self, state_id: str) -> Optional[State]:
return next((s for s in self.all() if s.state_id == state_id), None)
def save(self, state: State) -> None:
upsert(self.conn, pj_state, [{"state_id": state.state_id, "name": state.name, "position": state.position}],
("state_id",))
def seed(self) -> None:
"""The reference names, for the states the customer has not named yet; never one they have."""
have = {s.state_id for s in self.all()}
for sid, name, pos in DEFAULT_STATES:
if sid not in have:
self.save(State(sid, name, pos))
class SettingsStore:
"""The setting rows (module.yaml config_rows): JSON values, null = the reference behaviour."""
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
rows = {r._mapping["setting_key"]: dict(r._mapping) for r in self.conn.execute(select(pj_setting))}
return [{"setting_key": k, "value": json.loads(rows[k]["value"]) if rows.get(k, {}).get("value") else None,
"changes": SETTING_ROWS[k]} for k in SETTING_ROWS]
def get(self, setting_key: str) -> Optional[dict]:
return next((s for s in self.all() if s["setting_key"] == setting_key), None)
def set(self, setting_key: str, value) -> None:
upsert(self.conn, pj_setting, [{"setting_key": setting_key,
"value": None if value is None else json.dumps(value),
"changes": SETTING_ROWS[setting_key]}], ("setting_key",))
def current(self) -> Settings:
rows = {r._mapping["setting_key"]: r._mapping["value"] for r in self.conn.execute(select(pj_setting))}
return settings_from_rows(rows)
def seed(self) -> None:
"""A row for every setting the customer has not got yet; never touches one they have set."""
have = {r[0] for r in self.conn.execute(select(pj_setting.c.setting_key))}
for key in SETTING_ROWS:
if key not in have:
self.set(key, None)
class Repository:
"""Every repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.events = _Events(conn)
self.projects = ProjectStore(conn, self.events)
self.items = ItemStore(conn, self.events)
self.states = StateStore(conn)
self.settings = SettingsStore(conn)
def seed(self) -> None:
self.settings.seed()
self.states.seed()