dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/posting.py
"""Posting: what other modules send (POST /journal-entry/journals), manual journals with a second person, and reversals.
ONE path writes a journal (ledgerops.write_posted), so the period lock, the numbers and the hash chain hold for every module.
A source posting is IDEMPOTENT by its source: journal_id = derived_id(source_document, event[, ledger_id]), so the same call
again answers the journal already posted (replay), a different body for the same id is refused 409, and two callers racing for
one id land one journal (the loser's plain INSERT of the key fails and, retried, finds the winner).
"""
import uuid
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from erp_platform.auth import has_permission as auth_has
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from erp_platform.types import BigId, text
from brick_gl.application import ledgerops
from brick_gl.application.brick import (
ADJUSTMENT_POST,
GL,
JOURNAL_APPROVE,
JOURNAL_READ,
JOURNAL_WRITE,
PERIOD_CLOSE,
POST,
REVERSAL_APPROVE,
RepoLookups,
can_see,
claims,
company_row,
decide,
real_now,
repository,
same_person,
sub_of,
today,
visible_codes,
who,
)
from brick_gl.application.inputs import CompanyCode, Currency, FxRate, IsoDate, LedgerCode, Limit, Lines, Memo, Reason
from brick_gl.domain import dates, splitting
from brick_gl.domain import journal as jr
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.journal import Built, Draft, Line, LineIn
from brick_gl.domain.money import ZERO
from brick_gl.domain.readiness import met, refusal, unmet
from brick_gl.infrastructure import tables as t
# ── from a caller's lines to a draft ─────────────────────────────────────────────────────────────────────────────────
def to_lines(models: list) -> list:
out = []
for m in models:
out.append(LineIn(account_id=m.account_id, purpose=m.purpose, qualifier=m.qualifier, tax_code=m.tax_code,
bank_account_id=m.bank_account_id, asset_class=m.asset_class, debit=m.debit or ZERO,
credit=m.credit or ZERO, memo=m.memo or "", bp_id=m.bp_id, cost_centre=m.cost_centre,
profit_centre=m.profit_centre, partner_company=m.partner_company, movement_type=m.movement_type,
against_document=m.against_document, ic_reference=m.ic_reference, asset_id=m.asset_id,
company_code=getattr(m, "company_code", None)))
return out
def lines_of_row(rows: list) -> list:
"""LineIn from stored journal_line rows (a proposed journal rebuilt on the day it posts)."""
out = []
for r in rows:
purpose = r.get("purpose")
out.append(LineIn(account_id=None if purpose else r["account_id"], purpose=purpose, qualifier=r.get("qualifier"),
tax_code=r.get("tax_code"), debit=r["debit"] or ZERO, credit=r["credit"] or ZERO,
memo=r.get("line_memo") or "", bp_id=r.get("bp_id"), cost_centre=r.get("cost_centre"),
profit_centre=r.get("profit_centre"), partner_company=r.get("partner_company"),
movement_type=r.get("movement_type"), against_document=r.get("against_document"),
ic_reference=r.get("ic_reference"), asset_id=r.get("asset_id"), currency=r.get("currency"),
amount_currency=r.get("amount_currency"), odoo_line_id=r.get("odoo_line_id")))
return out
def fx_for(repo, company: dict, currency: str, given: Optional[Decimal], on: date) -> Optional[Decimal]:
"""The rate to the company's functional currency: the document's frozen one, else the spot rate in force on its date."""
if currency == company["functional_currency"]:
return None
if given is not None:
return given
r = repo.fx.in_force(currency, company["functional_currency"], "spot", on)
return r["rate"] if r else None
@dataclass
class PostSpec:
company_code: str
journal_date: date
currency: str
lines: list
source_module: str
source_document: str
event: str
ledger_id: Optional[str] = None
fx_rate: Optional[Decimal] = None
memo: str = ""
journal_id: Optional[int] = None
book: Optional[str] = None
manual: bool = False
adjustment: bool = False
reverse_next_period: bool = False
extras: dict = field(default_factory=dict) # recurring_id, accrual_id, allocation_run_id, reval_run_id, ic_document_id ...
actor: str = ""
actor_sub: str = ""
allow_zero: bool = False
def expected_journal_id(source_document: str, event: str, ledger_id: Optional[str]) -> int:
return derived_id(source_document, event, ledger_id) if ledger_id else derived_id(source_document, event)
def number_of(row: dict) -> str:
return f"{row['book'].upper()}/{row['fiscal_year']}/{int(row['entry_no']):04d}" if row.get("entry_no") else ""
def post_summary(row: dict, replay: bool, lines: int) -> dict:
return {"journal_id": row["journal_id"], "entry_no": row["entry_no"], "number": number_of(row), "period_id": row["period_id"],
"status": row["status"], "replay": replay, "lines": lines}
def post_once(repo, spec: PostSpec, *, at: datetime) -> tuple:
"""Post a journal in the caller's transaction: (summary, touched) where touched = (company, period, account ids) to re-derive
afterwards, or None for a replay. Every gate of post_from_source is decided here, in the words a person reads."""
company = company_row(repo, {"sub": spec.actor_sub, "email": spec.actor}, spec.company_code, missing=422, scoped=False)
code = company["company_code"]
expected = expected_journal_id(spec.source_document, spec.event, spec.ledger_id)
jid = spec.journal_id if spec.journal_id is not None else expected
if jid != expected and not spec.extras.get("keyed"):
raise refusal("journal_id", f"journal_id {jid} is not derived_id(source_document, event"
f"{', ledger_id' if spec.ledger_id else ''}) = {expected}",
"derive it from the source document and event", status=422)
draft = Draft(code, spec.journal_date, spec.currency, spec.lines, spec.ledger_id, spec.fx_rate, spec.memo, spec.source_module,
spec.source_document, spec.event, bool(spec.reverse_next_period))
digest = jr.body_digest(draft)
found = repo.journals.get(jid)
if found is not None:
if found["body_digest"] == digest:
n = len(repo.journals.lines(jid))
return post_summary(found, True, n), None
raise refusal("journal_conflict", f"journal {jid} ({spec.source_module} {spec.source_document} {spec.event}) was posted with "
"a different body", "send the same body, or a new source document or event")
period = ledgerops.period_for(repo, code, spec.journal_date, adjustment=spec.adjustment)
year = dates.fy_number(period["fiscal_year"])
if spec.ledger_id and spec.ledger_id not in repo.company_ledgers.assigned(code, year):
raise refusal("ledger_assigned", f"the ledger {spec.ledger_id} is not assigned to {code} for {period['fiscal_year']}",
"Companies: assign the ledger, or post to all ledgers", status=422)
fx = fx_for(repo, company, spec.currency, spec.fx_rate, spec.journal_date)
ctx = RepoLookups(repo, company)
ctx.prefetch(ln.account_id for ln in spec.lines if ln.account_id)
built = jr.build(draft, ctx, manual=spec.manual, fx=fx, allow_zero_lines=spec.allow_zero)
hdr = {"journal_id": jid, "company_code": code, "ledger_id": spec.ledger_id, "journal_date": spec.journal_date,
"period_id": period["period_id"], "fiscal_year": period["fiscal_year"],
"book": spec.book or jr.book_for(spec.source_module, spec.manual, spec.adjustment), "source_module": spec.source_module,
"source_document": spec.source_document, "event": spec.event, "currency": spec.currency, "memo": spec.memo,
"adjustment": spec.adjustment, "reverse_next_period": spec.reverse_next_period,
"reversal_status": "pending" if spec.reverse_next_period else "none", "body_digest": digest,
"proposed_by": spec.actor if spec.manual else None, "proposed_sub": spec.actor_sub if spec.manual else None,
**spec.extras}
hdr.pop("keyed", None)
written = ledgerops.write_posted(repo, ctx, company, hdr, built, existing=None, at=at)
repo.journals.audit(jid, "posted", spec.actor, spec.actor_sub, at)
touched = (code, period["period_id"], sorted({ln.account_id for ln in built.lines}))
return post_summary(written["journal"], False, len(built.lines)), touched
def finish(touched: Optional[tuple], summary: dict, reverses: Optional[bool] = None) -> None:
"""After the lines COMMIT: re-derive the balances (a second transaction) and post an auto-reversal that is already due."""
if touched is None:
return
code, period_id, accounts = touched
with repository() as repo:
ledgerops.derive_balances(repo, code, period_id, accounts)
if reverses is not False: # only a journal that reverses itself can have a reversal due (None: not known here)
post_due_reversals(code, [summary["journal_id"]])
# ── the source route ─────────────────────────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="command", route="POST /journal-entry/journals", permission=POST, idempotent=True)
def post_from_source(caller: dict, source_module: text(64, min_length=1), source_document: text(200, min_length=1),
event: text(64, min_length=1), journal_date: IsoDate, currency: Currency, lines: Lines,
journal_id: Optional[BigId] = None, fx_rate: Optional[FxRate] = None, memo: Optional[Memo] = None,
company_code: Optional[CompanyCode] = None, ledger_id: Optional[LedgerCode] = None,
reverse_next_period: bool = False) -> dict:
"""Post what another module sent as ONE balanced journal (invoicing, payments, tax, purchasing, inventory, fixed-assets).
Idempotent by source: journal_id is derived from (source_document, event[, ledger_id]), so the same call again answers
the journal already posted (replay true) and a different body for the same id is refused 409 journal_conflict. A line
names an account_id or a PURPOSE (receivable, revenue, tax + tax_code, bank, fx_gain ...) resolved by the account map row
in force on the journal date; an unmapped purpose is 422 purposes_mapped, a date in a closed period of the company is 409
period_open, an unbalanced journal is 422 balanced (in the journal's currency and in the functional currency). A posting
that names no company lands in the default company, and one that names no ledger counts in every ledger of its company."""
on = dates.parse(journal_date)
spec = PostSpec(company_code or "", on, currency, to_lines(lines), source_module, source_document, event, ledger_id,
fx_rate, memo or "", int(journal_id) if journal_id is not None else None,
reverse_next_period=reverse_next_period, actor=who(caller), actor_sub=sub_of(caller))
def unit():
with repository() as repo, decide():
return post_once(repo, spec, at=real_now())
summary, touched = claims(unit)
finish(touched, summary, reverses=bool(reverse_next_period))
return summary
# ── reversal ─────────────────────────────────────────────────────────────────────────────────────────────────────────
def mirror_splits(rows: list) -> list:
out = []
for r in rows:
out.append(splitting.SplitLine(
line_no=r["line_no"], split_no=r["split_no"], account_id=r["account_id"], debit=r["credit"], credit=r["debit"],
f_debit=r["functional_credit"], f_credit=r["functional_debit"], profit_centre=r["profit_centre"] or "",
segment=r["segment"] or "", cost_centre=r["cost_centre"] or "", partner_company=r["partner_company"] or "",
movement_type=r["movement_type"] or "", bp_id=r["bp_id"], asset_id=r["asset_id"] or "",
is_clearing=bool(r["is_clearing"])))
return out
def post_reversal(repo, original: dict, on: date, reason: str, actor: str, actor_sub: str, at: datetime, *, auto: bool = False) -> tuple:
"""The mirror journal of a posted one, keyed derived_id(journal_id, 'reversal'): split the same way, neither deleted. A
plain INSERT of the reversal's id is the claim, so reversing twice (or twice at once) posts one."""
code = original["company_code"]
rid = derived_id(original["journal_id"], "reversal")
existing = repo.journals.get(rid)
if existing is not None:
return None, existing
period = ledgerops.period_for(repo, code, on)
company = repo.companies.get(code)
ctx = RepoLookups(repo, company)
rows = repo.journals.lines(original["journal_id"])
lines = [Line(line_no=r["line_no"], account_id=r["account_id"], purpose=r["purpose"] or "", debit=r["debit"], credit=r["credit"],
f_debit=r["functional_debit"], f_credit=r["functional_credit"], memo=r["line_memo"] or "", bp_id=r["bp_id"],
tax_code=r["tax_code"] or "", cost_centre=r["cost_centre"] or "", profit_centre=r["profit_centre"] or "",
partner_company=r["partner_company"] or "", movement_type=r["movement_type"] or "",
against_document=r["against_document"] or "", ic_reference=r["ic_reference"] or "", asset_id=r["asset_id"] or "",
currency=r["currency"] or "", amount_currency=r["amount_currency"], odoo_line_id=r["odoo_line_id"] or "",
qualifier=r["qualifier"] or "") for r in rows]
built = Built(lines=jr.mirror(lines), fx_rate=original["fx_rate"] or Decimal(1), total=original["total"],
f_total=original["functional_total"])
hdr = {"journal_id": rid, "company_code": code, "ledger_id": original["ledger_id"], "journal_date": on,
"period_id": period["period_id"], "fiscal_year": period["fiscal_year"], "book": original["book"],
"source_module": "gl", "source_document": str(original["journal_id"]), "event": "reversal", "currency": original["currency"],
"memo": f"Reversal of {number_of(original)}: {reason}"[:2000], "adjustment": bool(original["adjustment"]),
"reverses_journal_id": original["journal_id"], "body_digest": "", "approved_by": actor, "approved_sub": actor_sub,
"reversal_status": "none"}
sp = mirror_splits(repo.journals.splits(original["journal_id"]))
written = ledgerops.write_posted(repo, ctx, company, hdr, built, existing=None, at=at, splits=sp)
repo.journals.set((original["journal_id"],), status="reversed", reversal_status="posted", reversal_reason=reason)
repo.journals.audit(original["journal_id"], "reversed" if not auto else "auto-reversed", actor, actor_sub, at)
repo.journals.audit(rid, "posted", actor, actor_sub, at)
return (code, period["period_id"], sorted({ln.account_id for ln in lines})), written["journal"]
def next_period_first_day(journal_date: date) -> date:
return dates.first_of_next_month(journal_date)
def post_due_reversals(company_code: str, only: Optional[list] = None) -> list:
"""Post every pending auto-reversal of a company whose date (the first day of the month after the journal) is in an open
period; each in its own transaction, keyed derived_id(journal_id, 'reversal') so a retry or a second open posts nothing."""
with repository() as repo:
pending = [j for j in repo.journals.pending_reversals(company_code) if only is None or j["journal_id"] in only]
due = []
for j in pending:
on = next_period_first_day(j["journal_date"])
p = repo.periods.for_date(company_code, on)
if p is not None and p["status"] == "open":
due.append((j["journal_id"], on))
done = []
for jid, on in due:
def unit(jid=jid, on=on):
with repository() as repo, decide():
j = repo.journals.get(jid)
if j is None or j["reversal_status"] != "pending" or j["status"] != "posted":
return None
return post_reversal(repo, j, on, "automatic reversal on the first day of the next period", "system", "system",
real_now(), auto=True)
touched, row = claims(unit) or (None, None)
if touched:
with repository() as repo:
ledgerops.derive_balances(repo, touched[0], touched[1], touched[2])
done.append(row["journal_id"])
return done
@GL.use_case(kind="command", route="POST /journals/{journal_id}/reverse-journal", permission=REVERSAL_APPROVE)
def reverse_journal(caller: dict, journal_id: BigId, reason: Reason, reversal_date: Optional[IsoDate] = None) -> dict:
"""Correct a posted journal with a mirror journal rather than an edit: dated today (the default) or the first day of the
next period (an accrual reversing itself), keyed derived_id(journal_id, 'reversal'), referencing the original; both stay
readable and the balances return to where they were. Reversing twice (or twice at once) posts one reversal; a journal
whose auto-reversal is pending cannot be reversed by hand (one reversal, never two). A cross-company journal reverses
with its partners."""
def unit():
with repository() as repo, decide():
j = repo.journals.get(int(journal_id))
if j is None or not can_see(repo, caller, j["company_code"]):
raise Refused(404, f"no journal {journal_id}")
if j["status"] == "reversed":
other = repo.journals.reversal_of(j["journal_id"])
raise refusal("not_reversed", f"journal {number_of(j)} is already reversed by journal "
f"{other['journal_id'] if other else '(its reversal)'}", "none needed")
if j["status"] != "posted":
raise RuleBroken(f"journal {journal_id} is {j['status']}: only a posted journal is reversed")
if j["reversal_status"] == "pending":
raise refusal("no_pending_reversal", f"an automatic reversal of {number_of(j)} is pending for the next period",
"open the next period; the reversal posts itself")
td = today(repo)
on = dates.parse(reversal_date) if reversal_date else td
if on not in (td, dates.first_of_next_month(td)):
raise RuleBroken(f"a reversal is dated today ({td}) or the first day of the next period "
f"({dates.first_of_next_month(td)}), not {on}", status=422)
group = [j] if not j["ic_document_id"] else repo.journals.by_ic_document(j["ic_document_id"])
touched_all, last = [], None
for g in group:
if not can_see(repo, caller, g["company_code"]):
raise Refused(404, f"no journal {g['journal_id']}")
if g["status"] != "posted":
continue
touched, row = post_reversal(repo, g, on, reason, who(caller), sub_of(caller), real_now())
if touched:
touched_all.append(touched)
if g["journal_id"] == j["journal_id"]:
last = row
return touched_all, last
touched_all, last = claims(unit)
for touched in touched_all:
with repository() as repo:
ledgerops.derive_balances(repo, touched[0], touched[1], touched[2])
return {"journal_id": journal_id, "status": "reversed", "reversal": post_summary(last, False, 0) if last else None}
@GL.use_case(kind="command", route="POST /journals/post-pending-reversals", permission=POST, idempotent=True)
def post_pending_reversal(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""The system posts every pending auto-reversal (an accrual, a revaluation, a re-split) whose date, the first day of the
next period, is now in an open period: once each, keyed derived_id(journal_id, 'reversal'), so a second run posts
nothing. Also runs when periods open and after a posting that asks for a reversal."""
with repository() as repo:
codes = [company_row(repo, caller, company_code, scoped=False)["company_code"]] if company_code else \
[c["company_code"] for c in repo.companies.all()]
posted = []
for c in codes:
posted += post_due_reversals(c)
return {"posted": posted}
# ── manual journals ──────────────────────────────────────────────────────────────────────────────────────────────────
def approval_needed(settings, functional_total: Decimal, *, adjustment: bool = False, cross_company: bool = False) -> bool:
if adjustment:
return True
mode = settings.manual_journal_approval
if mode == "always":
return True
if mode == "none" and not cross_company:
return False
return functional_total > Decimal(settings.manual_journal_threshold)
def _soft_ok(repo, caller: dict, j: dict, p: Optional[dict]) -> bool:
"""A journal proposed under the soft close may be approved into the closed period by someone who closes periods."""
return (p is not None and p["status"] == "closed" and not j["adjustment"] and auth_has(caller, PERIOD_CLOSE)
and bool(repo.settings.in_force(today(repo), j["company_code"]).soft_close))
def _proposal_gates(repo, caller: dict, j: dict) -> list:
"""The readiness of approving a proposed journal: {gate, met, reason, fix} for each gate."""
gates = []
mine = same_person(caller, j["proposed_by"], j["proposed_sub"])
gates.append(unmet("second_person", "you proposed it, and a second person must approve it", "ask another accountant")
if mine else met("second_person"))
p = repo.periods.get(j["company_code"], j["period_id"])
ok = p is not None and (p["status"] == "open" or (j["adjustment"] and p["status"] == "closed") or _soft_ok(repo, caller, j, p))
gates.append(met("period_open") if ok else unmet("period_open", f"its period {j['period_id']} closed while it waited",
"reject it and propose it in an open period"))
return gates
@GL.use_case(kind="command", route="POST /journals/propose-manual", permission=JOURNAL_WRITE)
def propose_manual(caller: dict, journal_date: IsoDate, lines: Lines, memo: Optional[Memo] = None,
currency: Optional[Currency] = None, company_code: Optional[CompanyCode] = None,
ledger_id: Optional[LedgerCode] = None, fx_rate: Optional[FxRate] = None,
reverse_next_period: bool = False) -> dict:
"""Propose a manual journal (an accrual, a correction, a payroll summary): balanced, dated in an open period and not after
today, by someone with access to the company. An unbalanced journal is refused with the difference named; a line on a
receivable, payable or asset-cost account is refused (source_only_account). Where the owner asks for a second person
(manual_journal_approval, above the threshold) it waits for approve_manual; otherwise it posts at once with its number. A
ledger may be named (then it counts in that ledger only); reverse_next_period reverses it on the first day of the next
period (an accrual)."""
on = dates.parse(journal_date)
actor, actor_sub = who(caller), sub_of(caller)
def unit():
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
td = today(repo)
if on > td:
raise RuleBroken(f"a manual journal is not dated after today ({td}): {on} is in the future", status=422)
code = company["company_code"]
settings = repo.settings.in_force(td, code)
soft = False
try:
period = ledgerops.period_for(repo, code, on)
except RuleBroken:
# a SOFT close (the owner's soft_close setting): the accountant who closes periods may still post into a closed one, with a
# warning and the trail; everyone else, and every module's posting, is refused as ever.
p0 = repo.periods.for_date(code, on)
if not (settings.soft_close and p0 is not None and p0["status"] == "closed" and auth_has(caller, PERIOD_CLOSE)):
raise
period, soft = p0, True
year = dates.fy_number(period["fiscal_year"])
if ledger_id and ledger_id not in repo.company_ledgers.assigned(code, year):
raise refusal("ledger_assigned", f"the ledger {ledger_id} is not assigned to {code} for {period['fiscal_year']}",
"Companies: assign the ledger, or post to all ledgers", status=422)
cur = currency or company["functional_currency"]
draft = Draft(code, on, cur, to_lines(lines), ledger_id, fx_rate, memo or "", "gl", "", "manual")
fx = fx_for(repo, company, cur, fx_rate, on)
ctx = RepoLookups(repo, company)
built = jr.build(draft, ctx, manual=True, fx=fx)
jid = derived_id("manual", uuid.uuid4().hex)
at = real_now()
hdr = {"journal_id": jid, "company_code": code, "ledger_id": ledger_id, "journal_date": on,
"period_id": period["period_id"], "fiscal_year": period["fiscal_year"], "book": "misc", "source_module": "gl",
"source_document": f"manual:{jid}", "event": "manual", "currency": cur, "memo": memo or "",
"reverse_next_period": reverse_next_period, "reversal_status": "pending" if reverse_next_period else "none",
"body_digest": jr.body_digest(draft), "proposed_by": actor, "proposed_sub": actor_sub}
if approval_needed(settings, built.f_total):
row = {**ledgerops.HEADER_DEFAULTS, **hdr, "status": "proposed", "total": built.total,
"functional_total": built.f_total, "fx_rate": built.fx_rate}
repo.journals.add(row)
repo.journals.add_lines(_line_rows(jid, built))
repo.journals.audit(jid, "proposed", actor, actor_sub, at)
return {"journal_id": jid, "status": "proposed", "entry_no": None, "number": "", "period_id": period["period_id"],
"needs_approval": True, "functional_total": built.f_total, "lines": len(built.lines)}, None
written = ledgerops.write_posted(repo, ctx, company, hdr, built, existing=None, at=at, soft_closed=soft)
repo.journals.audit(jid, "posted into a closed period (soft close)" if soft else "posted", actor, actor_sub, at)
out = {**post_summary(written["journal"], False, len(built.lines)), "needs_approval": False}
if soft:
out["warning"] = f"the period {period['period_id']} is closed: this journal was posted anyway under the soft close (recorded)"
return out, (code, period["period_id"], sorted({ln.account_id for ln in built.lines}))
summary, touched = claims(unit)
finish(touched, summary)
if summary.get("status") == "proposed":
_auto_check(summary["journal_id"])
return summary
def _auto_check(journal_id: int) -> None:
"""With the owner's AI check on (suggest), list the findings for the approver as soon as the journal waits for one - in a second
transaction, because the lines were not readable inside the proposing one. A finding never blocks anything."""
from brick_gl.application import ai
with repository() as repo:
j = repo.journals.get(journal_id)
td = today(repo)
if j and repo.settings.in_force(td, j["company_code"]).mode("ai_check_journal") == "suggest":
ai.run_check(repo, j, td)
def _line_rows(jid: int, built: Built) -> list:
return [{"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]
def approve_one(repo, caller: dict, j: dict, at: datetime, *, by_system: bool = False) -> tuple:
"""Post one proposed journal as approved by the caller: gates, a rebuild on the posting day, the one write path. The SYSTEM posts a
journal nobody needs to approve (the owner's rule now says so) without a second person."""
gates = [g for g in _proposal_gates(repo, caller, j) if not (by_system and g.gate == "second_person")]
bad = [g for g in gates if not g.met]
if bad:
raise refusal(bad[0].gate, bad[0].reason, bad[0].fix)
company = repo.companies.get(j["company_code"])
ctx = RepoLookups(repo, company)
rows = repo.journals.lines(j["journal_id"])
draft = Draft(j["company_code"], j["journal_date"], j["currency"], lines_of_row(rows), j["ledger_id"], j["fx_rate"],
j["memo"] or "", "gl", "", "manual")
fx = j["fx_rate"] if j["currency"] != company["functional_currency"] else None # frozen at the proposal
built = jr.build(draft, ctx, manual=True, fx=fx)
hdr = {k: j[k] for k in ("journal_id", "company_code", "ledger_id", "journal_date", "period_id", "fiscal_year", "book",
"source_module", "source_document", "event", "currency", "memo", "adjustment",
"reverse_next_period", "reversal_status", "body_digest", "proposed_by", "proposed_sub",
"ic_document_id")}
hdr.update(approved_by=None if by_system else who(caller), approved_sub=None if by_system else sub_of(caller))
soft = _soft_ok(repo, caller, j, repo.periods.get(j["company_code"], j["period_id"]))
written = ledgerops.write_posted(repo, ctx, company, hdr, built, existing=j, at=at, soft_closed=soft)
repo.journals.audit(j["journal_id"], ("posted by the system" if by_system else "approved") + (" into a closed period (soft close)" if soft else ""),
who(caller), sub_of(caller), at)
return written["journal"], (j["company_code"], j["period_id"], sorted({ln.account_id for ln in built.lines}), bool(j["adjustment"]))
@GL.use_case(kind="command", route="POST /journals/{journal_id}/approve-manual", permission=JOURNAL_APPROVE)
def approve_manual(caller: dict, journal_id: BigId) -> dict:
"""Approve a proposed manual journal (or audit adjustment, or cross-company document): it posts with its number and the
approver recorded. The proposer cannot approve it, matched on BOTH email and sub, so a second login of the same person is
still the proposer (gate second_person); the period must still be open (the closed last period, for an adjustment). A
cross-company document posts every company's journal in one transaction, or none."""
def unit():
with repository() as repo, decide():
j = repo.journals.get(int(journal_id))
if j is None or not can_see(repo, caller, j["company_code"]):
raise Refused(404, f"no journal {journal_id}")
if j["status"] != "proposed":
raise RuleBroken(f"journal {journal_id} is {j['status']}, not proposed")
group = repo.journals.by_ic_document(j["ic_document_id"]) if j["ic_document_id"] else [j]
at = real_now()
posted, touched_all = [], []
for g in sorted(group, key=lambda r: r["company_code"]):
if not can_see(repo, caller, g["company_code"]):
raise Refused(404, f"no journal {g['journal_id']}")
if g["status"] != "proposed":
raise RuleBroken(f"journal {g['journal_id']} is {g['status']}, not proposed")
row, touched = approve_one(repo, caller, g, at)
posted.append(row)
touched_all.append(touched)
if j["ic_document_id"]:
repo.conn.execute(t.gl_ic_document.update().where(t.gl_ic_document.c.ic_document_id == j["ic_document_id"]).values(
status="posted", approved_by=who(caller), approved_sub=sub_of(caller)))
mine = next(p for p in posted if p["journal_id"] == j["journal_id"])
return {**post_summary(mine, False, len(repo.journals.lines(j["journal_id"]))), "approved_by": who(caller),
"journals": [post_summary(p, False, 0) for p in posted] if j["ic_document_id"] else None}, touched_all
summary, touched_all = claims(unit)
for code, period_id, accounts, adjustment in touched_all:
with repository() as repo:
ledgerops.derive_balances(repo, code, period_id, accounts)
if adjustment:
_rederive_carry(code, period_id)
for code, period_id, accounts, adjustment in touched_all:
post_due_reversals(code)
return summary
@GL.use_case(kind="command", route="POST /journals/{journal_id}/post-manual", permission=POST, idempotent=True)
def post_manual(caller: dict, journal_id: BigId) -> dict:
"""The system posts a PROPOSED manual journal nobody needs to approve any more: the owner's manual_journal_approval is now none, or
its total is at or below the threshold. Never an audit adjustment, and never a cross-company document above the threshold."""
def unit():
with repository() as repo, decide():
j = repo.journals.get(int(journal_id))
if j is None:
raise Refused(404, f"no journal {journal_id}")
if j["status"] != "proposed":
raise RuleBroken(f"journal {journal_id} is {j['status']}, not proposed", status=409)
group = repo.journals.by_ic_document(j["ic_document_id"]) if j["ic_document_id"] else [j]
if any(g["status"] != "proposed" for g in group):
raise RuleBroken(f"the document of journal {journal_id} is not wholly proposed", status=409)
top = max(g["functional_total"] for g in group)
if any(g["adjustment"] for g in group) or approval_needed(
repo.settings.in_force(today(repo), j["company_code"]), top, cross_company=bool(j["ic_document_id"])):
raise refusal("second_person", "this journal still needs a second person to approve it", "approve_manual by another accountant")
at, posted, touched_all = real_now(), [], []
for g in sorted(group, key=lambda r: r["company_code"]): # every company's journal, or none: one transaction
row, touched = approve_one(repo, caller, g, at, by_system=True)
posted.append(row)
touched_all.append(touched)
if j["ic_document_id"]:
repo.conn.execute(t.gl_ic_document.update().where(t.gl_ic_document.c.ic_document_id == j["ic_document_id"]).values(status="posted"))
mine = next(p for p in posted if p["journal_id"] == j["journal_id"])
return post_summary(mine, False, 0), touched_all
summary, touched_all = claims(unit)
for code, period_id, accounts, _adj in touched_all:
with repository() as repo:
ledgerops.derive_balances(repo, code, period_id, accounts)
return summary
def _rederive_carry(company_code: str, period_id: str) -> None:
"""A posting into a closed year changes what the next year opens with (GL-23): derive it again."""
from brick_gl.application import carry
with repository() as repo:
company = repo.companies.get(company_code)
p = repo.periods.get(company_code, period_id)
nxt = dates.fy_number(p["fiscal_year"]) + 1
if repo.years.get(company_code, dates.fy_name(nxt)) is not None:
carry.derive(repo, company, nxt, real_now())
@GL.use_case(kind="command", route="POST /journals/propose-audit-adjustment", permission=ADJUSTMENT_POST)
def propose_audit_adjustment(caller: dict, journal_date: IsoDate, lines: Lines, memo: Optional[Memo] = None,
currency: Optional[Currency] = None, company_code: Optional[CompanyCode] = None,
ledger_id: Optional[LedgerCode] = None, fx_rate: Optional[FxRate] = None) -> dict:
"""Propose the auditor's year-end adjustment: dated in the last period of a financial year that is still OPEN, that period
already closed. It posts into the closed period flagged as an adjustment (its own book and numbers) and ALWAYS waits for a
second person (approve_manual). Operational postings stay refused; the trial balance can be run before or after
adjustments; the carry-forward into the next year is re-derived. Once the year is closed an adjustment is refused
(gate year_open): post a prior-period correction in the current year."""
on = dates.parse(journal_date)
actor, actor_sub = who(caller), sub_of(caller)
def unit():
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
p = repo.periods.for_date(code, on)
if p is None:
raise refusal("period_open", f"no period of {code} covers {on}", "open the financial year")
year = repo.years.get(code, p["fiscal_year"])
if year["status"] == "closed":
raise refusal("year_open", f"the financial year {p['fiscal_year']} is closed",
"post a prior-period correction in the current year", status=409)
last = repo.periods.of_company(code, p["fiscal_year"])[-1]
if p["period_id"] != last["period_id"] or p["status"] != "closed":
raise RuleBroken(f"an audit adjustment is dated in the CLOSED last period of the year ({last['period_id']}, "
f"{last['status']}); use a manual journal for {on}", status=409)
if ledger_id and ledger_id not in repo.company_ledgers.assigned(code, dates.fy_number(p["fiscal_year"])):
raise refusal("ledger_assigned", f"the ledger {ledger_id} is not assigned to {code} for {p['fiscal_year']}",
"Companies: assign the ledger", status=422)
cur = currency or company["functional_currency"]
draft = Draft(code, on, cur, to_lines(lines), ledger_id, fx_rate, memo or "", "gl", "", "adjustment")
ctx = RepoLookups(repo, company)
built = jr.build(draft, ctx, manual=True, fx=fx_for(repo, company, cur, fx_rate, on))
jid = derived_id("manual", uuid.uuid4().hex)
row = {**ledgerops.HEADER_DEFAULTS, "journal_id": jid, "company_code": code, "ledger_id": ledger_id, "journal_date": on,
"period_id": p["period_id"], "fiscal_year": p["fiscal_year"], "book": "adjustment", "source_module": "gl",
"source_document": f"adjustment:{jid}", "event": "adjustment", "currency": cur, "memo": memo or "",
"adjustment": True, "body_digest": jr.body_digest(draft), "proposed_by": actor, "proposed_sub": actor_sub,
"status": "proposed", "total": built.total, "functional_total": built.f_total, "fx_rate": built.fx_rate}
repo.journals.add(row)
repo.journals.add_lines(_line_rows(jid, built))
repo.journals.audit(jid, "proposed", actor, actor_sub, real_now())
return {"journal_id": jid, "status": "proposed", "number": "", "period_id": p["period_id"], "needs_approval": True,
"adjustment": True, "functional_total": built.f_total, "lines": len(built.lines)}
return claims(unit)
@GL.use_case(kind="command", route="POST /journals/{journal_id}/reject-manual", permission=JOURNAL_APPROVE)
def reject_manual(caller: dict, journal_id: BigId, reason: Reason) -> dict:
"""Reject a proposed manual journal (or the whole cross-company document it belongs to), with the reason: it stays readable,
takes no number, and counts nowhere."""
with repository() as repo, decide():
j = repo.journals.get(int(journal_id))
if j is None or not can_see(repo, caller, j["company_code"]):
raise Refused(404, f"no journal {journal_id}")
if j["status"] != "proposed":
raise RuleBroken(f"journal {journal_id} is {j['status']}, not proposed")
group = repo.journals.by_ic_document(j["ic_document_id"]) if j["ic_document_id"] else [j]
for g in group:
if not can_see(repo, caller, g["company_code"]):
raise Refused(404, f"no journal {g['journal_id']}")
repo.journals.set((g["journal_id"],), status="rejected", rejected_by=who(caller), reject_reason=reason)
repo.journals.audit(g["journal_id"], "rejected", who(caller), sub_of(caller), real_now())
if j["ic_document_id"]:
repo.conn.execute(t.gl_ic_document.update().where(t.gl_ic_document.c.ic_document_id == j["ic_document_id"]).values(
status="rejected", reason=reason))
return {"journal_id": j["journal_id"], "status": "rejected", "reason": reason}
# ── reading journals ─────────────────────────────────────────────────────────────────────────────────────────────────
_HEADER_VIEW = ("journal_id", "company_code", "ledger_id", "status", "journal_date", "period_id", "fiscal_year", "book", "entry_no",
"source_module", "source_document", "event", "currency", "fx_rate", "memo", "total", "functional_total",
"adjustment", "reverse_next_period", "reversal_status", "reversal_reason", "ic_document_id", "recurring_id",
"accrual_id", "allocation_run_id", "reval_run_id", "proposed_by", "approved_by", "rejected_by", "reject_reason",
"posted_at", "reverses_journal_id", "chain_seq", "prev_hash", "row_hash", "migration_run_id")
def journal_view(j: dict) -> dict:
out = {k: j.get(k) for k in _HEADER_VIEW}
out["number"] = number_of(j)
return out
@GL.use_case(kind="query", route="GET /journals", permission=JOURNAL_READ)
def list_journals(caller: dict, company_code: Optional[CompanyCode] = None, status: Optional[text(16)] = None,
period_id: Optional[text(7)] = None, source_module: Optional[text(64)] = None, book: Optional[text(16)] = None,
after: Optional[BigId] = None, limit: Limit = 50) -> dict:
"""Journals of the companies the caller may see, oldest id first (page with `after`): header only; get_journal for the
lines. Filter by company, status (proposed, posted, rejected, reversed), period, source module or book."""
with repository() as repo:
mine = visible_codes(repo, caller)
if company_code:
if company_code not in mine:
raise Refused(404, f"no company {company_code}")
mine = [company_code]
rows = repo.journals.listing(mine, status, period_id, source_module, book, int(after or 0), limit)
return {"journals": [journal_view(r) for r in rows], "next": rows[-1]["journal_id"] if len(rows) == limit else None}
@GL.use_case(kind="query", route="GET /journals/{journal_id}", permission=JOURNAL_READ)
def get_journal(caller: dict, journal_id: BigId) -> dict:
"""One journal, whole: its header, its lines as they were sent (and resolved), the split lines every balance derives from,
the trail of what happened to it, and - while it waits for approval - the readiness of approving it ({gate, met, reason,
fix}). A journal of a company the caller may not see is a 404."""
with repository() as repo:
j = repo.journals.get(int(journal_id))
if j is None or not can_see(repo, caller, j["company_code"]):
raise Refused(404, f"no journal {journal_id}")
out = journal_view(j)
out["lines"] = [{k: v for k, v in ln.items() if k not in ("created_at", "updated_at")} for ln in repo.journals.lines(j["journal_id"])]
out["split_lines"] = [{k: v for k, v in s.items() if k not in ("created_at", "updated_at")} for s in repo.journals.splits(j["journal_id"])]
out["trail"] = [{"event": a["event"], "by": a["by_email"], "at": a["event_at"]} for a in repo.journals.trail(j["journal_id"])]
reversal = repo.journals.reversal_of(j["journal_id"])
out["reversed_by_journal_id"] = reversal["journal_id"] if reversal else None
if j["status"] == "proposed":
out["readiness"] = [g.as_dict() for g in _proposal_gates(repo, caller, j)]
return out