dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/ledgerops.py
"""The shared mechanics of the books: write one posted journal (number, chain, lines, split lines, period guard, trail),
re-derive balances from split lines, and the closing figures of a company's ledger.
A posted journal is written by `write_posted` and by nothing else: that is what makes the period lock, the numbers and the
hash chain real on an engine with no cross-table constraints. It runs in the caller's transaction; the balances are then
re-derived in a SECOND committed transaction (`derive_balances`), because there is no read-your-writes inside an open one.
"""
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from sqlalchemy import update
from brick_gl.application.brick import RepoLookups, real_now
from brick_gl.domain import dates, hashchain, splitting
from brick_gl.domain.journal import Built
from brick_gl.domain.money import ZERO
from brick_gl.domain.readiness import refusal
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import utc
HEADER_DEFAULTS = {"ledger_id": None, "adjustment": False, "reverse_next_period": False, "reversal_status": "none", "ic_document_id": None,
"recurring_id": None, "accrual_id": None, "allocation_run_id": None, "reval_run_id": None, "proposed_by": None,
"proposed_sub": None, "approved_by": None, "approved_sub": None, "reverses_journal_id": None,
"migration_run_id": None, "memo": "", "reversal_reason": None, "rejected_by": None, "reject_reason": None}
def splitting_on(repo, company_code: str, on: date) -> bool:
"""Document splitting is on for a company's year when its control row says so on the journal's day."""
return repo.settings.in_force(on, company_code).document_splitting == "on"
def period_for(repo, company_code: str, on: date, *, closing_ok: bool = False, adjustment: bool = False) -> dict:
"""The period a journal dated `on` posts into. A closed period (or none) refuses an operational posting, naming the period;
an audit adjustment posts into the CLOSED last period of an OPEN year (GL-56)."""
p = repo.periods.for_date(company_code, on)
if p is None:
raise refusal("period_open", f"no period of {company_code} covers {on}", "post it dated in an open period, or ask the "
"accountant to open the year")
if p["status"] == "closed" and not (closing_ok or adjustment):
raise refusal("period_open", f"the period {p['period_id']} of {company_code} is closed",
"post it dated in an open period, or ask the accountant to reopen")
return p
def assert_period_still_open(repo, company_code: str, period_id: str, at: datetime, *, adjustment: bool = False) -> None:
"""Write the period row in the posting's own transaction (last_posted_at) ONLY while it is open: a close that committed
after this posting read the period leaves no row to update, and the posting is refused naming the period. A close racing
a posting on DataK3 conflicts on this same row and the loser retries."""
cond = (t.gl_period.c.company_code == company_code) & (t.gl_period.c.period_id == period_id)
if not adjustment:
cond = cond & (t.gl_period.c.status == "open")
res = repo.conn.execute(update(t.gl_period).where(cond).values(last_posted_at=utc(at), updated_at=utc(at)))
if res.rowcount == 0:
raise refusal("period_open", f"the period {period_id} of {company_code} is closed",
"post it dated in an open period, or ask the accountant to reopen")
def write_posted(repo, ctx: RepoLookups, company: dict, hdr: dict, built: Built, *, existing: Optional[dict], at: datetime,
splits: Optional[list] = None, soft_closed: bool = False) -> dict:
"""Write a posted journal in the caller's transaction. `hdr` names the journal; `existing` is the PROPOSED row it was
waiting as (updated in place), else it is new (a plain INSERT: a conflict on journal_id is the replay or a race lost)."""
hdr = {**HEADER_DEFAULTS, **hdr}
code, on = hdr["company_code"], hdr["journal_date"]
jid = hdr["journal_id"]
assert_period_still_open(repo, code, hdr["period_id"], at, adjustment=bool(hdr["adjustment"]) or soft_closed)
on_flag = splitting_on(repo, code, on)
chart_accounts = {}
def subtype_of(account_id: int) -> str:
if account_id not in chart_accounts:
a = ctx.account(account_id)
chart_accounts[account_id] = a.subtype if a else ""
return chart_accounts[account_id]
def seg(pc: str) -> str:
return ctx.segment_of(pc, on)
if splits is None:
splits = splitting.split(built.lines, subtype_of, splitting=on_flag, default_pc=company.get("default_profit_centre") or "",
segment_of=seg, passive=lambda doc, acc: repo.ledger.passive_weights(doc, acc) or None,
clearing_account=lambda: ctx.mapped("split_clearing", "", on))
entry_no = repo.numbers.claim(code, hdr["book"], hdr["fiscal_year"], jid, at)
seq, prev = repo.chain.claim(code, jid, at)
row = {**hdr, "status": "posted", "entry_no": entry_no, "chain_seq": seq, "prev_hash": prev, "total": built.total,
"functional_total": built.f_total, "fx_rate": built.fx_rate, "posted_at": utc(at)}
line_rows = [{"journal_id": jid, "line_no": ln.line_no, "account_id": ln.account_id, "purpose": ln.purpose or None,
"qualifier": ln.qualifier or None, "debit": ln.debit, "credit": ln.credit, "functional_debit": ln.f_debit,
"functional_credit": ln.f_credit, "currency": ln.currency or None, "amount_currency": ln.amount_currency,
"bp_id": ln.bp_id, "tax_code": ln.tax_code or None, "cost_centre": ln.cost_centre or None,
"profit_centre": ln.profit_centre or None, "partner_company": ln.partner_company or None,
"movement_type": ln.movement_type or None, "against_document": ln.against_document or None,
"ic_reference": ln.ic_reference or None, "asset_id": ln.asset_id or None, "line_memo": ln.memo or None,
"odoo_line_id": ln.odoo_line_id or None} for ln in built.lines]
ledger_id = hdr["ledger_id"]
split_rows = [{
"journal_id": jid, "line_no": s.line_no, "split_no": s.split_no, "company_code": code, "ledger_id": ledger_id,
"account_id": s.account_id, "period_id": hdr["period_id"], "journal_date": on, "debit": s.debit, "credit": s.credit,
"functional_debit": s.f_debit, "functional_credit": s.f_credit, "currency": hdr["currency"],
"profit_centre": s.profit_centre or None, "segment": s.segment or None, "cost_centre": s.cost_centre or None,
"partner_company": s.partner_company or None, "movement_type": s.movement_type or None, "bp_id": s.bp_id,
"asset_id": s.asset_id or None, "is_clearing": s.is_clearing, "adjustment": bool(hdr["adjustment"])} for s in splits]
row["row_hash"] = hashchain.row_hash(prev, row, line_rows, split_rows)
if existing is None:
repo.journals.add(row)
repo.journals.add_lines(line_rows)
else:
repo.journals.set((jid,), **{k: v for k, v in row.items() if k != "journal_id"})
repo.journals.put_lines(line_rows)
repo.journals.add_splits(split_rows)
return {"journal": row, "lines": line_rows, "splits": splits}
def _sums_and_adjustments(repo, company_code: str, ledger_id: str, period_id: str, account_ids: list) -> tuple:
"""The split-line sums per (account, profit centre, segment, partner) and those of the adjustment lines alone, from ONE grouped
read (grouped by the adjustment flag too, added up here: exact decimals)."""
dims = ("account_id", "profit_centre", "segment", "partner_company")
got = repo.ledger.sums(company_code, ledger_id, period_id=period_id, account_ids=account_ids, group_by=(*dims, "adjustment"))
totals: dict = {}
adj_by: dict = {}
money = ("debit", "credit", "f_debit", "f_credit")
for r in got:
key = (r["account_id"], r["profit_centre"] or "", r["segment"] or "", r["partner_company"] or "")
for into in ([totals] + ([adj_by] if r["adjustment"] is True or r["adjustment"] == 1 else [])):
have = into.get(key)
if have is None:
into[key] = {**{d: r[d] for d in dims}, **{m: r[m] for m in money}}
else:
for m in money:
have[m] = have[m] + r[m]
return list(totals.values()), adj_by
def derive_balances(repo, company_code: str, period_id: str, account_ids: list) -> int:
"""Re-derive the balances of the accounts a journal touched, for its period and every ledger of the company: the SUM of
posted split lines (a common journal counts in every ledger, a ledger-specific one in its own) - never an increment, so
N postings landing together land once each. Run in a SECOND committed transaction after the lines commit."""
p = repo.periods.get(company_code, period_id)
if p is None:
return 0
year = dates.fy_number(p["fiscal_year"])
ledgers = repo.company_ledgers.assigned(company_code, year)
derived = real_now() # an audit stamp: real time, naive UTC
n = 0
for ledger_id in ledgers:
rows, adj_by = _sums_and_adjustments(repo, company_code, ledger_id, period_id, account_ids)
out = []
for r in rows:
key = (r["account_id"], r["profit_centre"] or "", r["segment"] or "", r["partner_company"] or "")
a = adj_by.get(key)
out.append({"company_code": company_code, "ledger_id": ledger_id, "account_id": r["account_id"], "period_id": period_id,
"profit_centre": key[1], "segment": key[2], "partner_company": key[3], "debit": r["f_debit"],
"credit": r["f_credit"], "closing": r["f_debit"] - r["f_credit"],
"closing_currency": r["debit"] - r["credit"],
"adjustment_debit": a["f_debit"] if a else ZERO, "adjustment_credit": a["f_credit"] if a else ZERO,
"derived_at": derived})
repo.ledger.put_balances(out)
n += len(out)
return n
def ledger_ids_of_journal(repo, company_code: str, period_id: str) -> list:
p = repo.periods.get(company_code, period_id)
return repo.company_ledgers.assigned(company_code, dates.fy_number(p["fiscal_year"])) if p else []
def company_view(c: dict) -> dict:
return {k: c[k] for k in ("company_code", "name", "country", "functional_currency", "chart_id", "bp_id",
"financial_year_end_month")}
def refuse_unknown(what: str, key) -> Refused:
return Refused(404, f"no {what} {key}")
def money_str(v: Decimal) -> str:
return f"{v:.2f}"
def tb_digest(repo, company: dict, period: dict) -> str:
"""A digest of the company's trial balance through the end of a period, per ledger: the sign-off of a close records it, and a
posting that lands in the period afterwards changes it (the sign-off goes stale)."""
import hashlib
rows = []
for ledger_id in repo.company_ledgers.assigned(company["company_code"], dates.fy_number(period["fiscal_year"])):
for r in repo.ledger.sums(company["company_code"], ledger_id, upto=period["ends_on"], group_by=("account_id",)):
rows.append([ledger_id, r["account_id"], str(r["f_debit"]), str(r["f_credit"])])
return hashlib.sha256(str(sorted(rows)).encode()).hexdigest()