dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/application/use_cases.py
"""The use cases — what a person (or their agent) can do here.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and
nowhere else: WHO MAY (the `permission` on the declaration) and WHICH ROWS (`_visible_<root>()`, from the
caller: a row the caller may not see is a 404, never a 403). Each command is one unit of work: load the
aggregate, let the domain decide, save it whole at the version it was loaded (`If-Match` for a screen).
Row visibility (module.yaml): everyone who can consolidate sees the whole group; a person's gl company access
does not narrow group figures (they are the group's), but the drill-down from a group figure to gl's lines
answers 404 for a company the person cannot see in gl
Generated by `erp-compose scaffold` from module.yaml.
"""
import dataclasses
import json
from contextlib import contextmanager
from datetime import date, timedelta
from typing import Optional, Union
from erp_platform import work
from erp_platform.bricks import Brick, Refused
from erp_platform.seam import derived_id
from sqlalchemy.exc import IntegrityError
from brick_consolidation.application import carry, close, loader, migration, reports, runs, structure
from brick_consolidation.application.runs import the_period
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.cons_journal import ConsJournal
from brick_consolidation.domain.cons_migration_run import ConsMigrationRun
from brick_consolidation.domain.errors import Invalid, RuleBroken
from brick_consolidation.domain.group import AccountRole, AccountRule, EliminationRule, Group
from brick_consolidation.infrastructure import clock
CONS = Brick("consolidation", version="0.1.0",
instructions="Turn the books of every company in the group into one set of group accounts: translated into the group's currency, with what the companies sold, lent and charged each other taken out, the parent's investments replaced by the assets behind them, and the minority's share shown, signed off by a second person each period.")
# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.
CONS.declare_process("process.yaml")
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text_ = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text_ or "duplicate key" in text_ or "UNIQUE constraint" in text_
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a write conflict on a claim row (a run, an entry number, a sign-off row) can surface at COMMIT - the loser's
COMMIT raises 23505 after the use case has returned. It is caught here, around the commit, and answered as the retry-safe 409 a conflict
seen mid-use-case gets: never a 500, and nothing of the loser written (found by the live check, 2026-10-24: two racing translate runs)."""
from brick_consolidation.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago: load it again and retry", gate="write_conflict",
fix="repeat the request - what the other request wrote is returned, nothing is written twice") from None
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
repository = _repository_default # tests swap in one on SQLite
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words, through both doors."""
try:
yield
except Invalid as bad:
raise Refused(422, str(bad)) from None
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
finally:
clock.reset_gl_zone() # the zone read from gl belongs to the request that read it
def _who(caller: dict) -> str:
return caller.get("email") or caller.get("sub") or ""
def _given(**values) -> dict:
"""The inputs the caller sent: None means "left out", never "set to nothing"."""
return {k: v for k, v in values.items() if v is not None}
def _same_version(what: str, current: int, asked: Optional[int]) -> None:
if asked is not None and int(asked) != current:
raise Refused(409, f"{what} changed since you loaded it (you have version {asked}, it is at "
f"{current}) — load it again")
def _view(agg, version: int, key: str = "version") -> dict:
out = {}
for f in dataclasses.fields(agg):
if f.name == "events":
continue
v = getattr(agg, f.name)
out[f.name] = [dataclasses.asdict(x) for x in v] if isinstance(v, list) else v
out[key] = version
return out
def _visible_group(repo, caller: dict, group_id: str) -> tuple:
found = repo.groups.get(group_id)
# TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with
# the read permission sees every row.
if found is None:
raise Refused(404, f"no group {group_id}")
_read_zone(found[0])
return found
def _read_zone(group) -> None:
"""v3: business_timezone is read from gl's GET /companies/{top_company_code} per request, never stored (the top company is always of source gl).
gl being away leaves the environment's zone (Europe/London by default): a read of a zone never blocks a read of the group."""
if not group.top_company_code:
return
try:
clock.use_gl_zone(runs.ports.gl.company_row(group.top_company_code).get("business_timezone") or "")
except runs.Missing:
clock.reset_gl_zone()
@CONS.use_case(kind="query", route="GET /groups", permission="cons:group:read")
def list_groups(caller: dict) -> dict:
"""The groups the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"groups": repo.groups.ids()}
@CONS.use_case(kind="query", route="GET /groups/{group_id}", permission="cons:group:read")
def get_group(caller: dict, group_id: str) -> dict:
"""One group, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_group(repo, caller, group_id), key="version")
GROUP_FIXED_ONCE_RUN = {"group_currency", "ledger_id", "top_company_code", "fiscal_year_end_month", "group_chart_id"} # audit v1 MUST-3c
@CONS.use_case(kind="command", route="POST /groups", permission="cons:structure:write", idempotent=True, if_match="version")
def keep_group(caller: dict, group_id: str, version: Optional[int] = None, name: Optional[str] = None, group_currency: Optional[str] = None, ledger_id: Optional[str] = None, group_chart_id: Optional[str] = None, top_company_code: Optional[str] = None, fiscal_year_end_month: Optional[str] = None, fsv_id: Optional[str] = None, goodwill_life_years: Optional[str] = None, cash_flow_fsv_id: Optional[str] = None) -> dict:
"""Create or change a group: a set of companies reported together, in one currency, from one gl ledger and
one group chart, under a top company. Fields left out keep their value; saving the same values again
changes nothing.
"""
with repository() as repo, _decide():
found = repo.groups.get(group_id)
if found is None:
agg, current = Group(group_id=group_id), 0
else:
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
given = _given(name=name, group_currency=group_currency, ledger_id=ledger_id, group_chart_id=group_chart_id,
top_company_code=top_company_code, fiscal_year_end_month=fiscal_year_end_month, fsv_id=fsv_id,
goodwill_life_years=goodwill_life_years, cash_flow_fsv_id=cash_flow_fsv_id)
if "group_currency" in given and not (len(given["group_currency"]) == 3 and given["group_currency"].isalpha()
and given["group_currency"].isupper()):
raise Invalid("group_currency must be an ISO 4217 code (GBP)")
if "fiscal_year_end_month" in given and given["fiscal_year_end_month"] not in {str(m) for m in range(1, 13)}:
raise Invalid("fiscal_year_end_month must be 1 to 12")
changed = {k for k, v in given.items() if str(getattr(agg, k) or "") != str(v or "")}
ran = [x.period_id for x in agg.group_periods if x.status != "open"]
fixed = sorted(changed & GROUP_FIXED_ONCE_RUN)
if fixed and ran:
raise RuleBroken(f"{', '.join(fixed)} cannot change once a period has been run ({ran[0]}): every figure of the group was computed in "
"that currency, from that ledger, under that top company and year — set up a new group for a new basis")
for key, value in given.items():
setattr(agg, key, value)
if "goodwill_life_years" in changed:
agg.mark_stale() # the amortisation is cumulative from each acquisition: every unsigned run is stale
return _view(agg, repo.groups.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /groups/{group_id}/rules", permission="cons:settings:write", idempotent=True, if_match="version")
def keep_group_rules(caller: dict, group_id: str, version: Optional[int] = None, account_rules: Optional[list] = None, account_roles: Optional[list] = None, elimination_rules: Optional[list] = None) -> dict:
"""Set how the group translates and eliminates: per group account its translation rate type (closing, average
or historical) and whether it is intercompany or equity; which group account takes the translation reserve,
NCI, goodwill and differences (the roles); and the elimination rules (which accounts are removed against which,
the difference role and the tolerance in group currency). Rows are keyed; the same row again changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
held = (list(agg.account_rules), list(agg.account_roles), list(agg.elimination_rules))
for r in account_rules or []:
if r.get("rate_type") not in ("closing", "average", "historical"):
raise Invalid("rate_type must be closing, average or historical")
row = AccountRule(group_id=group_id, group_account_id=str(r["group_account_id"]), valid_from_period=C.check_period(r["valid_from_period"]),
rate_type=r["rate_type"], ic="true" if r.get("ic") else "", equity="true" if r.get("equity") else "")
agg.account_rules = [x for x in agg.account_rules if (x.group_account_id, x.valid_from_period) != (row.group_account_id, row.valid_from_period)] + [row]
for r in account_roles or []:
row = AccountRole(group_id=group_id, role=str(r["role"]), valid_from_period=C.check_period(r["valid_from_period"]),
group_account_id=str(r["group_account_id"]))
agg.account_roles = [x for x in agg.account_roles if (x.role, x.valid_from_period) != (row.role, row.valid_from_period)] + [row]
for r in elimination_rules or []:
if r.get("kind") not in ("ic_balance", "ic_pl", "dividend"):
raise Invalid("an elimination rule's kind is ic_balance, ic_pl or dividend")
row = EliminationRule(group_id=group_id, rule_id=str(r["rule_id"]), kind=r["kind"], left_accounts=str(r.get("left_accounts", "")),
right_accounts=str(r.get("right_accounts", "")), difference_role=str(r.get("difference_role", "")),
tolerance=C.amount(r.get("tolerance", "0.00"), "tolerance"))
agg.elimination_rules = [x for x in agg.elimination_rules if x.rule_id != row.rule_id] + [row]
# the runs computed under the old rules are stale from the changed rule's valid_from (an elimination rule has none: every unsigned period)
moved = [x for old_, new in zip(held, (agg.account_rules, agg.account_roles, agg.elimination_rules)) for x in old_ + new
if (x in old_) != (x in new)]
if moved:
agg.mark_stale(min(getattr(x, "valid_from_period", "") for x in moved))
return _view(agg, repo.groups.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods", permission="cons:period:open", idempotent=True, if_match="version")
def open_group_period(caller: dict, group_id: str, period_id: str, version: Optional[int] = None) -> dict:
"""Open the group period (group_period: none → open). Only when: the structure is approved for the period;
the previous period is signed (or this is the first). Then: scope and methods fixed from the structure in
force on the period's last day. A repeat changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
C.check_period(period_id)
if any(x.period_id == period_id for x in agg.group_periods):
if carry.first_period_of(agg, period_id) == period_id and carry.has_previous_year(agg, period_id) and \
next(x for x in agg.group_periods if x.period_id == period_id).status != "signed":
out = carry.derive(repo, agg, period_id) # opened again: the carry-forward is derived again, the same rows when nothing moved
return {**_view(agg, current, key="version"), "result": {"carry_forward": out}}
return _view(agg, current, key="version")
if not agg.group_structures: # a group that has sold its last subsidiary still closes: the structure exists, it has rows
raise RuleBroken(f"no structure is approved for {period_id}: record and approve the group's companies first (CON-01)")
before = [x for x in agg.group_periods if x.period_id < period_id]
if before and max(before, key=lambda x: x.period_id).status != "signed":
raise RuleBroken(f"the previous period {max(before, key=lambda x: x.period_id).period_id} is not signed")
if agg.open_group_period(period_id, by=_who(caller), run_no=1) is None:
return _view(agg, current, key="version")
out = carry.derive(repo, agg, period_id) if carry.first_period_of(agg, period_id) == period_id and carry.has_previous_year(agg, period_id) else None
view = _view(agg, repo.groups.save(agg, current), key="version")
return {**view, "result": {"carry_forward": out}} if out else view
def _run(caller, group_id, period_id, version, fn) -> dict:
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
result = fn(repo, agg, period_id, by=_who(caller))
if result is None: # nothing to do: nothing is written, not even a version
return {**_view(agg, current, key="version"), "result": {"changed": False}}
gp = the_period(agg, period_id) # the preparers of the period: whoever ran any step of any run (CON-14)
gp.prepared_by, gp.prepared_sub = caller.get("email") or "", caller.get("sub") or ""
gp.record_preparer(getattr(fn, "__name__", "step"), gp.prepared_by, gp.prepared_sub)
return {**_view(agg, repo.groups.save(agg, current), key="version"), "result": {"changed": True, **result}}
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/upload-reported-data", permission="cons:data:upload", if_match="version")
def upload_reported_data(caller: dict, group_id: str, period_id: str, company_code: str, currency: str, csv: str, version: Optional[int] = None) -> dict:
"""Upload the figures of a company kept outside gl (group_period: open → open). The CSV has the columns group_account_id,
partner_company, movement_type, segment, ytd_amount (year to date, in the company's currency). Only when: the company is in scope and
not kept in gl; the accounts exist in the group chart, amounts are NUMERIC(18,2), the partners are group companies and the figures
net to zero. Then: the rows wait for a second person to accept them, the file is kept under raw/consolidation/<group>/<period>/ with
its sha-256, and an earlier upload still waiting is replaced.
"""
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
out = close.upload(repo, agg, period_id, company=company_code, currency=currency, content=csv, by=caller.get("email") or "",
sub=caller.get("sub") or "")
agg.upload_reported_data(period_id, by=_who(caller))
return {**_view(agg, repo.groups.save(agg, current), key="version"), "result": out}
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/accept-reported-upload", permission="cons:data:accept", if_match="version")
def accept_reported_upload(caller: dict, group_id: str, period_id: str, company_code: str, currency: str, version: Optional[int] = None) -> dict:
"""Accept the upload (group_period: open → open). Only when: the accepter is not the uploader (email AND sub). Then: the rows become
the company's reported data (source upload). Accepting a replacement after the period was collected marks translation and
eliminations stale; the next runs take run_no + 1.
"""
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
out = close.accept(repo, agg, period_id, company=company_code, currency=currency, by=caller.get("email") or "", sub=caller.get("sub") or "")
if out is None:
return {**_view(agg, current, key="version"), "result": {"changed": False}}
agg.accept_reported_upload(period_id, by=_who(caller))
return {**_view(agg, repo.groups.save(agg, current), key="version"), "result": {"changed": True, **out}}
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/collect-group-data", permission="cons:data:collect", idempotent=True, if_match="version")
def collect_group_data(caller: dict, group_id: str, period_id: str, version: Optional[int] = None) -> dict:
"""Collect the companies' figures from gl (group_period: open → collected). Reads gl's group-reporting
balances per company in scope and stores them (year to date and period, by group account, partner, movement
and segment, in local currency) with gl's digest. Refused when a company's gl period is still open (unless the
owner allows it, then flagged provisional), when accounts are unmapped, or a company's figures do not net to
zero. Collecting again with unchanged digests writes nothing; a changed digest takes the next run number and
makes translation and eliminations stale.
"""
return _run(caller, group_id, period_id, version, runs.collect)
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/recollect-group-data", permission="cons:data:collect", if_match="version")
def recollect_group_data(caller: dict, group_id: str, period_id: str, version: Optional[int] = None) -> dict:
"""Collect again (group_period: collected, translated, eliminated → collected). Only when a company's digest
changed or the structure changed; the later steps are then stale and the next runs take run_no + 1.
"""
def again(repo, agg, pid, by):
if the_period(agg, pid).status == "open":
raise RuleBroken("nothing is collected yet — collect first")
out = runs.collect(repo, agg, pid, by=by)
if out is None:
raise RuleBroken("nothing changed: every company's digest is the one already held")
return out
return _run(caller, group_id, period_id, version, again)
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/translate-currencies", permission="cons:translate:run", idempotent=True, if_match="version")
def translate_currencies(caller: dict, group_id: str, period_id: str, version: Optional[int] = None) -> dict:
"""Translate into the group currency (group_period: collected → translated). Balance-sheet accounts at gl's
closing rate, income and expense month by month at gl's average rates, equity at the amounts fixed at
acquisition, each by the account's rule. The balancing difference of each foreign company is one automatic
entry to the translation reserve, with the minority's share on the NCI account. A missing rate in gl refuses
and names the currency and month. Translating again writes nothing; after a new collection it replaces the
earlier run's entries with mirror entries and takes the next run.
"""
return _run(caller, group_id, period_id, version, runs.translate)
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/run-eliminations", permission="cons:eliminate:run", idempotent=True, if_match="version")
def run_eliminations(caller: dict, group_id: str, period_id: str, version: Optional[int] = None) -> dict:
"""Run the eliminations (group_period: translated → eliminated). Intercompany receivable against payable,
revenue against cost, the dividend against equity, and the unrealised profit in stock with its deferred tax,
each removed per pair of companies by an automatic entry; differences go to the difference accounts. Refused
per pair above tolerance (with gl's break list and its fix), when intercompany stock is neither entered nor
stated none, or when a full or equity company has no approved acquisition. Running again writes nothing.
"""
return _run(caller, group_id, period_id, version, runs.eliminate)
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/ic-inventory", permission="cons:inventory:write", idempotent=True, if_match="version")
def enter_ic_inventory(caller: dict, group_id: str, period_id: str, holder_company: str, seller_company: str, inventory_amount: Optional[str] = None, margin_pct: Optional[str] = None, tax_rate_pct: Optional[str] = None, none_stated: bool = False, version: Optional[int] = None) -> dict:
"""Enter the stock a company holds that it bought from another group company (amount in the holder's currency,
the seller's margin and the holder's tax rate), or state that there is none. The elimination of unrealised
profit reads it. Entering the same figures again changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
gp = the_period(agg, period_id)
if gp.status in ("signed", "eliminated"):
raise RuleBroken(f"{period_id} is {gp.status}: collect again (CON-20) before changing intercompany stock")
members = C.members_at(agg, period_id)
for code in (holder_company, seller_company):
if code not in members:
raise RuleBroken(f"{code} is not in the group's scope in {period_id}")
if holder_company == seller_company:
raise RuleBroken("a company does not buy stock from itself")
if none_stated:
amount_, margin, tax = C.ZERO, C.pct("0", "margin", zero_ok=True), None
else:
if inventory_amount in (None, "") or margin_pct in (None, ""):
raise Invalid("enter the stock amount and the seller's margin, or state none")
amount_ = C.amount(inventory_amount, "inventory amount")
margin = C.pct(margin_pct, "margin", zero_ok=True)
tax = C.pct(tax_rate_pct, "tax rate", zero_ok=True) if tax_rate_pct not in (None, "") else None # v3: absent = gl's statutory rate at the period end
repo.data.write_inventory({"group_id": group_id, "period_id": period_id, "holder_company": holder_company, "seller_company": seller_company,
"inventory_amount": amount_, "margin_pct": margin, "none_stated": "true" if none_stated else "false",
"entered_by": caller.get("email") or "", "entered_sub": caller.get("sub") or "", "tax_rate_pct": tax})
return _view(agg, repo.groups.save(agg, current), key="version")
@CONS.use_case(kind="query", route="GET /groups/{group_id}/group-periods/{period_id}/readiness", permission="cons:group:read")
def group_close_readiness(caller: dict, group_id: str, period_id: str) -> dict:
"""Open the group close: each gate as {gate, met, reason, fix} — every company collected and current, each company's and the translated
data netting to zero, intercompany within tolerance, every investment eliminated, no manual entry waiting, no step stale, the
consolidated trial balance netting to zero and the balance sheet balancing. Each failing gate names the company, pair or account.
"""
with repository() as repo, _decide():
agg, _current = _visible_group(repo, caller, group_id)
return close.readiness(repo, agg, period_id)
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/sign-off-group-close", permission="cons:period:signoff", idempotent=True, if_match="version")
def sign_off_group_close(caller: dict, group_id: str, period_id: str, version: Optional[int] = None) -> dict:
"""Sign off the group close (group_period: eliminated → signed). Only when: readiness met (CON-13); the signer is not the preparer
(email AND sub). Then: who, when and the digest of the consolidated trial balance recorded in the period's history; the statements of
the group's statement versions materialised and stable. A signed period refuses every run until it is reopened. A repeat changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
out = close.sign_off(repo, agg, period_id, by=caller.get("email") or "", sub=caller.get("sub") or "")
if out is None:
return {**_view(agg, current, key="version"), "result": {"changed": False}}
return {**_view(agg, repo.groups.save(agg, current), key="version"), "result": {"changed": True, **out}}
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/materialise-statements", permission="cons:statement:materialise", idempotent=True)
def materialise_group_statements(caller: dict, group_id: str, period_id: str) -> dict:
"""Materialise the statements of a signed period (CON-21). Computes them afresh from the books at the sign-off digest and writes only
the rows that differ from what is stored (keyed report, group, period, node): the second run changes nothing. Refused when the period is
not signed, or when its books no longer match the digest it was signed at (then the stored statements stay as signed)."""
with repository() as repo, _decide():
agg, _current = _visible_group(repo, caller, group_id)
gp = the_period(agg, period_id)
if gp.status != "signed":
raise Refused(409, f"{period_id} is {gp.status}: statements are materialised when a period is signed", gate="signed", fix="sign the period off first")
return reports.materialise_signed(repo, agg, gp)
@CONS.use_case(kind="command", route="POST /groups/{group_id}/group-periods/{period_id}/reopen-group-period", permission="cons:period:reopen", idempotent=True, if_match="version")
def reopen_group_period(caller: dict, group_id: str, period_id: str, reason: str, version: Optional[int] = None) -> dict:
"""Reopen the group period (group_period: signed → collected). Only when: a reason is given; the reopener is not who signed it (email
AND sub); the next period is not signed. Then: appended to the period's sign-off history with who and why (the signed history is never
rewritten); the figures are provisional; the next runs take run_no + 1. A repeat changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_group(repo, caller, group_id)
_same_version(f"group {group_id}", current, version)
out = close.reopen(repo, agg, period_id, reason=reason, by=caller.get("email") or "", sub=caller.get("sub") or "")
if out is None:
return {**_view(agg, current, key="version"), "result": {"changed": False}}
return {**_view(agg, repo.groups.save(agg, current), key="version"), "result": {"changed": True, **out}}
@CONS.use_case(kind="query", route="GET /groups/{group_id}/group-periods/{period_id}/signoffs", permission="cons:group:read")
def get_signoff_history(caller: dict, group_id: str, period_id: str) -> dict:
"""The period's sign-off history: every sign-off and reopen, with who, when, the digest and the reason. Append-only."""
with repository() as repo:
_visible_group(repo, caller, group_id)
return {"signoffs": repo.data.signoffs(group_id, period_id)}
@CONS.use_case(kind="query", route="GET /groups/{group_id}/group-periods/{period_id}/lines", permission="cons:group:read")
def get_figure_lines(caller: dict, group_id: str, period_id: str, company_code: str, group_account_id: str, figure: str = "year_to_date",
limit: int = 500, offset: int = 0) -> dict:
"""CON-11: the drill-down from a company's group-account figure to gl's line history behind it (gl GL-04), local currency, before translation.
`figure` year_to_date reads from the first day of the company's fiscal year, period from the period's first day, to the period's last day. gl is
asked UNDER THE CALLER, so gl's company scope applies: a company the caller cannot see in gl answers 404. A company of source upload has no gl lines: its
drill stops at the accepted upload (source upload, no lines)."""
with repository() as repo, _decide():
if figure not in ("year_to_date", "period"):
raise Invalid("figure is year_to_date or period")
agg, _current = _visible_group(repo, caller, group_id)
the_period(agg, period_id)
members = C.members_at(agg, period_id)
if company_code not in members:
raise Refused(404, f"{company_code} is not in the group in {period_id}")
row = next((r for r in agg.group_structures if r.company_code == company_code and r.valid_from_period <= period_id), None)
if row is not None and row.source == "upload":
return {"company_code": company_code, "group_account_id": group_account_id, "period_id": period_id, "source": "upload", "lines": [],
"note": "this company's figures are an accepted upload (CON-12): the drill-down stops at the upload, there is no gl line history"}
first = C.fiscal_months(period_id, agg.fiscal_year_end_month)[0] if figure == "year_to_date" else period_id
frm = date(int(first[:4]), int(first[5:7]), 1)
try:
got = runs.ports.gl.account_lines(group_account_id, company_code, agg.ledger_id, frm, C.period_end(period_id), limit, offset, caller=caller)
except runs.Missing as m:
raise Refused(404 if m.status == 404 else 502, f"gl: {m.reason}") from None
return {"company_code": company_code, "group_account_id": group_account_id, "period_id": period_id, "source": "gl", **got}
@CONS.use_case(kind="query", route="GET /groups/{group_id}/group-periods/{period_id}/report", permission="cons:report:read")
def run_group_report(caller: dict, group_id: str, period_id: str, fsv_id: Optional[str] = None, report: str = "statement") -> dict:
"""A group statement: each line is the sum of translated data and consolidation entries through gl's published statement version
(balance sheet and profit and loss, or cash flow — indirect, with translation differences on cash as their own line), with the
companies' and segments' contributions, the entries behind it and the prior year beside it. report=equity_changes reads the movement
types instead. A signed period's report is read from what was materialised at sign-off. It names the version, ledger and run.
"""
with repository() as repo, _decide():
agg, _current = _visible_group(repo, caller, group_id)
gp = the_period(agg, period_id)
if report == "equity_changes":
return reports.equity_changes(repo, agg, gp)
chosen = fsv_id or agg.fsv_id
if not chosen:
raise RuleBroken("the group names no statement version: report by group account only, or set fsv_id on the group")
return reports.run(repo, agg, gp, chosen)
def _visible_structure_change(repo, caller: dict, change_id: str) -> tuple:
found = repo.structure_changes.get(change_id)
if found is None:
raise Refused(404, f"no structure change {change_id}")
return found
@CONS.use_case(kind="query", route="GET /structure-changes", permission="cons:group:read")
def list_structure_changes(caller: dict) -> dict:
"""The structure changes the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"structure_changes": repo.structure_changes.ids()}
@CONS.use_case(kind="query", route="GET /structure-changes/{change_id}", permission="cons:group:read")
def get_structure_change(caller: dict, change_id: str) -> dict:
"""One structure change, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_structure_change(repo, caller, change_id), key="version")
@CONS.use_case(kind="command", route="POST /structure-changes", permission="cons:structure:write", idempotent=True)
def propose_structure_change(caller: dict, group_id: str, kind: str, company_code: str, investor_company_code: str, ownership_pct: str, effective_date: date, method: str = "full", voting_pct: Optional[str] = None, consideration: Optional[str] = None, fair_value_adjustments: Optional[list] = None, proceeds: Optional[str] = None, previously_held_fair_value: Optional[str] = None, source: str = "gl", local_currency: Optional[str] = None, name: Optional[str] = None) -> dict:
"""Propose a structure change: an acquisition (or ownership or method change) of a company by another in the
group, from a date. The company and its investor must exist in gl; ownership and voting are above 0 and at most
100; it cannot take effect in a signed period. For an acquisition the net assets are read from gl, the goodwill
(consideration less the share of net assets) and the non-controlling interest (the rest) are computed in local and
group currency at gl's rate on the day, and shown; fair-value adjustments (amount, tax_rate_pct, the group_account_id stepped up) are lines
of the acquisition with their deferred tax. A divestiture (the last day of a period, with the proceeds) shows the gain, the recycled
translation reserve and the minority derecognised, computed from the period's translated data; an ownership change either shows the
equity transaction (the stake bought without a change of control) or, for an associate taken to a subsidiary, the goodwill at that date
and the gain on the interest held before (previously_held_fair_value). A second person approves. Proposing the same again returns it.
v3: `source` says where the company's books are: gl (the default: gl must know the company) or upload (gl must NOT know it: the code is 1 to 4
upper-case letters or digits and unique in the group, and the row names local_currency and name); a later move of an upload company into gl is a
method_change row of source gl. A fair-value adjustment's deferred tax uses its own tax_rate_pct, else gl's statutory_tax_rate of the acquired company
on the effective date, else the proposal is refused (gate tax_rate_known).
"""
with repository() as repo, _decide():
agg, _current = _visible_group(repo, caller, group_id)
it = structure.propose(repo, agg, source=source, local_currency=local_currency, name=name, kind=kind, company_code=company_code, investor_company_code=investor_company_code,
ownership_pct=ownership_pct, voting_pct=voting_pct, method=method, effective_date=effective_date,
consideration=consideration, fair_value_adjustments=fair_value_adjustments, proceeds=proceeds,
previously_held_fair_value=previously_held_fair_value,
by=caller.get("email") or "", sub=caller.get("sub") or "")
found = repo.structure_changes.get(it.change_id)
if found is not None:
return _view(*found, key="version")
it.proposed(by=_who(caller))
return _view(it, repo.structure_changes.save(it, 0), key="version")
@CONS.use_case(kind="command", route="POST /structure-changes/{change_id}/approve-structure-change", permission="cons:structure:approve", idempotent=True, if_match="version")
def approve_structure_change(caller: dict, change_id: str, version: Optional[int] = None) -> dict:
"""Approve the structure change (structure_change: proposed → effective). The approver is not the proposer
(email and sub). Writes the dated structure row, fixes the acquired equity at the rate of the day (historical
amounts) and makes the periods collected from then on stale. A change cannot take effect in a signed period.
A repeat changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_structure_change(repo, caller, change_id)
_same_version(f"structure change {change_id}", current, version)
if agg.status == "effective":
return _view(agg, current, key="version")
group, group_version = _visible_group(repo, caller, change_id.split(":")[0])
structure.approve(repo, group, agg, by=caller.get("email") or "", sub=caller.get("sub") or "")
agg.approved(by=_who(caller))
repo.groups.save(group, group_version) # claims the group's next version: a run at the same moment collides
return _view(agg, repo.structure_changes.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /structure-changes/{change_id}/reject-structure-change", permission="cons:structure:approve", idempotent=True, if_match="version")
def reject_structure_change(caller: dict, change_id: str, reason: str, version: Optional[int] = None) -> dict:
"""Reject the structure change (structure_change: proposed → rejected). A repeat changes nothing."""
with repository() as repo, _decide():
agg, current = _visible_structure_change(repo, caller, change_id)
_same_version(f"structure change {change_id}", current, version)
if agg.reject_structure_change(reason=reason, by=_who(caller)) is None:
return _view(agg, current, key="version")
return _view(agg, repo.structure_changes.save(agg, current), key="version")
def _visible_cons_journal(repo, caller: dict, cons_journal_id: str) -> tuple:
found = repo.cons_journals.get(cons_journal_id)
# TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with
# the read permission sees every row.
if found is None:
raise Refused(404, f"no cons journal {cons_journal_id}")
return found
@CONS.use_case(kind="query", route="GET /cons-journals", permission="cons:group:read")
def list_cons_journals(caller: dict) -> dict:
"""The cons journals the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"cons_journals": repo.cons_journals.ids()}
@CONS.use_case(kind="query", route="GET /cons-journals/{cons_journal_id}", permission="cons:group:read")
def get_cons_journal(caller: dict, cons_journal_id: str) -> dict:
"""One cons journal, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_cons_journal(repo, caller, cons_journal_id), key="version")
# POST /cons-journals (the scaffold's raw post_cons_auto_entry door) is removed (audit v1 MUST-2b): automatic entries are posted only by
# the runs themselves (runs.translate / runs.eliminate, keyed derived_id, the period not signed), manual ones only through propose + approve.
@CONS.use_case(kind="command", route="POST /cons-journals/propose-cons-entry", permission="cons:journal:write", idempotent=True)
def propose_cons_entry(caller: dict, cons_journal_id: str, group_id: str, period_id: str, lines: list[dict], memo: str = "", period_specific: bool = False) -> dict:
"""Propose a manual consolidation entry (cons_journal: none → proposed; document type manual). `cons_journal_id` is the caller's key for
it: asking again with the same key returns the entry. Each line names group_account_id, company_code, amount (and partner_company, a
movement_type where the account tracks movements, segment). Only when: balanced in group currency; every line names a company in scope;
the period not signed. A period-specific entry reverses at the start of the next period; a cumulative one carries through the year.
"""
with repository() as repo, _decide():
key = str(derived_id("cons-manual", group_id, cons_journal_id))
found = repo.cons_journals.get(key)
if found is not None:
return _view(*_visible_cons_journal(repo, caller, key), key="version")
agg, current = _visible_group(repo, caller, group_id)
gp = the_period(agg, period_id)
entry = ConsJournal.propose_cons_entry(key, by=_who(caller), lines=lines, scope=set(C.members_at(agg, period_id)), rules=agg.account_rules,
period_signed=gp.status == "signed", group_id=group_id, period_id=period_id, run_no=gp.run_no,
period_specific="true" if period_specific else "false", memo=memo or "",
proposed_by=caller.get("email") or "", proposed_sub=caller.get("sub") or "")
repo.groups.save(agg, current) # the group's next version: a run or a sign-off at the same moment collides
return _view(entry, repo.cons_journals.save(entry, 0), key="version")
@CONS.use_case(kind="command", route="POST /cons-journals/{cons_journal_id}/approve-cons-entry", permission="cons:journal:approve", idempotent=True, if_match="version")
def approve_cons_entry(caller: dict, cons_journal_id: str, version: Optional[int] = None) -> dict:
"""Approve the entry (cons_journal: proposed → posted). Only when: the approver is not the proposer (email AND sub) where
signoff_second_person is on; the period not signed. Then: numbered in the group's book and year, above the high-water mark (a rejected
entry takes none); a period-specific entry's reversal is due at the next period's start. A repeat changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_cons_journal(repo, caller, cons_journal_id)
_same_version(f"cons journal {cons_journal_id}", current, version)
if agg.status == "posted":
return _view(agg, current, key="version")
group, group_version = _visible_group(repo, caller, agg.group_id)
gp = the_period(group, agg.period_id)
fy = C.fiscal_year_of(agg.period_id, group.fiscal_year_end_month)
agg.approve_cons_entry(by=caller.get("email") or "", sub=caller.get("sub") or "", second_person_required=runs.settings(repo)["signoff_second_person"],
period_signed=gp.status == "signed",
claim_number=lambda: repo.data.claim_numbers(group.group_id, fy, [agg.cons_journal_id])[0])
repo.groups.save(group, group_version) # claims the group's next version: a run, a sign-off or a reopen at the same moment collides
return _view(agg, repo.cons_journals.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /cons-journals/{cons_journal_id}/reject-cons-entry", permission="cons:journal:approve", idempotent=True, if_match="version")
def reject_cons_entry(caller: dict, cons_journal_id: str, reason: str, version: Optional[int] = None) -> dict:
"""Reject the entry (cons_journal: proposed → rejected). A repeat changes nothing."""
with repository() as repo, _decide():
agg, current = _visible_cons_journal(repo, caller, cons_journal_id)
_same_version(f"cons journal {cons_journal_id}", current, version)
if agg.reject_cons_entry(reason=reason, by=_who(caller)) is None:
return _view(agg, current, key="version")
return _view(agg, repo.cons_journals.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /cons-journals/{cons_journal_id}/reverse-cons-entry", permission="cons:journal:reverse", idempotent=True, if_match="version")
def reverse_cons_entry(caller: dict, cons_journal_id: str, reason: str, version: Optional[int] = None) -> dict:
"""Reverse the entry (cons_journal: posted → reversed). Only when: the period not signed; a reason is given. Then: a mirror entry keyed
derived_id(cons_journal_id, 'reversal') is posted, numbered; neither is deleted. A correction is a reversal, never an edit. A repeat
changes nothing.
"""
with repository() as repo, _decide():
agg, current = _visible_cons_journal(repo, caller, cons_journal_id)
_same_version(f"cons journal {cons_journal_id}", current, version)
if agg.status == "reversed":
return _view(agg, current, key="version")
group, group_version = _visible_group(repo, caller, agg.group_id)
gp = the_period(group, agg.period_id)
if agg.reverse_cons_entry(reason=reason, by=_who(caller), period_signed=gp.status == "signed") is None:
return _view(agg, current, key="version")
original = {"cons_journal_id": agg.cons_journal_id, "group_id": agg.group_id, "period_id": agg.period_id, "document_type": agg.document_type,
"run_no": agg.run_no, "reverses_id": "", "lines": [dataclasses.asdict(x) for x in agg.cons_journal_lines]}
[mirror] = runs._mirror([original], agg.run_no)
mirror["period_specific"], mirror["memo"] = agg.period_specific, f"reverses {agg.cons_journal_id}: {reason}"
repo.data.post_entries([mirror], C.fiscal_year_of(agg.period_id, group.fiscal_year_end_month))
repo.groups.save(group, group_version)
return _view(agg, repo.cons_journals.save(agg, current), key="version")
def _visible_statement(repo, caller: dict, report: str, group_id: str, period_id: str, node_id: str) -> tuple:
found = repo.statements.get(report, group_id, period_id, node_id)
# TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with
# the read permission sees every row.
if found is None:
raise Refused(404, f"no statement {report} {group_id} {period_id} {node_id}")
return found
@CONS.use_case(kind="query", route="GET /statements", permission="cons:group:read")
def list_statements(caller: dict) -> dict:
"""The statements the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"statements": repo.statements.ids()}
@CONS.use_case(kind="query", route="GET /statements/{report}/{group_id}/{period_id}/{node_id}", permission="cons:group:read")
def get_statement(caller: dict, report: str, group_id: str, period_id: str, node_id: str) -> dict:
"""One statement, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_statement(repo, caller, report, group_id, period_id, node_id), key="version")
# POST /statements (the scaffold's keep_statement) is removed (audit v1 MUST-2a): a statement is materialised from the signed trial
# balance by sign-off and materialise-statements only, never written by hand.
def _visible_ai_finding(repo, caller: dict, finding_id: str) -> tuple:
found = repo.ai_findings.get(finding_id)
# TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with
# the read permission sees every row.
if found is None:
raise Refused(404, f"no ai finding {finding_id}")
return found
@CONS.use_case(kind="query", route="GET /ai-findings", permission="cons:group:read")
def list_ai_findings(caller: dict) -> dict:
"""The ai findings the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"ai_findings": repo.ai_findings.ids()}
@CONS.use_case(kind="query", route="GET /ai-findings/{finding_id}", permission="cons:group:read")
def get_ai_finding(caller: dict, finding_id: str) -> dict:
"""One ai finding, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_ai_finding(repo, caller, finding_id), key="version")
# POST /ai-findings (the scaffold's keep_ai_finding) is removed (audit v1): the spec has no such door — the AI step
# (explain_elimination_difference, suggest only) appends its findings itself, and a decision is the person's, never a field anyone sets.
def _visible_migration_run(repo, caller: dict, run_id: str) -> tuple:
found = repo.migration_runs.get(run_id)
# TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with
# the read permission sees every row.
if found is None:
raise Refused(404, f"no migration run {run_id}")
return found
@CONS.use_case(kind="query", route="GET /migration-runs", permission="cons:group:read")
def list_cons_load_runs(caller: dict) -> dict:
"""The migration runs the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"migration_runs": repo.migration_runs.ids()}
@CONS.use_case(kind="query", route="GET /migration-runs/{run_id}", permission="cons:group:read")
def get_cons_load_run(caller: dict, run_id: str) -> dict:
"""One migration run, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_migration_run(repo, caller, run_id), key="version")
# POST /migration-runs (the scaffold's keep_cons_load_run) is removed (audit v1): it could set status=signed with no checks. A load
# goes through POST /cons-migration-runs, its reconcile and its signature.
def _visible_migration_map(repo, caller: dict, odoo_db: str, odoo_model: str, odoo_id: str) -> tuple:
found = repo.migration_maps.get(odoo_db, odoo_model, odoo_id)
# TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with
# the read permission sees every row.
if found is None:
raise Refused(404, f"no migration map {odoo_db} {odoo_model} {odoo_id}")
return found
@CONS.use_case(kind="query", route="GET /migration-maps", permission="cons:group:read")
def list_cons_id_maps(caller: dict) -> dict:
"""The migration maps the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"migration_maps": repo.migration_maps.ids()}
@CONS.use_case(kind="query", route="GET /migration-maps/{odoo_db}/{odoo_model}/{odoo_id}", permission="cons:group:read")
def get_cons_migration_map(caller: dict, odoo_db: str, odoo_model: str, odoo_id: str) -> dict:
"""One migration map, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_migration_map(repo, caller, odoo_db, odoo_model, odoo_id), key="version")
# POST /migration-maps (the scaffold's keep_cons_id_map) is removed (audit v1): the loader (CON-34) is the only writer of the Odoo -> our-key
# map, through repo.migration.write_maps inside a load under cons:migration:load.
def loader_doors(agg) -> dict:
return json.loads(agg.doors_json)
def _visible_cons_migration_run(repo, caller: dict, cons_migration_run_id: str) -> tuple:
found = repo.cons_migration_runs.get(cons_migration_run_id)
# TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with
# the read permission sees every row.
if found is None:
raise Refused(404, f"no cons migration run {cons_migration_run_id}")
return found
@CONS.use_case(kind="query", route="GET /cons-migration-runs", permission="cons:group:read")
def list_cons_migration_runs(caller: dict) -> dict:
"""The cons migration runs the caller may see. Ids only; get one for the rest."""
with repository() as repo:
return {"cons_migration_runs": repo.cons_migration_runs.ids()}
@CONS.use_case(kind="query", route="GET /cons-migration-runs/{cons_migration_run_id}", permission="cons:group:read")
def get_cons_migration_run(caller: dict, cons_migration_run_id: str) -> dict:
"""One cons migration run, whole, with its version, if the caller may see it."""
with repository() as repo:
return _view(*_visible_cons_migration_run(repo, caller, cons_migration_run_id), key="version")
@CONS.use_case(kind="command", route="POST /cons-migration-runs", permission="cons:migration:load", idempotent=True)
def load_cons_from_odoo(caller: dict, cons_migration_run_id: str, doors_json: str = "", checks_json: str = "") -> dict:
"""Load Odoo's consolidation (cons_migration_run: none → loaded). Only when: the doors decided; gl's
migration signed (companies and accounts mapped). Then: group, structure per period, account rules and
closed periods' consolidation journals written read-only, keyed on the Odoo record. doors_json names the group, the closed
periods to carry and the decision per company below 100% (full, equity or none); checks_json is Odoo's side of each reconcile check
(consolidated_balance, structure, journals, proportional). Making the same one again returns it.
"""
# the permission is not in the spec's catalog — add it there.
with repository() as repo, _decide():
found = repo.cons_migration_runs.get(cons_migration_run_id)
if found is not None:
existing = _view(*_visible_cons_migration_run(repo, caller, cons_migration_run_id), key="version")
if found is not None:
if found[0].status != "loaded" or not loader.wired():
return existing # a signed, reconciled or discarded run is never loaded again
with _decide():
loader.load(repository, cons_migration_run_id, loader_doors(found[0])) # a load that stopped resumes where it stopped
return existing
plan = None
with _decide():
doors = ConsMigrationRun.parse_doors_of(doors_json)
if loader.wired():
plan = loader.preflight(cons_migration_run_id, doors) # gl's migration signed (mapped) BEFORE the run exists
checks_json = checks_json or json.dumps(loader.odoo_checks(plan, doors))
with repository() as repo, _decide():
agg = ConsMigrationRun.load_cons_from_odoo(cons_migration_run_id, by=_who(caller), doors_json=doors_json, checks_json=checks_json)
view = _view(agg, repo.cons_migration_runs.save(agg, 0), key="version")
if plan is not None:
with _decide():
loader.load(repository, cons_migration_run_id, doors, plan)
return view
@CONS.use_case(kind="command", route="POST /cons-migration-runs/{cons_migration_run_id}/reconcile-cons-migration", permission="cons:migration:reconcile", idempotent=True, if_match="version")
def reconcile_cons_migration(caller: dict, cons_migration_run_id: str, version: Optional[int] = None) -> dict:
"""Reconcile against Odoo (cons_migration_run: loaded → reconciled). Then: every migration.odoo.reconcile
check computed on both sides. A repeat changes nothing.
"""
# the permission is not in the spec's catalog — add it there.
with repository() as repo, _decide():
agg, current = _visible_cons_migration_run(repo, caller, cons_migration_run_id)
_same_version(f"cons migration run {cons_migration_run_id}", current, version)
if agg.reconcile_cons_migration(migration.ours(repo, agg) if agg.doors_json else {}, by=_who(caller)) is None:
return _view(agg, current, key="version")
return _view(agg, repo.cons_migration_runs.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /cons-migration-runs/{cons_migration_run_id}/explain-cons-difference", permission="cons:settings:write", if_match="version")
def explain_cons_difference(caller: dict, cons_migration_run_id: str, reason: str, check: str = "", version: Optional[int] = None) -> dict:
"""Explain each difference (cons_migration_run: reconciled → reconciled). Only when: the check differs and may be explained (a
company consolidated proportionally in Odoo and fully or by equity here); a balance, structure or journal check must be equal and is
never explained. `check` names one; left out, every differing check that may be explained is.
"""
# the permission is not in the spec's catalog — add it there.
with repository() as repo, _decide():
agg, current = _visible_cons_migration_run(repo, caller, cons_migration_run_id)
_same_version(f"cons migration run {cons_migration_run_id}", current, version)
if agg.explain_cons_difference(reason=reason, by=_who(caller), check=check, at=work.now()) is None:
return _view(agg, current, key="version")
return _view(agg, repo.cons_migration_runs.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /cons-migration-runs/{cons_migration_run_id}/reload-cons-migration", permission="cons:migration:load", if_match="version")
def reload_cons_migration(caller: dict, cons_migration_run_id: str, version: Optional[int] = None) -> dict:
"""Load again after a fix (cons_migration_run: loaded, reconciled → loaded). Only when: not signed. Odoo is read again — its side of every
check, and its journals: what is missing is written by the same keys, a journal whose Odoo amounts changed is superseded by its next version.
A signed run is never touched."""
# the permission is not in the spec's catalog — add it there.
plan = odoo = None
with repository() as repo, _decide():
agg, current = _visible_cons_migration_run(repo, caller, cons_migration_run_id)
_same_version(f"cons migration run {cons_migration_run_id}", current, version)
if loader.wired() and agg.doors_json and agg.status in ("loaded", "reconciled"):
plan = loader.preflight(cons_migration_run_id, loader_doors(agg)) # Odoo (and gl) read again BEFORE anything is written
odoo = loader.odoo_checks(plan, loader_doors(agg)) # Odoo's side of every check, as Odoo reports it now
if agg.reload_cons_migration(by=_who(caller), odoo=odoo) is None:
return _view(agg, current, key="version")
view = _view(agg, repo.cons_migration_runs.save(agg, current), key="version")
doors = loader_doors(agg)
if plan is not None:
with _decide():
loader.load(repository, cons_migration_run_id, doors, plan, refresh=True) # a journal Odoo corrected is superseded
return view
@CONS.use_case(kind="command", route="POST /cons-migration-runs/{cons_migration_run_id}/sign-cons-reconciliation", permission="cons:migration:sign", idempotent=True, if_match="version")
def sign_cons_reconciliation(caller: dict, cons_migration_run_id: str, version: Optional[int] = None) -> dict:
"""Sign the reconciliation (cons_migration_run: reconciled → signed). Only when: every check equal, or
explained where allowed. Then: who, when and checksums recorded. A repeat changes nothing.
"""
# the permission is not in the spec's catalog — add it there.
with repository() as repo, _decide():
agg, current = _visible_cons_migration_run(repo, caller, cons_migration_run_id)
_same_version(f"cons migration run {cons_migration_run_id}", current, version)
if agg.sign_cons_reconciliation(by=caller.get("email") or "", sub=caller.get("sub") or "", at=work.now()) is None:
return _view(agg, current, key="version")
return _view(agg, repo.cons_migration_runs.save(agg, current), key="version")
@CONS.use_case(kind="command", route="POST /cons-migration-runs/{cons_migration_run_id}/discard-cons-migration", permission="cons:migration:sign", idempotent=True, if_match="version")
def discard_cons_migration(caller: dict, cons_migration_run_id: str, reason: str, version: Optional[int] = None) -> dict:
"""Discard the run (cons_migration_run: loaded, reconciled → discarded). A repeat changes nothing."""
# the permission is not in the spec's catalog — add it there.
with repository() as repo, _decide():
agg, current = _visible_cons_migration_run(repo, caller, cons_migration_run_id)
_same_version(f"cons migration run {cons_migration_run_id}", current, version)
if agg.discard_cons_migration(reason=reason, by=_who(caller)) is None:
return _view(agg, current, key="version")
return _view(agg, repo.cons_migration_runs.save(agg, current), key="version")
CONTROLS = ("signoff_second_person", "allow_open_period_collect", "ai.explain_elimination_difference.mode") # the owner's: cons:settings:controls
@CONS.use_case(kind="query", route="GET /settings", permission="cons:group:read")
def list_cons_settings(caller: dict) -> dict:
"""The settings the customer edits (module.yaml config_rows), one row per setting with its kind and the value in force on the business
date; a value of null is the reference behaviour. `control` marks the owner's controls."""
with repository() as repo:
return {"today": clock.today(), "settings": [{**s, "control": s["setting_key"] in CONTROLS} for s in repo.settings.all(clock.today())]}
@CONS.use_case(kind="query", route="GET /settings/{setting_key}/history", permission="cons:group:read")
def get_cons_setting_history(caller: dict, setting_key: str) -> dict:
"""Every dated row of one setting, oldest first, with who set it and when. Rows are only ever added, so this is the whole history."""
with repository() as repo:
if repo.settings.get(setting_key) is None:
raise Refused(404, f"no setting {setting_key}")
return {"setting_key": setting_key, "control": setting_key in CONTROLS, "rows": repo.settings.history(setting_key)}
def _dated_setting(repo, caller: dict, setting_key: str, value, valid_from: Optional[date]) -> dict:
"""A new dated row, from tomorrow by default. A row dated today or earlier is refused, the first row of a key excepted (config_rows)."""
row = repo.settings.get(setting_key)
if row is None:
raise Refused(404, f"no setting {setting_key}")
today = clock.today()
first = not repo.settings.history(setting_key)
start = valid_from or today + timedelta(days=1)
if start < today or (start == today and not first):
raise Refused(409, f"{setting_key} takes effect after today ({today}, the business date in {clock.business_timezone()}): date the row "
f"{today + timedelta(days=1)} or later" + ("" if first else " (only the first row of a key may start today)"),
gate="dated_from_tomorrow", fix=f"set valid_from to {today + timedelta(days=1)} or leave it out")
try:
repo.settings.add(setting_key, start, value, caller.get("email") or "", caller.get("sub") or "", work.now())
except ValueError as e:
raise Refused(422, f"{setting_key} is {e}") from None
except IntegrityError:
raise Refused(409, f"{setting_key} already has a row from {start}: rows are never edited, date the new one later") from None
return {"setting_key": setting_key, "value": value, "valid_from": start, "set_by": _who(caller)}
@CONS.use_case(kind="command", route="POST /settings/{setting_key}", permission="cons:settings:write", idempotent=True)
def set_cons_setting(caller: dict, setting_key: str, value: Optional[Union[bool, int, str]] = None, valid_from: Optional[date] = None) -> dict:
"""Change one day-to-day setting as a new dated row (from tomorrow unless a later date is given), typed by its kind. null puts the
reference behaviour back. The owner's controls (signoff_second_person, allow_open_period_collect, the AI step mode) are refused here: whoever
holds cons:settings:write could otherwise switch off the second person for their own entries. They are set with set_cons_control."""
with repository() as repo:
if setting_key in CONTROLS:
raise Refused(403, f"{setting_key} is a control on the people who consolidate: it is set by the owner (cons:settings:controls), not with "
"cons:settings:write", gate="settings_control", fix="ask the owner to set it (set_cons_control)")
return _dated_setting(repo, caller, setting_key, value, valid_from)
@CONS.use_case(kind="command", route="POST /controls/{setting_key}", permission="cons:settings:controls", idempotent=True)
def set_cons_control(caller: dict, setting_key: str, value: Optional[Union[bool, int, str]] = None, valid_from: Optional[date] = None) -> dict:
"""The owner changes a control on the group close (signoff_second_person, allow_open_period_collect, the AI step mode) as a new dated
row. It takes effect from tomorrow (the business date in the business timezone), never today or earlier; the first row of a key may
start today. The history of every row is readable (get_cons_setting_history)."""
with repository() as repo:
if setting_key not in CONTROLS:
raise Refused(409, f"{setting_key} is not a control: it is set with set_cons_setting (cons:settings:write)", gate="not_a_control")
return _dated_setting(repo, caller, setting_key, value, valid_from)
# cons_audit: append-only history, appended by the aggregates' saves — no command writes it.
# Reading it is a query you add with its row visibility (TODO(visibility)).