dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/infrastructure/tables.py
"""The tables, and one repository per aggregate that loads and saves it whole.
Every table is prefixed, has a natural primary key (never SERIAL) and, for a part, a key that starts with
its root's — the repository finds stale rows under a root by that prefix. Money is NUMERIC(18,2).
schema.sql is generated from the same column model; tests/test_schema.py keeps the two equal.
Generated by `erp-compose scaffold` from module.yaml.
"""
import dataclasses
import json
from datetime import date, datetime, timezone
from typing import Optional
from erp_platform.repository import Aggregate, Part, upsert
from sqlalchemy import TIMESTAMP, Column, Date, Integer, MetaData, Numeric, String, Table, insert, select
from brick_consolidation.domain.ai_finding import AiFinding
from brick_consolidation.domain.cons_journal import ConsJournal, ConsJournalLine
from brick_consolidation.domain.cons_migration_run import ConsMigrationRun
from brick_consolidation.domain.group import (
AccountRole,
AccountRule,
EliminationRule,
Group,
GroupPeriod,
GroupStructure,
)
from brick_consolidation.domain.migration_map import MigrationMap
from brick_consolidation.domain.migration_run import MigrationRun
from brick_consolidation.domain.statement import Statement
from brick_consolidation.domain.structure_change import StructureChange
md = MetaData()
cons_group = Table("cons_group", md,
Column("group_id", String, primary_key=True),
Column("name", String),
Column("group_currency", String),
Column("ledger_id", String),
Column("group_chart_id", String),
Column("top_company_code", String),
Column("fiscal_year_end_month", String),
Column("fsv_id", String),
Column("goodwill_life_years", String),
Column("cash_flow_fsv_id", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_GROUP_COLUMNS = ("group_id", "name", "group_currency", "ledger_id", "group_chart_id", "top_company_code", "fiscal_year_end_month", "fsv_id", "goodwill_life_years", "cash_flow_fsv_id")
cons_group_structure = Table("cons_group_structure", md,
Column("group_id", String, primary_key=True),
Column("company_code", String, primary_key=True),
Column("valid_from_period", String, primary_key=True),
Column("investor_company_code", String),
Column("ownership_pct", Numeric(9, 6)),
Column("voting_pct", Numeric(9, 6)),
Column("method", String),
Column("change_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("source", String), # v3: gl | upload (a company kept outside gl)
Column("local_currency", String), # v3: for source upload, the currency its upload is in
Column("name", String)) # v3: for source upload, its name
CONS_GROUP_STRUCTURE_COLUMNS = ("group_id", "company_code", "valid_from_period", "investor_company_code", "ownership_pct", "voting_pct", "method", "change_id", "source", "local_currency", "name")
cons_account_rule = Table("cons_account_rule", md,
Column("group_id", String, primary_key=True),
Column("group_account_id", String, primary_key=True),
Column("valid_from_period", String, primary_key=True),
Column("rate_type", String),
Column("ic", String),
Column("equity", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_ACCOUNT_RULE_COLUMNS = ("group_id", "group_account_id", "valid_from_period", "rate_type", "ic", "equity")
cons_account_role = Table("cons_account_role", md,
Column("group_id", String, primary_key=True),
Column("role", String, primary_key=True),
Column("valid_from_period", String, primary_key=True),
Column("group_account_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_ACCOUNT_ROLE_COLUMNS = ("group_id", "role", "valid_from_period", "group_account_id")
cons_elimination_rule = Table("cons_elimination_rule", md,
Column("group_id", String, primary_key=True),
Column("rule_id", String, primary_key=True),
Column("kind", String),
Column("left_accounts", String),
Column("right_accounts", String),
Column("difference_role", String),
Column("tolerance", Numeric(18, 2)),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_ELIMINATION_RULE_COLUMNS = ("group_id", "rule_id", "kind", "left_accounts", "right_accounts", "difference_role", "tolerance")
cons_group_period = Table("cons_group_period", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("run_no", Integer),
Column("stale_from_step", String),
Column("prepared_by", String),
Column("prepared_sub", String),
Column("signed_by", String),
Column("signed_sub", String),
Column("signed_at", TIMESTAMP),
Column("tb_digest", String),
Column("reopened_reason", String),
Column("status", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("preparers", String))
CONS_GROUP_PERIOD_COLUMNS = ("group_id", "period_id", "run_no", "stale_from_step", "prepared_by", "prepared_sub", "signed_by", "signed_sub", "signed_at", "tb_digest", "reopened_reason", "status", "preparers")
cons_reported_data = Table("cons_reported_data", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("company_code", String, primary_key=True),
Column("group_account_id", String, primary_key=True),
Column("partner_company", String, primary_key=True),
Column("movement_type", String, primary_key=True),
Column("segment", String, primary_key=True),
Column("ytd_local", Numeric(18, 2)),
Column("period_local", Numeric(18, 2)),
Column("local_currency", String),
Column("source", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_REPORTED_DATA_COLUMNS = ("group_id", "period_id", "company_code", "group_account_id", "partner_company", "movement_type", "segment", "ytd_local", "period_local", "local_currency", "source")
cons_upload = Table("cons_upload", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("company_code", String, primary_key=True),
Column("upload_no", Integer, primary_key=True),
Column("object_key", String),
Column("sha256", String),
Column("rows", String),
Column("status", String),
Column("submitted_by", String),
Column("submitted_sub", String),
Column("accepted_by", String),
Column("accepted_sub", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_UPLOAD_COLUMNS = ("group_id", "period_id", "company_code", "upload_no", "object_key", "sha256", "rows", "status", "submitted_by", "submitted_sub", "accepted_by", "accepted_sub")
cons_translated_data = Table("cons_translated_data", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("run_no", Integer, primary_key=True),
Column("company_code", String, primary_key=True),
Column("group_account_id", String, primary_key=True),
Column("partner_company", String, primary_key=True),
Column("movement_type", String, primary_key=True),
Column("segment", String, primary_key=True),
Column("ytd_group", Numeric(18, 2)),
Column("rate_type", String),
Column("rate", Numeric(19, 6)),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_TRANSLATED_DATA_COLUMNS = ("group_id", "period_id", "run_no", "company_code", "group_account_id", "partner_company", "movement_type", "segment", "ytd_group", "rate_type", "rate")
cons_ic_inventory = Table("cons_ic_inventory", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("holder_company", String, primary_key=True),
Column("seller_company", String, primary_key=True),
Column("inventory_amount", Numeric(18, 2)),
Column("margin_pct", Numeric(9, 6)),
Column("none_stated", String),
Column("entered_by", String),
Column("entered_sub", String),
Column("tax_rate_pct", Numeric(9, 6)),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_IC_INVENTORY_COLUMNS = ("group_id", "period_id", "holder_company", "seller_company", "inventory_amount", "margin_pct", "none_stated", "entered_by", "entered_sub", "tax_rate_pct")
cons_historical_amount = Table("cons_historical_amount", md,
Column("group_id", String, primary_key=True),
Column("company_code", String, primary_key=True),
Column("group_account_id", String, primary_key=True),
Column("arose_period", String, primary_key=True),
Column("amount_local", Numeric(18, 2)),
Column("amount_group", Numeric(18, 2)),
Column("rate", Numeric(19, 6)),
Column("change_id", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_HISTORICAL_AMOUNT_COLUMNS = ("group_id", "company_code", "group_account_id", "arose_period", "amount_local", "amount_group", "rate", "change_id")
cons_number_series = Table("cons_number_series", md,
Column("group_id", String, primary_key=True),
Column("fiscal_year", String, primary_key=True),
Column("entry_no", Integer, primary_key=True),
Column("cons_journal_id", String),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_NUMBER_SERIES_COLUMNS = ("group_id", "fiscal_year", "entry_no", "cons_journal_id", "claimed_at")
cons_carry_forward = Table("cons_carry_forward", md,
Column("group_id", String, primary_key=True),
Column("fiscal_year", String, primary_key=True),
Column("group_account_id", String, primary_key=True),
Column("company_code", String, primary_key=True),
Column("partner_company", String, primary_key=True),
Column("segment", String, primary_key=True),
Column("opening", String),
Column("derived_at", TIMESTAMP),
Column("source_digest", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_CARRY_FORWARD_COLUMNS = ("group_id", "fiscal_year", "group_account_id", "company_code", "partner_company", "segment", "opening", "derived_at", "source_digest")
cons_run_claim = Table("cons_run_claim", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("task", String, primary_key=True),
Column("run_no", Integer, primary_key=True),
Column("claimed_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_RUN_CLAIM_COLUMNS = ("group_id", "period_id", "task", "run_no", "claimed_at")
cons_collection = Table("cons_collection", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("company_code", String, primary_key=True),
Column("run_no", Integer, primary_key=True),
Column("gl_digest", String),
Column("period_status", String),
Column("local_currency", String),
Column("provisional", String),
Column("collected_at", TIMESTAMP),
Column("source", String),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_COLLECTION_COLUMNS = ("group_id", "period_id", "company_code", "run_no", "gl_digest", "period_status", "local_currency", "provisional", "collected_at", "source")
cons_signoff = Table("cons_signoff", md,
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("seq", Integer, primary_key=True),
Column("action", String),
Column("by_email", String),
Column("by_sub", String),
Column("at", TIMESTAMP),
Column("tb_digest", String),
Column("reason", String),
Column("run_no", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
GROUP = Aggregate(Part(cons_group, ("group_id",)),
Part(cons_group_structure, ("group_id", "company_code", "valid_from_period")),
Part(cons_account_rule, ("group_id", "group_account_id", "valid_from_period")),
Part(cons_account_role, ("group_id", "role", "valid_from_period")),
Part(cons_elimination_rule, ("group_id", "rule_id")),
Part(cons_group_period, ("group_id", "period_id")),
version="version")
cons_structure_change = Table("cons_structure_change", md,
Column("change_id", String, primary_key=True),
Column("kind", String),
Column("company_code", String),
Column("investor_company_code", String),
Column("ownership_pct", Numeric(9, 6)),
Column("voting_pct", Numeric(9, 6)),
Column("method", String),
Column("currency", String),
Column("effective_date", Date),
Column("consideration", Numeric(18, 2)),
Column("proceeds", Numeric(18, 2)),
Column("previously_held_fair_value", Numeric(18, 2)),
Column("net_assets_json", String),
Column("fair_value_adjustments_json", String),
Column("net_assets", Numeric(18, 2)),
Column("goodwill", Numeric(18, 2)),
Column("nci", Numeric(18, 2)),
Column("gain_loss", Numeric(18, 2)),
Column("rate", Numeric(19, 6)),
Column("goodwill_group", Numeric(18, 2)),
Column("nci_group", Numeric(18, 2)),
Column("bargain_purchase", String),
Column("proposed_by", String),
Column("proposed_sub", String),
Column("approved_by", String),
Column("approved_sub", String),
Column("status", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("figures_json", String),
Column("source", String), # v3: gl | upload
Column("name", String)) # v3: for source upload
CONS_STRUCTURE_CHANGE_COLUMNS = ("change_id", "kind", "company_code", "investor_company_code", "ownership_pct", "voting_pct", "method", "currency", "effective_date", "consideration", "proceeds", "previously_held_fair_value", "net_assets_json", "fair_value_adjustments_json", "net_assets", "goodwill", "nci", "gain_loss", "rate", "goodwill_group", "nci_group", "bargain_purchase", "proposed_by", "proposed_sub", "approved_by", "approved_sub", "status", "figures_json", "source", "name")
STRUCTURE_CHANGE = Aggregate(Part(cons_structure_change, ("change_id",)),
version="version")
cons_journal = Table("cons_journal", md,
Column("cons_journal_id", String, primary_key=True),
Column("group_id", String),
Column("period_id", String),
Column("document_type", String),
Column("auto", String),
Column("task", String),
Column("run_no", Integer),
Column("period_specific", String),
Column("entry_no", Integer),
Column("proposed_by", String),
Column("proposed_sub", String),
Column("approved_by", String),
Column("approved_sub", String),
Column("reverses_id", String),
Column("memo", String),
Column("migration_run_id", String),
Column("status", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_JOURNAL_COLUMNS = ("cons_journal_id", "group_id", "period_id", "document_type", "auto", "task", "run_no", "period_specific", "entry_no", "proposed_by", "proposed_sub", "approved_by", "approved_sub", "reverses_id", "memo", "migration_run_id", "status")
cons_journal_line = Table("cons_journal_line", md,
Column("cons_journal_id", String, primary_key=True),
Column("line_no", String, primary_key=True),
Column("group_account_id", String),
Column("company_code", String),
Column("partner_company", String),
Column("movement_type", String),
Column("segment", String),
Column("amount", Numeric(18, 2)),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_JOURNAL_LINE_COLUMNS = ("cons_journal_id", "line_no", "group_account_id", "company_code", "partner_company", "movement_type", "segment", "amount")
CONS_JOURNAL = Aggregate(Part(cons_journal, ("cons_journal_id",)),
Part(cons_journal_line, ("cons_journal_id", "line_no")),
version="version")
cons_statement = Table("cons_statement", md,
Column("report", String, primary_key=True),
Column("group_id", String, primary_key=True),
Column("period_id", String, primary_key=True),
Column("node_id", String, primary_key=True),
Column("amount", Numeric(18, 2)),
Column("prior_year", String),
Column("produced_at", TIMESTAMP),
Column("digest", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_STATEMENT_COLUMNS = ("report", "group_id", "period_id", "node_id", "amount", "prior_year", "produced_at", "digest")
STATEMENT = Aggregate(Part(cons_statement, ("report", "group_id", "period_id", "node_id")),
version="version")
cons_ai_finding = Table("cons_ai_finding", md,
Column("finding_id", String, primary_key=True),
Column("group_id", String),
Column("period_id", String),
Column("pair", String),
Column("model", String),
Column("inputs_digest", String),
Column("finding", String),
Column("evidence", String),
Column("decision", String),
Column("decided_by", String),
Column("decided_at", TIMESTAMP),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_AI_FINDING_COLUMNS = ("finding_id", "group_id", "period_id", "pair", "model", "inputs_digest", "finding", "evidence", "decision", "decided_by", "decided_at")
AI_FINDING = Aggregate(Part(cons_ai_finding, ("finding_id",)),
version="version")
cons_migration_run = Table("cons_migration_run", md,
Column("run_id", String, primary_key=True),
Column("status", String),
Column("doors_json", String),
Column("checks_json", String),
Column("signed_by", String),
Column("signed_at", TIMESTAMP),
Column("checksums", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_MIGRATION_RUN_COLUMNS = ("run_id", "status", "doors_json", "checks_json", "signed_by", "signed_at", "checksums")
MIGRATION_RUN = Aggregate(Part(cons_migration_run, ("run_id",)),
version="version")
cons_migration_map = Table("cons_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("natural_key", String),
Column("run_id", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_MIGRATION_MAP_COLUMNS = ("odoo_db", "odoo_model", "odoo_id", "natural_key", "run_id")
MIGRATION_MAP = Aggregate(Part(cons_migration_map, ("odoo_db", "odoo_model", "odoo_id")),
version="version")
cons_cons_migration_runs = Table("cons_cons_migration_runs", md,
Column("cons_migration_run_id", String, primary_key=True),
Column("status", String),
Column("version", Integer),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP),
Column("doors_json", String),
Column("checks_json", String),
Column("signed_by", String),
Column("signed_sub", String),
Column("signed_at", TIMESTAMP),
Column("checksums", String))
CONS_CONS_MIGRATION_RUNS_COLUMNS = ("cons_migration_run_id", "status", "doors_json", "checks_json", "signed_by", "signed_sub", "signed_at", "checksums")
CONS_MIGRATION_RUN = Aggregate(Part(cons_cons_migration_runs, ("cons_migration_run_id",)),
version="version")
cons_audit = Table("cons_audit", md,
Column("audit_id", String, primary_key=True),
Column("group_id", String),
Column("period_id", String),
Column("subject", String),
Column("event", String),
Column("sub", String),
Column("event_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
CONS_AUDIT_COLUMNS = ("audit_id", "group_id", "period_id", "subject", "event", "sub", "event_at")
def _append_history(conn, root: str, key: dict, state, events: list, version: int) -> None:
"""One cons_audit row per event a save recorded — append-only: a plain INSERT, its key derived from
the aggregate, its version and the event's place, so one save never appends twice. A column the event
does not carry is None: TODO(rule) fill it where the brick knows it (e.g. from_state)."""
if not events:
return
ref = "|".join(str(v) for v in key.values())
now = datetime.now(timezone.utc).replace(tzinfo=None)
rows = []
for n, event in enumerate(events, 1):
data = dataclasses.asdict(event)
rows.append({
"audit_id": f"{root}:{ref}:v{version}:{n}",
"group_id": key.get("group_id", data.get("group_id")),
"period_id": key.get("period_id", data.get("period_id")),
"subject": ref,
"event": type(event).__name__,
"sub": key.get("sub", data.get("sub")),
"event_at": key.get("event_at", data.get("event_at")),
"created_at": now, "updated_at": now})
conn.execute(insert(cons_audit), rows)
cons_settings = Table("cons_settings", md,
Column("setting_key", String, primary_key=True),
Column("value", String),
Column("changes", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
# CON-19: every setting is also a DATED row (key, scope, valid_from): rows are only ever added, the one in force on the business date is the
# value, the history is the table. cons_settings (above) keeps the setting's kind and its reference behaviour.
cons_setting_row = Table("cons_setting_row", md,
Column("setting_key", String, primary_key=True),
Column("scope", String, primary_key=True),
Column("valid_from", Date, primary_key=True),
Column("value", String),
Column("set_by", String),
Column("set_sub", String),
Column("set_at", TIMESTAMP),
Column("created_at", TIMESTAMP),
Column("updated_at", TIMESTAMP))
# module.yaml config_rows: ONE ROW PER SETTING, typed (key, kind, what it changes, a mode's choices).
# The value is stored as JSON text; NULL = the reference behaviour. TODO(rule): each setting's default.
SETTINGS_SEED = (
("signoff_second_person", "bool", "whether manual entries, uploads, structure changes and the sign-off need a second person; cons:settings:controls", ()),
("allow_open_period_collect", "bool", "collect from a company whose gl period is still open (flagged provisional); cons:settings:controls", ()),
("translation_pl_rate", "str", "monthly_average | ytd_average: how income and expense are translated (IAS 21.40 permits an average that approximates the transaction rates)", ()),
("nci_measurement", "str", "proportionate share of net assets (IFRS 3.19); full goodwill is an extension", ()),
("ai.explain_elimination_difference.mode", "mode", "a pair whose intercompany balances differ comes with a likely cause (an invoice in one month and the bill in the next, a different rate, a missing recharge): off or suggest", ("off", "suggest", )),
)
SETTING_KINDS = {k: (kind, choices) for k, kind, _, choices in SETTINGS_SEED}
def _encode(kind: str, choices: tuple, value) -> Optional[str]:
if value is None:
return None
if kind == "bool" and not isinstance(value, bool):
raise ValueError("true or false")
if kind == "int" and (isinstance(value, bool) or not isinstance(value, int)):
raise ValueError("a whole number")
if kind == "date":
value = date.fromisoformat(str(value)).isoformat()
if kind == "mode" and value not in choices:
raise ValueError("one of " + ", ".join(choices))
if kind in ("str", "mode") and not isinstance(value, str):
raise ValueError("text")
return json.dumps(value)
def _decode(raw: Optional[str]):
return None if raw is None else json.loads(raw)
def _pick(obj, names: tuple) -> dict:
return {n: getattr(obj, n) for n in names}
_V3_DEFAULTS = {"source": "gl", "local_currency": "", "name": ""} # v3 columns: a row written before them reads as source gl
def _make(cls, row: dict, names: tuple):
return cls(**{n: (row[n] if row[n] is not None or n not in _V3_DEFAULTS else _V3_DEFAULTS[n]) for n in names})
def _sorted(rows: list, key: tuple) -> list:
return sorted(rows, key=lambda r: tuple(str(r[k]) for k in key))
class GroupRepository:
"""Loads and saves a group whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, group_id) -> Optional[tuple]:
"""(Group, the version it was loaded at), or None."""
rows = GROUP.load(self.conn, {"group_id": group_id})
if rows is None:
return None
agg = _make(Group, rows["cons_group"][0], CONS_GROUP_COLUMNS)
agg.group_structures = [_make(GroupStructure, x, CONS_GROUP_STRUCTURE_COLUMNS)
for x in _sorted(rows["cons_group_structure"], CONS_GROUP_STRUCTURE_COLUMNS[:3])]
agg.account_rules = [_make(AccountRule, x, CONS_ACCOUNT_RULE_COLUMNS)
for x in _sorted(rows["cons_account_rule"], CONS_ACCOUNT_RULE_COLUMNS[:3])]
agg.account_roles = [_make(AccountRole, x, CONS_ACCOUNT_ROLE_COLUMNS)
for x in _sorted(rows["cons_account_role"], CONS_ACCOUNT_ROLE_COLUMNS[:3])]
agg.elimination_rules = [_make(EliminationRule, x, CONS_ELIMINATION_RULE_COLUMNS)
for x in _sorted(rows["cons_elimination_rule"], CONS_ELIMINATION_RULE_COLUMNS[:2])]
agg.group_periods = [_make(GroupPeriod, x, CONS_GROUP_PERIOD_COLUMNS)
for x in _sorted(rows["cons_group_period"], CONS_GROUP_PERIOD_COLUMNS[:2])]
agg._snapshot = rows.snapshot # erp-platform 0.9.6: a save writes only what changed since this load
return agg, GROUP.version_of(rows)
def save(self, agg: Group, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = GROUP.save(self.conn, {
"cons_group": [_pick(agg, CONS_GROUP_COLUMNS)],
"cons_group_structure": [_pick(x, CONS_GROUP_STRUCTURE_COLUMNS) for x in agg.group_structures],
"cons_account_rule": [_pick(x, CONS_ACCOUNT_RULE_COLUMNS) for x in agg.account_rules],
"cons_account_role": [_pick(x, CONS_ACCOUNT_ROLE_COLUMNS) for x in agg.account_roles],
"cons_elimination_rule": [_pick(x, CONS_ELIMINATION_RULE_COLUMNS) for x in agg.elimination_rules],
"cons_group_period": [_pick(x, CONS_GROUP_PERIOD_COLUMNS) for x in agg.group_periods]},
expected_version=expected_version, snapshot=getattr(agg, "_snapshot", None))
agg._snapshot = None
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "group", {"group_id": agg.group_id}, None, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_group.c.group_id)
return sorted(row[0] for row in self.conn.execute(stmt))
class StructureChangeRepository:
"""Loads and saves a structure change whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, change_id) -> Optional[tuple]:
"""(StructureChange, the version it was loaded at), or None."""
rows = STRUCTURE_CHANGE.load(self.conn, {"change_id": change_id})
if rows is None:
return None
agg = _make(StructureChange, rows["cons_structure_change"][0], CONS_STRUCTURE_CHANGE_COLUMNS)
return agg, STRUCTURE_CHANGE.version_of(rows)
def save(self, agg: StructureChange, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = STRUCTURE_CHANGE.save(self.conn, {
"cons_structure_change": [_pick(agg, CONS_STRUCTURE_CHANGE_COLUMNS)]},
expected_version=expected_version)
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "structure_change", {"change_id": agg.change_id}, agg.status, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_structure_change.c.change_id)
return sorted(row[0] for row in self.conn.execute(stmt))
class ConsJournalRepository:
"""Loads and saves a cons journal whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, cons_journal_id) -> Optional[tuple]:
"""(ConsJournal, the version it was loaded at), or None."""
rows = CONS_JOURNAL.load(self.conn, {"cons_journal_id": cons_journal_id})
if rows is None:
return None
agg = _make(ConsJournal, rows["cons_journal"][0], CONS_JOURNAL_COLUMNS)
agg.cons_journal_lines = [_make(ConsJournalLine, x, CONS_JOURNAL_LINE_COLUMNS)
for x in _sorted(rows["cons_journal_line"], CONS_JOURNAL_LINE_COLUMNS[:2])]
return agg, CONS_JOURNAL.version_of(rows)
def save(self, agg: ConsJournal, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = CONS_JOURNAL.save(self.conn, {
"cons_journal": [_pick(agg, CONS_JOURNAL_COLUMNS)],
"cons_journal_line": [_pick(x, CONS_JOURNAL_LINE_COLUMNS) for x in agg.cons_journal_lines]},
expected_version=expected_version)
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "cons_journal", {"cons_journal_id": agg.cons_journal_id}, agg.status, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_journal.c.cons_journal_id)
return sorted(row[0] for row in self.conn.execute(stmt))
class StatementRepository:
"""Loads and saves a statement whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, report, group_id, period_id, node_id) -> Optional[tuple]:
"""(Statement, the version it was loaded at), or None."""
rows = STATEMENT.load(self.conn, {"report": report, "group_id": group_id, "period_id": period_id, "node_id": node_id})
if rows is None:
return None
agg = _make(Statement, rows["cons_statement"][0], CONS_STATEMENT_COLUMNS)
return agg, STATEMENT.version_of(rows)
def save(self, agg: Statement, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = STATEMENT.save(self.conn, {
"cons_statement": [_pick(agg, CONS_STATEMENT_COLUMNS)]},
expected_version=expected_version)
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "statement", {"report": agg.report, "group_id": agg.group_id, "period_id": agg.period_id, "node_id": agg.node_id}, None, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_statement.c.report, cons_statement.c.group_id, cons_statement.c.period_id, cons_statement.c.node_id)
return sorted(list(row) for row in self.conn.execute(stmt))
class AiFindingRepository:
"""Loads and saves a ai finding whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, finding_id) -> Optional[tuple]:
"""(AiFinding, the version it was loaded at), or None."""
rows = AI_FINDING.load(self.conn, {"finding_id": finding_id})
if rows is None:
return None
agg = _make(AiFinding, rows["cons_ai_finding"][0], CONS_AI_FINDING_COLUMNS)
return agg, AI_FINDING.version_of(rows)
def save(self, agg: AiFinding, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = AI_FINDING.save(self.conn, {
"cons_ai_finding": [_pick(agg, CONS_AI_FINDING_COLUMNS)]},
expected_version=expected_version)
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "ai_finding", {"finding_id": agg.finding_id}, None, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_ai_finding.c.finding_id)
return sorted(row[0] for row in self.conn.execute(stmt))
class MigrationRunRepository:
"""Loads and saves a migration run whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, run_id) -> Optional[tuple]:
"""(MigrationRun, the version it was loaded at), or None."""
rows = MIGRATION_RUN.load(self.conn, {"run_id": run_id})
if rows is None:
return None
agg = _make(MigrationRun, rows["cons_migration_run"][0], CONS_MIGRATION_RUN_COLUMNS)
return agg, MIGRATION_RUN.version_of(rows)
def save(self, agg: MigrationRun, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = MIGRATION_RUN.save(self.conn, {
"cons_migration_run": [_pick(agg, CONS_MIGRATION_RUN_COLUMNS)]},
expected_version=expected_version)
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "migration_run", {"run_id": agg.run_id}, None, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_migration_run.c.run_id)
return sorted(row[0] for row in self.conn.execute(stmt))
class MigrationMapRepository:
"""Loads and saves a migration map whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, odoo_db, odoo_model, odoo_id) -> Optional[tuple]:
"""(MigrationMap, the version it was loaded at), or None."""
rows = MIGRATION_MAP.load(self.conn, {"odoo_db": odoo_db, "odoo_model": odoo_model, "odoo_id": odoo_id})
if rows is None:
return None
agg = _make(MigrationMap, rows["cons_migration_map"][0], CONS_MIGRATION_MAP_COLUMNS)
return agg, MIGRATION_MAP.version_of(rows)
def save(self, agg: MigrationMap, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = MIGRATION_MAP.save(self.conn, {
"cons_migration_map": [_pick(agg, CONS_MIGRATION_MAP_COLUMNS)]},
expected_version=expected_version)
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "migration_map", {"odoo_db": agg.odoo_db, "odoo_model": agg.odoo_model, "odoo_id": agg.odoo_id}, None, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_migration_map.c.odoo_db, cons_migration_map.c.odoo_model, cons_migration_map.c.odoo_id)
return sorted(list(row) for row in self.conn.execute(stmt))
class ConsMigrationRunRepository:
"""Loads and saves a cons migration run whole. Takes an open connection: the use case owns the transaction."""
def __init__(self, conn):
self.conn = conn
def get(self, cons_migration_run_id) -> Optional[tuple]:
"""(ConsMigrationRun, the version it was loaded at), or None."""
rows = CONS_MIGRATION_RUN.load(self.conn, {"cons_migration_run_id": cons_migration_run_id})
if rows is None:
return None
agg = _make(ConsMigrationRun, rows["cons_cons_migration_runs"][0], CONS_CONS_MIGRATION_RUNS_COLUMNS)
if agg.signed_at is not None and agg.signed_at.tzinfo is None:
agg.signed_at = agg.signed_at.replace(tzinfo=timezone.utc) # stored as UTC without a zone; a moment is aware
return agg, CONS_MIGRATION_RUN.version_of(rows)
def save(self, agg: ConsMigrationRun, expected_version: int) -> int:
"""Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""
version = CONS_MIGRATION_RUN.save(self.conn, {
"cons_cons_migration_runs": [_pick(agg, CONS_CONS_MIGRATION_RUNS_COLUMNS)]},
expected_version=expected_version)
# append-only history (cons_audit): one row per event this save records, never rewritten
_append_history(self.conn, "cons_migration_run", {"cons_migration_run_id": agg.cons_migration_run_id}, agg.status, agg.events, version or expected_version)
agg.events.clear()
return version
def ids(self, owner: Optional[str] = None) -> list:
stmt = select(cons_cons_migration_runs.c.cons_migration_run_id)
return sorted(row[0] for row in self.conn.execute(stmt))
class AuditRepository:
"""Append-only history (cons_audit): what happened in the group close and to each entry (append only). Written by the aggregates' saves
(`_append_history`), never by a use case, never updated."""
def __init__(self, conn):
self.conn = conn
def all(self) -> list:
stmt = select(cons_audit).order_by(cons_audit.c.audit_id)
return [{k: r._mapping[k] for k in CONS_AUDIT_COLUMNS} for r in self.conn.execute(stmt)]
class SettingsRepository:
"""The setting rows. cons_settings holds each setting's kind and its reference behaviour (one row per setting); cons_setting_row holds the
dated values (CON-19): rows are only ever added, the one in force on the business date is the value, the table is the history."""
def __init__(self, conn):
self.conn = conn
def _base(self) -> list:
stmt = select(cons_settings).order_by(cons_settings.c.setting_key)
return [{"setting_key": r._mapping["setting_key"], "value": _decode(r._mapping["value"]),
"kind": SETTING_KINDS.get(r._mapping["setting_key"], ("str", ()))[0],
"changes": r._mapping["changes"]} for r in self.conn.execute(stmt)]
def dated(self) -> list:
t = cons_setting_row
return [dict(r) | {"value": _decode(r["value"])} for r in self.conn.execute(select(t).order_by(t.c.setting_key, t.c.valid_from)).mappings().all()]
def all(self, today: Optional[date] = None) -> list:
"""Every setting with the value in force on `today` (a dated row, else the reference row's value; None = the reference behaviour)."""
out = self._base()
if today is None:
return out
rows = self.dated()
for s in out:
mine = [r for r in rows if r["setting_key"] == s["setting_key"] and r["valid_from"] <= today]
if mine:
s["value"] = max(mine, key=lambda r: r["valid_from"])["value"]
return out
def get(self, setting_key: str) -> Optional[dict]:
return next((s for s in self._base() if s["setting_key"] == setting_key), None)
def history(self, setting_key: str) -> list:
t = cons_setting_row
stmt = select(t).where(t.c.setting_key == setting_key).order_by(t.c.valid_from)
return [{"setting_key": r["setting_key"], "scope": r["scope"], "valid_from": r["valid_from"], "value": _decode(r["value"]), "set_by": r["set_by"],
"set_sub": r["set_sub"], "set_at": r["set_at"]} for r in self.conn.execute(stmt).mappings().all()]
def add(self, setting_key: str, valid_from: date, value, set_by: str, set_sub: str, set_at, scope: str = "") -> None:
"""Appends a dated row (a plain INSERT: a second row with the same key, scope and date is a 409, rows are never edited). Raises ValueError
when `value` is not of the setting's kind."""
kind, choices = SETTING_KINDS.get(setting_key, ("str", ()))
now = datetime.now(timezone.utc).replace(tzinfo=None)
self.conn.execute(insert(cons_setting_row).values(setting_key=setting_key, scope=scope, valid_from=valid_from, value=_encode(kind, choices, value),
set_by=set_by, set_sub=set_sub, set_at=set_at.astimezone(timezone.utc).replace(tzinfo=None),
created_at=now, updated_at=now))
def set(self, setting_key: str, value, changes: str) -> None:
"""The reference row of a setting (its kind and what it changes; seeding). Raises ValueError when `value` is not of the setting's kind
(None puts the reference back)."""
kind, choices = SETTING_KINDS.get(setting_key, ("str", ()))
upsert(self.conn, cons_settings, [{"setting_key": setting_key, "value": _encode(kind, choices, value),
"changes": changes}], ("setting_key",))
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(cons_settings.c.setting_key))}
for key, _kind, changes, _choices in SETTINGS_SEED:
if key not in have:
self.set(key, None, changes)
class Repository:
"""Every aggregate's repository on one connection: what `use_cases.repository()` yields."""
def __init__(self, conn):
self.conn = conn
self.groups = GroupRepository(conn)
self.structure_changes = StructureChangeRepository(conn)
self.cons_journals = ConsJournalRepository(conn)
self.statements = StatementRepository(conn)
self.ai_findings = AiFindingRepository(conn)
self.migration_runs = MigrationRunRepository(conn)
self.migration_maps = MigrationMapRepository(conn)
self.cons_migration_runs = ConsMigrationRunRepository(conn)
self.audits = AuditRepository(conn)
self.settings = SettingsRepository(conn)
from brick_consolidation.infrastructure.data import ConsData, MigrationData
self.data = ConsData(conn)
self.migration = MigrationData(conn)
def seed(self) -> None:
"""The rows a new bucket starts with: the settings."""
self.settings.seed()