dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/infrastructure/data.py
"""The bulk rows of a close — reported data, translated data, historical amounts, intercompany stock, the automatic entries.
These never load as part of the Group aggregate: a group's year of reported data is thousands of rows, and a use case needs the
period's rows for ALL its companies at once. Every read here is ONE statement whatever the number of companies (an IN-list costs
what one key costs); every write is one executemany per table. The group's version claim (cons_group_versions) is the ONE key two
different operations of the same group collide on; the run claims below (cons_run_claim) are the second, named per task and run.
"""
from datetime import datetime, timezone
from decimal import Decimal
from erp_platform.repository import claim_tables, upsert
from sqlalchemy import func, insert, or_, select, update
from brick_consolidation.infrastructure.tables import (
CONS_JOURNAL,
cons_audit,
cons_carry_forward,
cons_collection,
cons_historical_amount,
cons_ic_inventory,
cons_journal,
cons_journal_line,
cons_migration_map,
cons_number_series,
cons_reported_data,
cons_run_claim,
cons_signoff,
cons_statement,
cons_structure_change,
cons_translated_data,
cons_upload,
)
claim_tables("cons_run_claim", "cons_number_series", "cons_upload", "cons_signoff", "cons_setting_row") # a duplicate key here is a 409: the same task of the same run, the same number, upload or sign-off twice
REPORTED_KEY = ("group_id", "period_id", "company_code", "group_account_id", "partner_company", "movement_type", "segment")
TRANSLATION_DOCS = ("translation",)
_CHANGE_COLUMNS = tuple(getattr(cons_structure_change.c, n) for n in (
"change_id", "company_code", "kind", "status", "investor_company_code", "ownership_pct", "effective_date", "net_assets", "goodwill", "nci",
"goodwill_group", "nci_group", "rate", "bargain_purchase", "fair_value_adjustments_json", "figures_json", "consideration", "proceeds",
"previously_held_fair_value", "gain_loss", "method", "currency"))
def _now() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None)
def _stamp(rows: list) -> list:
now = _now()
return [{**r, "created_at": now, "updated_at": now} for r in rows]
class ConsData:
def __init__(self, conn):
self.conn = conn
# ── claims ────────────────────────────────────────────────────────────────────────────────────────────────
def claim(self, group_id: str, period_id: str, task: str, run_no: int) -> None:
"""A plain INSERT: the second operation of the same task and run fails 23505, which claim_tables makes a 409."""
self.conn.execute(insert(cons_run_claim).values(group_id=group_id, period_id=period_id, task=task, run_no=run_no,
claimed_at=_now(), created_at=_now(), updated_at=_now()))
# ── collection ────────────────────────────────────────────────────────────────────────────────────────────
def collections(self, group_id: str, period_id: str) -> list:
t = cons_collection
rows = self.conn.execute(select(t).where(t.c.group_id == group_id, t.c.period_id == period_id)).mappings().all()
return [dict(r) for r in rows]
def add_collections(self, rows: list) -> None:
if rows:
self.conn.execute(insert(cons_collection), _stamp(rows))
def write_collections(self, rows: list) -> None:
upsert(self.conn, cons_collection, _stamp(rows), ("group_id", "period_id", "company_code", "run_no"))
def reported(self, group_id: str, periods: list, companies: list) -> list:
t = cons_reported_data
stmt = select(t).where(t.c.group_id == group_id, t.c.period_id.in_(periods), t.c.company_code.in_(companies))
return [dict(r) for r in self.conn.execute(stmt).mappings().all()]
def write_reported(self, rows: list) -> None:
upsert(self.conn, cons_reported_data, rows, REPORTED_KEY)
# ── translation ───────────────────────────────────────────────────────────────────────────────────────────
def translated(self, group_id: str, period_id: str, run_no: int) -> list:
t = cons_translated_data
stmt = select(t).where(t.c.group_id == group_id, t.c.period_id == period_id, t.c.run_no == run_no)
return [dict(r) for r in self.conn.execute(stmt).mappings().all()]
def add_translated(self, rows: list) -> None:
if rows:
self.conn.execute(insert(cons_translated_data), _stamp(rows))
def historical(self, group_id: str, companies: list) -> list:
t = cons_historical_amount
return [dict(r) for r in self.conn.execute(select(t).where(t.c.group_id == group_id, t.c.company_code.in_(companies))).mappings().all()]
def write_historical(self, rows: list) -> None:
upsert(self.conn, cons_historical_amount, rows, ("group_id", "company_code", "group_account_id", "arose_period"))
# ── intercompany stock ────────────────────────────────────────────────────────────────────────────────────
def inventory(self, group_id: str, period_id: str) -> list:
t = cons_ic_inventory
return [dict(r) for r in self.conn.execute(select(t).where(t.c.group_id == group_id, t.c.period_id == period_id)).mappings().all()]
def write_inventory(self, row: dict) -> None:
upsert(self.conn, cons_ic_inventory, [row], ("group_id", "period_id", "holder_company", "seller_company"))
# ── structure changes ─────────────────────────────────────────────────────────────────────────────────────
def changes(self, group_id: str, companies: list) -> list:
"""The structure changes of these companies in this group (change_id starts with the group's id): one statement."""
t = cons_structure_change
stmt = select(*_CHANGE_COLUMNS).where(t.c.change_id.like(f"{group_id}:%"), t.c.company_code.in_(companies))
return [dict(r) for r in self.conn.execute(stmt).mappings().all()]
def effective_changes(self, group_id: str, kinds: tuple = ()) -> list:
"""Every effective change of the group (of these kinds, when given), whatever the company (a company that has left the structure is not
in any period's members): one statement."""
t = cons_structure_change
stmt = select(*_CHANGE_COLUMNS).where(t.c.change_id.like(f"{group_id}:%"), t.c.status == "effective")
if kinds:
stmt = stmt.where(t.c.kind.in_(kinds))
return [dict(r) for r in self.conn.execute(stmt.order_by(t.c.effective_date, t.c.change_id)).mappings().all()]
# ── the automatic entries ─────────────────────────────────────────────────────────────────────────────────
def posted_entries(self, group_id: str, period_id: str, documents: tuple, below_run: int) -> list:
"""The posted automatic entries of earlier runs for these document types, each with its lines: two statements."""
j, ln = cons_journal, cons_journal_line
heads = [dict(r) for r in self.conn.execute(select(j).where(
j.c.group_id == group_id, j.c.period_id == period_id, j.c.auto == "true", j.c.status == "posted",
j.c.document_type.in_(documents), j.c.run_no < below_run)).mappings().all()]
if not heads:
return []
by_id = {h["cons_journal_id"]: {**h, "lines": []} for h in heads}
for r in self.conn.execute(select(ln).where(ln.c.cons_journal_id.in_(list(by_id)))).mappings().all():
by_id[r["cons_journal_id"]]["lines"].append(dict(r))
return list(by_id.values())
def entries(self, group_id: str, period_id: str, run_no: int) -> list:
"""Every posted or reversed entry of the period (this run's and the mirrors) with its lines: the consolidated books."""
j, ln = cons_journal, cons_journal_line
heads = [dict(r) for r in self.conn.execute(select(j).where(
j.c.group_id == group_id, j.c.period_id == period_id, j.c.status.in_(("posted", "reversed")))).mappings().all()]
by_id = {h["cons_journal_id"]: {**h, "lines": []} for h in heads}
if by_id:
for r in self.conn.execute(select(ln).where(ln.c.cons_journal_id.in_(list(by_id)))).mappings().all():
by_id[r["cons_journal_id"]]["lines"].append(dict(r))
return sorted(by_id.values(), key=lambda e: e["cons_journal_id"])
def claim_numbers(self, group_id: str, fiscal_year: str, journal_ids: list) -> list:
"""The next entry numbers of the group's book and year, ABOVE the high-water mark and never filling a gap, claimed by a plain
INSERT keyed on the number (a racer's duplicate is a 23505 -> 409). Two statements whatever the count."""
if not journal_ids:
return []
t = cons_number_series
top = self.conn.execute(select(func.max(t.c.entry_no)).where(t.c.group_id == group_id, t.c.fiscal_year == fiscal_year)).scalar() or 0
now = _now()
numbers = list(range(top + 1, top + 1 + len(journal_ids)))
self.conn.execute(insert(t), [{"group_id": group_id, "fiscal_year": fiscal_year, "entry_no": n, "cons_journal_id": j, "claimed_at": now,
"created_at": now, "updated_at": now} for n, j in zip(numbers, journal_ids)])
return numbers
def post_entries(self, entries: list, fiscal_year: str = "") -> None:
"""Bulk-post automatic entries (status posted, version 1), each numbered in the group's book and year. A key that already
exists is a 23505 -> 409: the replay."""
if not entries:
return
now = _now()
numbers = self.claim_numbers(entries[0]["group_id"], fiscal_year, [e["cons_journal_id"] for e in entries]) if fiscal_year else [0] * len(entries)
heads, lines, versions, audits = [], [], [], []
for e, number in zip(entries, numbers):
heads.append({**{k: v for k, v in e.items() if k != "lines"}, "status": "posted", "version": 1, "auto": "true", "entry_no": number,
"created_at": now, "updated_at": now})
for n, ln in enumerate(e["lines"], 1):
lines.append({"cons_journal_id": e["cons_journal_id"], "line_no": n, **{k: ln[k] for k in (
"group_account_id", "company_code", "partner_company", "movement_type", "segment", "amount")},
"created_at": now, "updated_at": now})
versions.append({"cons_journal_id": e["cons_journal_id"], "version": 1, "saved_at": now})
audits.append({"audit_id": f"cons_journal:{e['cons_journal_id']}:v1:1", "group_id": e["group_id"], "period_id": e["period_id"],
"subject": e["cons_journal_id"], "event": "ConsJournalPosted", "sub": "", "event_at": now,
"created_at": now, "updated_at": now})
self.conn.execute(insert(cons_journal), heads)
self.conn.execute(insert(cons_journal_line), lines)
self.conn.execute(insert(CONS_JOURNAL.versions), versions)
self.conn.execute(insert(cons_audit), audits)
def books_entries(self, group_id: str, period_id: str, carried: list) -> list:
"""The consolidated books' entries: this period's (posted, and reversed with their mirrors) and the cumulative manual entries of
the year's earlier periods (a period-specific entry is reversed at the next period's start: it is not carried). Two statements."""
j, ln = cons_journal, cons_journal_line
here = j.c.period_id == period_id
if carried:
here = or_(here, (j.c.period_id.in_(carried)) & (j.c.document_type == "manual") & (j.c.period_specific != "true"))
heads = [dict(r) for r in self.conn.execute(select(j).where(j.c.group_id == group_id, j.c.status.in_(("posted", "reversed")), here))
.mappings().all()]
by_id = {h["cons_journal_id"]: {**h, "lines": []} for h in heads}
if by_id:
for r in self.conn.execute(select(ln).where(ln.c.cons_journal_id.in_(list(by_id)))).mappings().all():
by_id[r["cons_journal_id"]]["lines"].append(dict(r))
return sorted(by_id.values(), key=lambda e: (e["period_id"], e["entry_no"] or 0, e["cons_journal_id"]))
def waiting(self, group_id: str, period_id: str) -> list:
"""The manual entries proposed and not yet decided: one statement."""
j = cons_journal
return sorted(r[0] for r in self.conn.execute(select(j.c.cons_journal_id).where(
j.c.group_id == group_id, j.c.period_id == period_id, j.c.status == "proposed")))
# ── uploads ───────────────────────────────────────────────────────────────────────────────────────────────
def uploads(self, group_id: str, period_id: str, company: str = "") -> list:
t = cons_upload
stmt = select(t).where(t.c.group_id == group_id, t.c.period_id == period_id)
if company:
stmt = stmt.where(t.c.company_code == company)
return [dict(r) for r in self.conn.execute(stmt.order_by(t.c.company_code, t.c.upload_no)).mappings().all()]
def add_upload(self, row: dict) -> None:
"""The next upload number of the company and period, claimed by a plain INSERT (a racer is a 409)."""
now = _now()
self.conn.execute(insert(cons_upload), [{**row, "created_at": now, "updated_at": now}])
def set_upload_status(self, group_id: str, period_id: str, company: str, numbers: list, status: str, **more) -> None:
if numbers:
t = cons_upload
self.conn.execute(update(t).where(t.c.group_id == group_id, t.c.period_id == period_id, t.c.company_code == company,
t.c.upload_no.in_(numbers)).values(status=status, updated_at=_now(), **more))
# ── sign-off history (append-only) and the statements of a signed period ──────────────────────────────────
def signoffs(self, group_id: str, period_id: str) -> list:
t = cons_signoff
rows = [dict(r) for r in self.conn.execute(select(t).where(t.c.group_id == group_id, t.c.period_id == period_id).order_by(t.c.seq)).mappings().all()]
for r in rows: # stored as naive UTC, read back as an aware moment
if r.get("at") is not None and r["at"].tzinfo is None:
r["at"] = r["at"].replace(tzinfo=timezone.utc)
return rows
def add_signoff(self, group_id: str, period_id: str, held: list, **row) -> None:
"""Append one row to the period's sign-off history: the next seq, a plain INSERT. Nothing is ever updated or deleted."""
now = _now()
seq = max([h["seq"] for h in held], default=0) + 1 # erp-lint: allow L6 claimed by a plain INSERT on the key (group, period, seq)
self.conn.execute(insert(cons_signoff), [{"group_id": group_id, "period_id": period_id, "seq": seq, "at": now, **row,
"created_at": now, "updated_at": now}])
def statements(self, group_id: str, period_id: str, reports: list) -> list:
t = cons_statement
return [dict(r) for r in self.conn.execute(select(t).where(t.c.group_id == group_id, t.c.period_id == period_id,
t.c.report.in_(reports))).mappings().all()]
def write_statements(self, rows: list) -> None:
upsert(self.conn, cons_statement, _stamp(rows), ("report", "group_id", "period_id", "node_id"))
# ── the year-end carry-forward (CON-15) ───────────────────────────────────────────────────────────────────
def carry_forward(self, group_id: str, fiscal_year: str) -> list:
t = cons_carry_forward
return [dict(r) for r in self.conn.execute(select(t).where(t.c.group_id == group_id, t.c.fiscal_year == fiscal_year)).mappings().all()]
def write_carry_forward(self, rows: list) -> None:
"""Keyed [group, fiscal year, account, company, partner, segment], every value in VALUES: the same derivation writes the same rows."""
upsert(self.conn, cons_carry_forward, _stamp(rows), ("group_id", "fiscal_year", "group_account_id", "company_code", "partner_company", "segment"))
def mark_reversed(self, entries: list) -> None:
"""The earlier run's entries become `reversed` (never deleted), version 2 claimed by a plain INSERT."""
if not entries:
return
now = _now()
ids = [e["cons_journal_id"] for e in entries]
self.conn.execute(insert(CONS_JOURNAL.versions), [{"cons_journal_id": i, "version": 2, "saved_at": now} for i in ids])
self.conn.execute(update(cons_journal).where(cons_journal.c.cons_journal_id.in_(ids))
.values(status="reversed", version=2, updated_at=now))
self.conn.execute(insert(cons_audit), [{"audit_id": f"cons_journal:{i}:v2:1", "group_id": entries[0]["group_id"],
"period_id": entries[0]["period_id"], "subject": i, "event": "ConsJournalReversed",
"sub": "", "event_at": now, "created_at": now, "updated_at": now} for i in ids])
def zero_fill(existing_keys: set, fresh_keys: set, base: dict) -> list:
"""Rows a company had before and gl no longer sends are written as zero, so storage matches gl without a delete."""
return [{**base, **dict(zip(REPORTED_KEY[2:], k)), "ytd_local": Decimal("0.00"), "period_local": Decimal("0.00")}
for k in sorted(existing_keys - fresh_keys)]
class MigrationData:
"""The rows the Odoo loader (CON-34) reads and writes, each a statement whatever the number of companies, accounts or journals in a chunk."""
def __init__(self, conn):
self.conn = conn
def existing_journals(self, journal_ids: list) -> set:
if not journal_ids:
return set()
return {r[0] for r in self.conn.execute(select(cons_journal.c.cons_journal_id).where(cons_journal.c.cons_journal_id.in_(journal_ids)))}
def journal_versions(self, journal_ids: list) -> dict:
"""{journal: (version, [lines in line order])} of the carried journals already written: one statement (heads joined to lines)."""
if not journal_ids:
return {}
j, ln = cons_journal, cons_journal_line
stmt = (select(j.c.cons_journal_id.label("jid"), j.c.version.label("head_version"), ln)
.select_from(j.outerjoin(ln, ln.c.cons_journal_id == j.c.cons_journal_id))
.where(j.c.cons_journal_id.in_(journal_ids)).order_by(j.c.cons_journal_id, ln.c.line_no))
out: dict = {}
for r in self.conn.execute(stmt).mappings().all():
lines = out.setdefault(r["jid"], (r["head_version"], []))[1]
if r["cons_journal_id"] is not None: # a journal with no lines has one row of NULLs
lines.append({k: v for k, v in r.items() if k not in ("jid", "head_version")})
return out
def supersede_journals(self, changed: list) -> None:
"""Carried journals whose amounts Odoo has since corrected: the next version of each (claimed by a plain INSERT on [journal, version]),
its lines rewritten to Odoo's (a line Odoo no longer has is written as zero, never deleted), an audit row per journal. Five statements
for the chunk, whatever the number of journals. changed: [{cons_journal_id, version, group_id, period_id, lines}]."""
if not changed:
return
now = _now()
self.conn.execute(insert(CONS_JOURNAL.versions), [{"cons_journal_id": c["cons_journal_id"], "version": c["version"] + 1, "saved_at": now}
for c in changed])
ids = [c["cons_journal_id"] for c in changed]
self.conn.execute(update(cons_journal).where(cons_journal.c.cons_journal_id.in_(ids))
.values(version=cons_journal.c.version + 1, updated_at=now))
rows = []
for c in changed:
for n, line in enumerate(c["lines"], 1):
rows.append({"cons_journal_id": c["cons_journal_id"], "line_no": n, **{k: line[k] for k in (
"group_account_id", "company_code", "partner_company", "movement_type", "segment", "amount")}})
for n in range(len(c["lines"]) + 1, c.get("held_lines", 0) + 1):
rows.append({"cons_journal_id": c["cons_journal_id"], "line_no": n, "group_account_id": c["old"][n - 1]["group_account_id"],
"company_code": c["old"][n - 1]["company_code"], "partner_company": "", "movement_type": "", "segment": "",
"amount": Decimal("0.00")})
upsert(self.conn, cons_journal_line, _stamp(rows), ("cons_journal_id", "line_no"))
self.conn.execute(insert(cons_audit), [{"audit_id": f"cons_journal:{c['cons_journal_id']}:v{c['version'] + 1}:1", "group_id": c["group_id"],
"period_id": c["period_id"], "subject": c["cons_journal_id"], "event": "ConsJournalSuperseded",
"sub": "", "event_at": now, "created_at": now, "updated_at": now} for c in changed])
def maps(self, odoo_db: str, odoo_model: str, odoo_ids: list) -> dict:
t = cons_migration_map
if not odoo_ids:
return {}
return {r[0]: r[1] for r in self.conn.execute(select(t.c.odoo_id, t.c.natural_key).where(
t.c.odoo_db == odoo_db, t.c.odoo_model == odoo_model, t.c.odoo_id.in_(odoo_ids)))}
def write_maps(self, rows: list) -> None:
"""Keyed [odoo_db, odoo_model, odoo_id]; the caller passes only the rows that are missing or differ."""
if rows:
upsert(self.conn, cons_migration_map, _stamp([{**r, "version": 1} for r in rows]), ("odoo_db", "odoo_model", "odoo_id"))
def balances(self, group_id: str, periods: list) -> dict:
"""{(period, group account): SUM(amount)} of the carried migration journals: what the report `migration` holds. One statement."""
j, ln = cons_journal, cons_journal_line
stmt = (select(j.c.period_id, ln.c.group_account_id, func.sum(ln.c.amount)).select_from(j.join(ln, ln.c.cons_journal_id == j.c.cons_journal_id))
.where(j.c.group_id == group_id, j.c.document_type == "migration", j.c.period_id.in_(periods))
.group_by(j.c.period_id, ln.c.group_account_id))
return {(p, a): Decimal(str(v)).quantize(Decimal("0.01")) for p, a, v in self.conn.execute(stmt)}