dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/infrastructure/seams.py
"""The neighbour this brick reads: gl, behind one small PORT shaped from gl's signed spec (module.yaml seams `with: consolidation`).
companies() GET /companies the companies gl keeps, with their functional currency
period_status(company, day) GET /periods/open the gate company_periods_closed
balances(company, ledger_id, period_id) GET /group-reporting/balances one company's figures by group account, with gl's digest
fx_rate(currency, on, kind, to) GET /fx-rates/{currency} closing | average | spot, gl's rates only (no rate table here)
reconciliation(company, partner, as_at) GET /intercompany/reconciliation gl's matched items and breaks for a pair (None: no match was run)
chart_accounts(chart_id) GET /charts/{chart_id}/accounts the group chart's accounts
statement_version(fsv_id, version_no) GET /statement-versions/{fsv_id} the published version of a statement (nodes and assignments)
Nothing here is imported by the domain. A neighbour that is not part of the installation, refuses or does not answer is a `Missing`
with its own reason (and fix, when gl gave one) — never a guess. Tests replace `ports.gl` with tests/fakes.py's FakeGl.
"""
import json
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.seam import Seam
GL = Seam("gl", "GL_BASE_URL", roles=["service"],
permissions=["gl:group_data:extract", "gl:company:read", "gl:fx:read", "gl:period:read", "gl:report:read", "gl:migration:read", "gl:journal:read"])
MAP_CHUNK = 500 # ids per GET /migration-map call: one request per model and chunk, never one per id
class Missing(Exception):
"""gl could not answer. `reason` is its own words (with its fix); `status` what it said, 0 if nothing."""
def __init__(self, what: str, reason: str, status: int = 0, installed: bool = True):
super().__init__(f"{what}: {reason}")
self.what, self.reason, self.status, self.installed = what, reason, status, installed
def _reason(raw) -> str:
try:
d = json.loads(raw) if isinstance(raw, str) else raw
if isinstance(d, dict):
text = str(d.get("reason") or d.get("detail") or raw)
return (text + (f" — fix: {d['fix']}" if d.get("fix") else ""))[:400]
except (ValueError, TypeError):
pass
return str(raw)[:300]
def _rows(data: dict, what: str) -> list:
"""A list answer: the spec says the body IS the list (the Seam wraps it as {"body": [...]}). Any other shape is refused rather than read as an
empty list."""
if isinstance(data.get("body"), list):
return data["body"]
raise Missing(what, f"gl answered something that is not the list the spec promises (keys: {', '.join(sorted(data)) or 'none'})")
def _ask(reply, what: str):
if reply.ok:
return reply.data
raise Missing(what, _reason(reply.reason), reply.status, installed="not installed" not in reply.reason)
# ── the requests: the real adapter sends these, the fake checks the same ones against gl's spec ───────────────────
def balances_params(company: str, ledger_id: str, period_id: str) -> dict:
return {"company_code": company, "ledger_id": ledger_id, "period_id": period_id}
def period_params(company: str, day: date) -> dict:
return {"date": day.isoformat(), "company_code": company}
def fx_params(on: date, kind: str, to: str) -> dict:
"""gl v6: `on` answers the newest rate of the type dated on or before it; a monthly average is addressed by month (type=average,
month=YYYY-MM instead of `on`), which answers exactly that month's average or 404s - never a nearer one."""
if kind == "average":
return {"month": f"{on.year:04d}-{on.month:02d}", "type": kind, "to": to}
return {"on": on.isoformat(), "type": kind, "to": to}
def reconciliation_params(company: str, partner: str, as_at: date) -> dict:
return {"company_code": company, "partner_company": partner, "as_at": as_at.isoformat()}
def company_params(on: Optional[date]) -> Optional[dict]:
"""gl v7: GET /companies/{company_code}?on= answers statutory_tax_rate as of `on` (a period's last day)."""
return {"on": on.isoformat()} if on else None
def lines_params(company: str, ledger_id: str, frm: date, to: date, group_account_id: str, limit: int, offset: int) -> dict:
"""gl v7 GL-04, exactly: from and to (never date_from / date_to), the group account in the path AND as group_account_id."""
return {"company_code": company, **({"ledger_id": ledger_id} if ledger_id else {}), "from": frm.isoformat(), "to": to.isoformat(),
"group_account_id": str(group_account_id), "limit": limit, "offset": offset}
def status_params(db: str) -> dict:
return {"db": db}
def map_params(ids) -> dict:
return {"ids": ",".join(str(i) for i in ids)}
def version_params(version_no) -> dict:
return {"version_no": int(version_no)} if version_no not in (None, "") else {}
class GlPort:
def companies(self) -> list:
raise NotImplementedError
def period_status(self, company: str, day: date) -> dict:
raise NotImplementedError
def balances(self, company: str, ledger_id: str, period_id: str) -> dict:
raise NotImplementedError
def fx_rate(self, currency: str, on: date, kind: str, to: str) -> Decimal:
raise NotImplementedError
def reconciliation(self, company: str, partner: str, as_at: date) -> Optional[dict]:
raise NotImplementedError
def chart_accounts(self, chart_id: str) -> list:
raise NotImplementedError
def statement_version(self, fsv_id: str, version_no=None) -> dict:
raise NotImplementedError
def migration_map(self, db: str, model: str, ids) -> dict:
"""{odoo_id (str): our natural key} for the ids gl's migration carried; an id not migrated is absent. One IN-list per MAP_CHUNK ids."""
raise NotImplementedError
def company_row(self, company_code: str, on: Optional[date] = None) -> dict:
"""GET /companies/{company_code}?on=: {..., statutory_tax_rate (string fraction or null, as of `on`), business_timezone}; Missing 404 when unknown."""
raise NotImplementedError
def migration_status(self, db: str) -> dict:
"""GET /migration/status?db= : {db, run_id, status (none | loaded | reconciled | signed), cutover_date, ...}; never a 404."""
raise NotImplementedError
def account_lines(self, group_account_id: str, company: str, ledger_id: str, frm: date, to: date, limit: int, offset: int, caller: Optional[dict] = None) -> dict:
"""GET /accounts/{group_account_id}/lines under the CALLER (gl's company scope applies): the lines behind a group figure, local currency."""
raise NotImplementedError
class RealGl(GlPort):
def companies(self) -> list:
return _rows(_ask(GL.get("/companies", actor="consolidation"), "gl"), "gl companies")
def period_status(self, company, day):
return _ask(GL.get("/periods/open", params=period_params(company, day), actor="consolidation"), "gl")
def balances(self, company, ledger_id, period_id):
return _ask(GL.get("/group-reporting/balances", params=balances_params(company, ledger_id, period_id), actor="consolidation"),
"gl")
def fx_rate(self, currency, on, kind, to):
if currency == to:
return Decimal("1.000000")
data = _ask(GL.get(f"/fx-rates/{currency}", params=fx_params(on, kind, to), actor="consolidation"), "gl")
# gl answers the rate IN FORCE: the newest one dated on or before the day. For a closing rate that must be the day itself and for
# an average the month itself - an older month's average is a different rate, and translating with it would be a wrong figure.
got = date.fromisoformat(str(data.get("rate_date") or on)[:10])
if (kind == "closing" and got != on) or (kind == "average" and (got.year, got.month) != (on.year, on.month)):
raise Missing("gl", f"fx_rate: no {kind} rate for {currency} on {on} — gl answered the rate of {got}, which is not this "
f"{'month' if kind == 'average' else 'date'}'s; enter the {kind} rate dated {on} in gl", 404)
return Decimal(str(data["rate"]))
def reconciliation(self, company, partner, as_at):
r = GL.get("/intercompany/reconciliation", params=reconciliation_params(company, partner, as_at), actor="consolidation")
if not r.ok and r.status == 404:
return None
return _ask(r, "gl")
def chart_accounts(self, chart_id):
return _rows(_ask(GL.get(f"/charts/{chart_id}/accounts", actor="consolidation"), "gl"), "gl chart accounts")
def statement_version(self, fsv_id, version_no=None):
return _ask(GL.get(f"/statement-versions/{fsv_id}", params=version_params(version_no), actor="consolidation"), "gl")
def migration_map(self, db, model, ids):
out: dict = {}
wanted = sorted({str(i) for i in ids})
for n in range(0, len(wanted), MAP_CHUNK):
data = _ask(GL.get(f"/migration-map/{db}/{model}", params=map_params(wanted[n:n + MAP_CHUNK]), actor="consolidation"), "gl")
for row in _rows(data, "gl migration map"):
out[str(row["odoo_id"])] = str(row["ours"])
return out
def company_row(self, company_code, on=None):
return _ask(GL.get(f"/companies/{company_code}", params=company_params(on), actor="consolidation"), "gl")
def migration_status(self, db):
return _ask(GL.get("/migration/status", params=status_params(db), actor="consolidation"), "gl")
def account_lines(self, group_account_id, company, ledger_id, frm, to, limit, offset, caller=None):
return _ask(GL.get(f"/accounts/{group_account_id}/lines", params=lines_params(company, ledger_id, frm, to, group_account_id, limit, offset),
actor="consolidation", caller=caller), "gl")
class FilePort:
"""Where an uploaded file is kept (`raw/consolidation/<group>/<period>/...`). erp-platform has no object-store helper yet: the real
adapter says so rather than pretending; a composition sets `ports.files` to its DataK3 bucket's S3 endpoint."""
def put(self, key: str, data: bytes, sha256: str) -> None:
raise NotImplementedError
class NoStore(FilePort):
def put(self, key, data, sha256):
raise Missing("object store", f"no object store is configured for {key} (neighbour gap: erp-platform has no object-storage helper)")
class OdooUnavailable(Exception):
pass
class OdooReader:
"""Odoo 16 Enterprise `account_consolidation`, read through one port (module.yaml migration.odoo.reads). Nothing is ever written to Odoo.
Rows are plain dicts with the spec's field names; amounts are Decimal, dates `date`. The loader reads in chunks: `journals` by keyset
(id above `after`), `journal_lines` for the ids of one chunk. `aggregates` is ODOO's OWN read_group (so the reconcile compares what Odoo
says with what was loaded, never the loaded rows with themselves): {"balance": {period_id: {odoo account id: Decimal}},
"structure": {period_id: {odoo company id: rate_consolidation}}, "journals": {period_id: {"count": int, "total": Decimal}}}."""
wired = True
def chart(self, chart_id): raise OdooUnavailable("no Odoo reader is wired") # {id, name, currency, company_ids}
def accounts(self, chart_id): raise OdooUnavailable("no Odoo reader is wired") # [{id, code, name, currency_mode, account_ids, invert_sign}]
def periods(self, chart_id): raise OdooUnavailable("no Odoo reader is wired") # closed: [{id, date_analysis_begin, date_analysis_end, state}]
def company_periods(self, period_ids): raise OdooUnavailable("no Odoo reader is wired") # [{id, period_id, company_id, rate_consolidation}]
def journals(self, period_ids, after, limit): raise OdooUnavailable("no Odoo reader is wired") # [{id, period_id, company_period_id, auto_generated, name}]
def journal_lines(self, journal_ids): raise OdooUnavailable("no Odoo reader is wired") # [{id, journal_id, account_id, amount, note}]
def aggregates(self, chart_id, period_ids): raise OdooUnavailable("no Odoo reader is wired") # see the class
class NoOdoo(OdooReader):
wired = False
class Ports:
"""The one place a use case reaches a neighbour from. Replace an attribute to swap a neighbour (tests do)."""
def __init__(self):
self.gl: GlPort = RealGl()
self.files: FilePort = NoStore()
self.odoo = NoOdoo()
def gl_installed(self) -> bool:
return isinstance(self.gl, RealGl) is False or GL.installed
ports = Ports()