dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/scheduled.py
"""Work gl does on a schedule, posted through the SAME write path as everything else, keyed so that running it twice posts once:
recurring entries from an approved template (GL-45), accruals and deferrals released over the months they cover (GL-49).
There is no scheduler on this platform: the accountant (or the closing cockpit, or a job calling the machine route) runs a
period; the run is idempotent - a journal id is derived from the template/registration and the period, and a plain INSERT of
that id is the claim, so a retry or a second run posts nothing and answers what it already posted.
"""
import json
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from erp_platform.types import BigId
from brick_gl.application import ledgerops
from brick_gl.application.brick import (
ACCRUAL_APPROVE,
ACCRUAL_RUN,
ACCRUAL_WRITE,
GL,
PERIOD_READ,
RECURRING_APPROVE,
RECURRING_RUN,
RECURRING_WRITE,
claims,
company_row,
decide,
real_now,
repository,
same_person,
sub_of,
today,
who,
)
from brick_gl.application.inputs import Code, CompanyCode, LedgerCode, Lines, Memo, Money, PeriodId, Reason
from brick_gl.application.posting import PostSpec, finish, post_once, to_lines
from brick_gl.domain import dates, schedule
from brick_gl.domain import journal as jr
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.journal import Draft, LineIn
from brick_gl.domain.money import ZERO, q2
from brick_gl.domain.readiness import refusal
from brick_gl.domain.values import SOURCE_ONLY_SUBTYPES # noqa: F401
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, utc
def _line_json(models: list) -> str:
out = []
for m in models:
d = {k: (str(v) if v is not None else None) for k, v in m.model_dump().items() if v is not None}
out.append(d)
return json.dumps(out)
def _lines_from_json(raw: str) -> list:
out = []
for d in json.loads(raw or "[]"):
out.append(LineIn(account_id=int(d["account_id"]) if d.get("account_id") else None, purpose=d.get("purpose"),
qualifier=d.get("qualifier"), tax_code=d.get("tax_code"), bank_account_id=d.get("bank_account_id"),
asset_class=d.get("asset_class"), debit=Decimal(d["debit"]) if d.get("debit") else ZERO,
credit=Decimal(d["credit"]) if d.get("credit") else ZERO, memo=d.get("memo") or "",
bp_id=int(d["bp_id"]) if d.get("bp_id") else None, cost_centre=d.get("cost_centre"),
profit_centre=d.get("profit_centre"), partner_company=d.get("partner_company"),
movement_type=d.get("movement_type")))
return out
def post_as_gl(repo, *, company_code: str, key: int, source_document: str, event: str, on: date, currency: str, lines: list,
ledger_id: Optional[str], memo: str, book: str, extras: dict, reverse_next_period: bool = False, actor: str,
actor_sub: str, at, manual: bool = True) -> tuple:
"""Post a journal gl itself originates (a template, a release, an allocation, a revaluation) keyed by `key`."""
spec = PostSpec(company_code, on, currency, lines, "gl", source_document, event, ledger_id, None, memo, key, book=book,
manual=manual, reverse_next_period=reverse_next_period, extras={**extras, "keyed": True}, actor=actor,
actor_sub=actor_sub, allow_zero=False)
return post_once(repo, spec, at=at)
# ═══ recurring entries (GL-45) ══════════════════════════════════════════════════════════════════════════════════════════
def _recurring_view(r: dict) -> dict:
return {k: r[k] for k in ("recurring_id", "version", "company_code", "ledger_id", "memo", "frequency", "first_period", "last_period",
"posting_day", "status", "proposed_by", "approved_by", "ended_at", "end_reason")} | {
"lines": json.loads(r["lines_json"] or "[]")}
def _validate_template(repo, company: dict, lines: list, ledger_id, first_period: str, memo: str) -> None:
"""Balanced, the company's accounts valid: a dry run of the build on the first day of its first period."""
from brick_gl.application.brick import RepoLookups
on = dates.period_bounds(first_period)[0]
cur = company["functional_currency"]
jr.build(Draft(company["company_code"], on, cur, lines, ledger_id, None, memo, "gl", "", "recurring"), RepoLookups(repo, company), manual=True)
@GL.use_case(kind="command", route="POST /recurring", permission=RECURRING_WRITE)
def propose_recurring(caller: dict, recurring_id: Code, lines: Lines, first_period: PeriodId, frequency: Code = "monthly",
last_period: Optional[PeriodId] = None, posting_day: Optional[Code] = "last", memo: Optional[Memo] = None,
company_code: Optional[CompanyCode] = None, ledger_id: Optional[LedgerCode] = None) -> dict:
"""Propose a recurring entry (rent, a lease, a standing charge) as a template: balanced lines, the first and last period it is
due (monthly, quarterly or yearly), the posting day (last = the month's last day, or 1 to 28), the ledger (none = every
ledger). Version 1, the proposer recorded (email and sub); it posts nothing until another accountant approves it where the
owner asks for a second person."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
if frequency not in ("monthly", "quarterly", "yearly"):
raise RuleBroken("frequency is monthly, quarterly or yearly", status=422)
if posting_day not in (None, "last") and not (str(posting_day).isdigit() and 1 <= int(posting_day) <= 28):
raise RuleBroken("posting_day is 'last' or a day from 1 to 28", status=422)
if last_period and last_period < first_period:
raise RuleBroken("last_period is not before first_period", status=422)
if repo.conn.execute(t.gl_recurring_journal.select().where(t.gl_recurring_journal.c.recurring_id == recurring_id)).first():
raise RuleBroken(f"recurring entry {recurring_id} exists: revise_recurring makes a new version", status=409)
_validate_template(repo, company, to_lines(lines), ledger_id, first_period, memo or "")
row = {"recurring_id": recurring_id, "version": 1, "company_code": code, "ledger_id": ledger_id, "lines_json": _line_json(lines),
"memo": memo or "", "frequency": frequency, "first_period": first_period, "last_period": last_period,
"posting_day": str(posting_day or "last"), "status": "proposed", "proposed_by": who(caller), "proposed_sub": sub_of(caller),
"approved_by": None, "approved_sub": None, "ended_at": None, "end_reason": None}
_Keyed(repo.conn, t.gl_recurring_journal, ("recurring_id", "version")).add(row)
return _recurring_view(row)
def _template(repo, caller, recurring_id: str, version: Optional[int] = None) -> dict:
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_recurring_journal.select().where(
t.gl_recurring_journal.c.recurring_id == recurring_id).order_by(t.gl_recurring_journal.c.version))]
if not rows:
raise Refused(404, f"no recurring entry {recurring_id}")
row = next((r for r in rows if r["version"] == version), None) if version else rows[-1]
if row is None:
raise Refused(404, f"no version {version} of {recurring_id}")
company_row(repo, caller, row["company_code"]) # a company the caller may not see is a 404
return row
@GL.use_case(kind="command", route="POST /recurring/{recurring_id}/approve-recurring", permission=RECURRING_APPROVE)
def approve_recurring(caller: dict, recurring_id: Code, version: Optional[int] = None) -> dict:
"""Approve a proposed recurring entry: it runs from its first period. Where the owner asks for a second person
(manual_journal_approval) the approver is not the proposer, matched on email AND sub."""
with repository() as repo, decide():
r = _template(repo, caller, recurring_id, version)
if r["status"] != "proposed":
raise RuleBroken(f"{recurring_id} version {r['version']} is {r['status']}, not proposed", status=409)
settings = repo.settings.in_force(today(repo), r["company_code"])
if settings.manual_journal_approval != "none" and same_person(caller, r["proposed_by"], r["proposed_sub"]):
raise refusal("second_person", "you proposed it, and the owner asks for a second person", "ask another accountant")
_Keyed(repo.conn, t.gl_recurring_journal, ("recurring_id", "version")).set(
(recurring_id, r["version"]), status="approved", approved_by=who(caller), approved_sub=sub_of(caller))
return _recurring_view(_template(repo, caller, recurring_id, r["version"]))
@GL.use_case(kind="command", route="POST /recurring/{recurring_id}/reject-recurring", permission=RECURRING_APPROVE)
def reject_recurring(caller: dict, recurring_id: Code, reason: Reason, version: Optional[int] = None) -> dict:
"""Reject a proposed recurring entry, with the reason."""
with repository() as repo, decide():
r = _template(repo, caller, recurring_id, version)
if r["status"] != "proposed":
raise RuleBroken(f"{recurring_id} version {r['version']} is {r['status']}, not proposed", status=409)
_Keyed(repo.conn, t.gl_recurring_journal, ("recurring_id", "version")).set(
(recurring_id, r["version"]), status="rejected", end_reason=reason, ended_at=utc(real_now()))
return _recurring_view(_template(repo, caller, recurring_id, r["version"]))
@GL.use_case(kind="command", route="POST /recurring/{recurring_id}/revise-recurring", permission=RECURRING_WRITE)
def revise_recurring(caller: dict, recurring_id: Code, reason: Reason, lines: Optional[Lines] = None, memo: Optional[Memo] = None,
last_period: Optional[PeriodId] = None, from_period: Optional[PeriodId] = None,
posting_day: Optional[Code] = None) -> dict:
"""Change an approved recurring entry: this version ENDS at the change (its last period is the one before from_period, by
default the current period) and version + 1 is proposed with the changes - it needs approval again, so a period with
neither version approved is skipped and listed. Posted journals are untouched."""
with repository() as repo, decide():
r = _template(repo, caller, recurring_id)
if r["status"] != "approved":
raise RuleBroken(f"{recurring_id} version {r['version']} is {r['status']}: only an approved version is revised", status=409)
company = repo.companies.get(r["company_code"])
start = from_period or dates.period_id(today(repo))
if start <= r["first_period"]:
start = dates.next_period(r["first_period"]) if start < r["first_period"] else start
new_lines = _line_json(lines) if lines else r["lines_json"]
if lines:
_validate_template(repo, company, to_lines(lines), r["ledger_id"], start, memo or r["memo"])
K = _Keyed(repo.conn, t.gl_recurring_journal, ("recurring_id", "version"))
K.set((recurring_id, r["version"]), status="ended", ended_at=utc(real_now()), end_reason=reason,
last_period=dates.prev_period(start) if start > r["first_period"] else r["first_period"])
row = {"recurring_id": recurring_id, "version": r["version"] + 1, "company_code": r["company_code"], "ledger_id": r["ledger_id"],
"lines_json": new_lines, "memo": memo if memo is not None else r["memo"], "frequency": r["frequency"],
"first_period": start, "last_period": last_period or r["last_period"], "posting_day": str(posting_day or r["posting_day"]),
"status": "proposed", "proposed_by": who(caller), "proposed_sub": sub_of(caller), "approved_by": None,
"approved_sub": None, "ended_at": None, "end_reason": None}
K.add(row)
return _recurring_view(row)
@GL.use_case(kind="command", route="POST /recurring/{recurring_id}/end-recurring", permission=RECURRING_WRITE)
def end_recurring(caller: dict, recurring_id: Code, reason: Reason) -> dict:
"""End an approved recurring entry: no period after the end posts; journals already posted are untouched."""
with repository() as repo, decide():
r = _template(repo, caller, recurring_id)
if r["status"] not in ("approved", "proposed"):
raise RuleBroken(f"{recurring_id} is {r['status']}", status=409)
last = dates.period_id(today(repo))
_Keyed(repo.conn, t.gl_recurring_journal, ("recurring_id", "version")).set(
(recurring_id, r["version"]), status="ended", ended_at=utc(real_now()), end_reason=reason, last_period=min(last, r["last_period"] or last))
return _recurring_view(_template(repo, caller, recurring_id, r["version"]))
@GL.use_case(kind="query", route="GET /recurring", permission=PERIOD_READ)
def list_recurring(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""Recurring entries with their versions, approvals and what each period posted."""
with repository() as repo:
company = company_row(repo, caller, company_code)
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_recurring_journal.select().where(
t.gl_recurring_journal.c.company_code == company["company_code"]).order_by(
t.gl_recurring_journal.c.recurring_id, t.gl_recurring_journal.c.version))]
posted = [dict(r._mapping) for r in repo.conn.execute(t.gl_recurring_posting.select())]
return {"recurring": [{**_recurring_view(r), "postings": [{"period_id": p["period_id"], "journal_id": p["journal_id"]}
for p in posted if p["recurring_id"] == r["recurring_id"] and p["version"] == r["version"]]}
for r in rows]}
def _post_recurring_period(caller: dict, recurring_id: str, version: int, period_id: str) -> dict:
"""One template's journal for one period, in its own transaction: replay when it already posted."""
def unit():
with repository() as repo, decide():
r = _template(repo, caller, recurring_id, version)
posted = repo.conn.execute(t.gl_recurring_posting.select().where(
t.gl_recurring_posting.c.recurring_id == recurring_id, t.gl_recurring_posting.c.version == version,
t.gl_recurring_posting.c.period_id == period_id)).first()
if posted is not None:
p = dict(posted._mapping)
return {"recurring_id": recurring_id, "version": version, "journal_id": p["journal_id"], "replay": True}, None
key = derived_id("recurring", recurring_id, version, period_id)
on = schedule.posting_date(period_id, r["posting_day"])
at = real_now()
summary, touched = post_as_gl(
repo, company_code=r["company_code"], key=key, source_document=f"recurring:{recurring_id}:{version}:{period_id}",
event="recurring", on=on, currency=repo.companies.get(r["company_code"])["functional_currency"],
lines=_lines_from_json(r["lines_json"]), ledger_id=r["ledger_id"], memo=f"{r['memo'] or recurring_id} {period_id}",
book="misc", extras={"recurring_id": recurring_id}, actor=who(caller), actor_sub=sub_of(caller), at=at)
if not summary["replay"]:
_Keyed(repo.conn, t.gl_recurring_posting, ("recurring_id", "version", "period_id")).add(
{"recurring_id": recurring_id, "version": version, "period_id": period_id, "journal_id": key, "skipped_reason": None})
return {"recurring_id": recurring_id, "version": version, "journal_id": summary["journal_id"], "number": summary["number"],
"replay": summary["replay"]}, (touched, summary)
out, extra = claims(unit)
if extra:
finish(extra[0], extra[1])
return out
@GL.use_case(kind="command", route="POST /recurring/{recurring_id}/post-recurring-period", permission=RECURRING_RUN, idempotent=True)
def post_recurring_period(caller: dict, recurring_id: Code, period_id: PeriodId, version: Optional[int] = None) -> dict:
"""Post ONE template's journal for one period (a person or the system): keyed derived_id('recurring', recurring_id, version,
period_id), so twice posts once; refused while the version is not approved or the period is closed."""
with repository() as repo, decide():
r = _template(repo, caller, recurring_id, version)
if r["status"] not in ("approved", "ended"):
raise RuleBroken(f"{recurring_id} version {r['version']} is {r['status']}: only an approved version posts", status=409)
if not schedule.recurring_due(r["frequency"], r["first_period"], r["last_period"], period_id):
raise RuleBroken(f"{recurring_id} version {r['version']} is not due in {period_id}", status=409)
p = repo.periods.get(r["company_code"], period_id)
if p is None or p["status"] != "open":
raise refusal("period_open", f"the period {period_id} is closed", "post it in an open period by revising the template, or reopen")
v = r["version"]
return _post_recurring_period(caller, recurring_id, v, period_id)
@GL.use_case(kind="command", route="POST /periods/{period_id}/run-recurring-entries", permission=RECURRING_RUN, idempotent=True)
def run_recurring_entries(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None) -> dict:
"""Run the recurring entries of a period: one journal per due, approved template, keyed derived_id('recurring', recurring_id,
version, period_id), in book misc with its number. Run twice (or twice at once) it posts nothing new and answers what it
already posted. The answer lists each template posted or skipped with the reason (not approved, period closed, ended)."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
p = repo.periods.get(code, period_id)
if p is None:
raise Refused(404, f"no period {period_id} of {code}")
tpl = [dict(r._mapping) for r in repo.conn.execute(t.gl_recurring_journal.select().where(
t.gl_recurring_journal.c.company_code == code).order_by(t.gl_recurring_journal.c.recurring_id, t.gl_recurring_journal.c.version))]
todo, skipped = [], []
for r in tpl:
if not schedule.recurring_due(r["frequency"], r["first_period"], r["last_period"], period_id):
continue
if r["status"] == "proposed":
skipped.append({"recurring_id": r["recurring_id"], "version": r["version"], "reason": "not approved"})
elif r["status"] == "ended" and not (r["last_period"] and period_id <= r["last_period"]):
skipped.append({"recurring_id": r["recurring_id"], "version": r["version"], "reason": "ended"})
elif r["status"] in ("approved", "ended"):
if p["status"] != "open":
skipped.append({"recurring_id": r["recurring_id"], "version": r["version"], "reason": f"period {period_id} closed"})
else:
todo.append(r)
posted = []
for r in todo:
posted.append(_post_recurring_period(caller, r["recurring_id"], r["version"], period_id))
return {"company_code": code, "period_id": period_id, "posted": posted, "skipped": skipped}
# ═══ accruals and deferrals (GL-49, GL-50) ══════════════════════════════════════════════════════════════════════════════
KINDS = {"prepaid_expense": ("prepaid", "expense"), "accrued_expense": ("accrual_balance", "expense"),
"deferred_revenue": ("deferred_revenue", "income"), "accrued_income": ("accrual_balance", "income")}
def _accrual_view(a: dict, releases: list) -> dict:
sched = schedule.release_schedule(a["amount"], a["start_period"], a["end_period"], a["method"])
done = {r["period_id"]: r for r in releases}
released = sum((r["amount"] for r in releases), ZERO)
return {"company_code": a["company_code"], "accrual_code": a["accrual_code"], "kind": a["kind"], "status": a["status"], "amount": a["amount"],
"currency": a["currency"], "start_period": a["start_period"], "end_period": a["end_period"], "method": a["method"],
"pl_account_id": a["pl_account_id"], "balance_account_id": a["balance_account_id"], "ledger_id": a["ledger_id"],
"terminated_period": a["terminated_period"], "released": released, "remaining": a["amount"] - released,
"schedule": [{"period_id": p, "amount": v, "released": p in done, "journal_id": done[p]["journal_id"] if p in done else None}
for p, v in sched.items()]}
def _accrual(repo, caller, company_code: str, code: str) -> dict:
company_row(repo, caller, company_code)
r = repo.conn.execute(t.gl_accrual.select().where(t.gl_accrual.c.company_code == company_code, t.gl_accrual.c.accrual_code == code)).first()
if r is None:
raise Refused(404, f"no accrual {code} of {company_code}")
return dict(r._mapping)
def _releases(repo, company_code: str, code: str) -> list:
return [dict(r._mapping) for r in repo.conn.execute(t.gl_accrual_release.select().where(
t.gl_accrual_release.c.company_code == company_code, t.gl_accrual_release.c.accrual_code == code).order_by(
t.gl_accrual_release.c.period_id))]
@GL.use_case(kind="command", route="POST /accruals", permission=ACCRUAL_WRITE)
def register_accrual(caller: dict, accrual_code: Code, kind: Code, amount: Money, start_period: PeriodId, end_period: PeriodId,
pl_account_id: BigId, balance_account_id: Optional[BigId] = None, method: Code = "straight_line",
company_code: Optional[CompanyCode] = None, ledger_id: Optional[LedgerCode] = None,
cost_centre: Optional[Code] = None, profit_centre: Optional[Code] = None,
source_journal_id: Optional[BigId] = None) -> dict:
"""Register a prepayment, deferred income or expected accrual so that each month carries its share without a manual journal
every month: kind prepaid_expense (Dr expense / Cr prepaid), deferred_revenue (Dr deferred_revenue / Cr income),
accrued_expense (Dr expense / Cr accrual_balance) or accrued_income; `pl_account_id` is the expense or income account and
`balance_account_id` the balance-sheet side (default: the account map's purpose). Straight line by period, or daily
proration; the last period takes the rounding remainder so the total released equals the amount to the penny. Above
manual_journal_threshold it needs a second person (approve_accrual), otherwise it is active at once."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
if kind not in KINDS:
raise RuleBroken(f"kind is one of {', '.join(KINDS)}", status=422)
if method not in ("straight_line", "daily"):
raise RuleBroken("method is straight_line or daily", status=422)
if amount <= 0:
raise RuleBroken("the amount is above zero", status=422)
if start_period > end_period:
raise RuleBroken("the start period is not after the end period", status=422)
pl = repo.accounts.get(int(pl_account_id))
if pl is None or pl["chart_id"] != company["chart_id"]:
raise RuleBroken(f"account {pl_account_id} is not an account of {company['chart_id']}", status=422)
bal_id = int(balance_account_id) if balance_account_id else repo.account_map.account(
company["chart_id"], KINDS[kind][0], "", dates.period_bounds(start_period)[0])
if bal_id is None or repo.accounts.get(bal_id) is None:
raise refusal("purposes_mapped", f"the purpose {KINDS[kind][0]} has no account in the map", "Account map: map it (GL-12)", status=422)
if repo.conn.execute(t.gl_accrual.select().where(t.gl_accrual.c.company_code == code, t.gl_accrual.c.accrual_code == accrual_code)).first():
raise RuleBroken(f"{code} already has an accrual {accrual_code}", status=409)
td = today(repo)
settings = repo.settings.in_force(td, code)
needs = q2(amount) > Decimal(settings.manual_journal_threshold) and settings.manual_journal_approval != "none"
row = {"company_code": code, "accrual_code": accrual_code, "kind": kind, "ledger_id": ledger_id, "amount": q2(amount),
"currency": company["functional_currency"], "pl_account_id": pl["account_id"], "balance_account_id": bal_id,
"cost_centre": cost_centre, "profit_centre": profit_centre, "start_period": start_period, "end_period": end_period,
"method": method, "source_journal_id": int(source_journal_id) if source_journal_id else None,
"status": "proposed" if needs else "active", "registered_by": who(caller), "registered_sub": sub_of(caller),
"approved_by": None, "approved_sub": None, "terminated_period": None, "reject_reason": None}
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).add(row)
return _accrual_view(row, [])
@GL.use_case(kind="command", route="POST /accruals/{accrual_code}/approve-accrual", permission=ACCRUAL_APPROVE)
def approve_accrual(caller: dict, accrual_code: Code, company_code: Optional[CompanyCode] = None) -> dict:
"""Approve a registration waiting for a second person (above the threshold): the registrant cannot approve it, matched on
email AND sub. It is then released from its start period."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
a = _accrual(repo, caller, company["company_code"], accrual_code)
if a["status"] != "proposed":
raise RuleBroken(f"{accrual_code} is {a['status']}, not proposed", status=409)
if same_person(caller, a["registered_by"], a["registered_sub"]):
raise refusal("second_person", "you registered it, and it is above the threshold", "ask another accountant")
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).set((a["company_code"], accrual_code), status="active",
approved_by=who(caller), approved_sub=sub_of(caller))
return _accrual_view(_accrual(repo, caller, a["company_code"], accrual_code), _releases(repo, a["company_code"], accrual_code))
@GL.use_case(kind="command", route="POST /accruals/{accrual_code}/reject-accrual", permission=ACCRUAL_APPROVE)
def reject_accrual(caller: dict, accrual_code: Code, reason: Reason, company_code: Optional[CompanyCode] = None) -> dict:
"""Reject a registration waiting for approval, with the reason."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
a = _accrual(repo, caller, company["company_code"], accrual_code)
if a["status"] != "proposed":
raise RuleBroken(f"{accrual_code} is {a['status']}, not proposed", status=409)
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).set((a["company_code"], accrual_code), status="rejected", reject_reason=reason)
return _accrual_view(_accrual(repo, caller, a["company_code"], accrual_code), [])
def _release_one(caller: dict, company_code: str, accrual_code: str, period_id: str) -> Optional[dict]:
def unit():
with repository() as repo, decide():
a = _accrual(repo, caller, company_code, accrual_code)
if a["status"] not in ("active",):
return None, None
rels = _releases(repo, company_code, accrual_code)
done = {r["period_id"]: r for r in rels}
if period_id in done:
return {"accrual_code": accrual_code, "period_id": period_id, "amount": done[period_id]["amount"],
"journal_id": done[period_id]["journal_id"], "replay": True}, None
sched = schedule.release_schedule(a["amount"], a["start_period"], a["end_period"], a["method"])
amount = sched.get(period_id)
if a["terminated_period"] and period_id == a["terminated_period"]:
amount = a["amount"] - sum((r["amount"] for r in rels), ZERO)
elif amount is None or (a["terminated_period"] and period_id > a["terminated_period"]):
return None, None
if amount == 0:
return None, None
p = repo.periods.get(company_code, period_id)
if p is None or p["status"] != "open":
raise refusal("period_open", f"the period {period_id} of {company_code} is closed", "release it in the next open period by "
"terminating and re-registering, or reopen")
pl_side, bal_side = ("debit", "credit") if KINDS[a["kind"]][1] == "expense" else ("credit", "debit")
ln = [LineIn(account_id=a["pl_account_id"], cost_centre=a["cost_centre"], profit_centre=a["profit_centre"], memo=accrual_code,
**{pl_side: amount}), LineIn(account_id=a["balance_account_id"], memo=accrual_code, **{bal_side: amount})]
key = derived_id("accrual", f"{company_code}:{accrual_code}", period_id)
at = real_now()
company = repo.companies.get(company_code)
summary, touched = post_as_gl(
repo, company_code=company_code, key=key, source_document=f"accrual:{company_code}:{accrual_code}:{period_id}",
event="accrual", on=dates.period_bounds(period_id)[1], currency=company["functional_currency"], lines=ln,
ledger_id=a["ledger_id"], memo=f"{accrual_code} {period_id}", book="accrual", extras={"accrual_id": f"{company_code}:{accrual_code}"},
actor=who(caller), actor_sub=sub_of(caller), at=at, manual=False)
if not summary["replay"]:
_Keyed(repo.conn, t.gl_accrual_release, ("company_code", "accrual_code", "period_id")).add(
{"company_code": company_code, "accrual_code": accrual_code, "period_id": period_id, "amount": amount, "journal_id": key})
total = sum((r["amount"] for r in rels), ZERO) + amount
if total == a["amount"]:
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).set((company_code, accrual_code), status="completed")
return {"accrual_code": accrual_code, "period_id": period_id, "amount": amount, "journal_id": summary["journal_id"],
"replay": summary["replay"]}, (touched, summary)
out, extra = claims(unit)
if extra:
finish(extra[0], extra[1])
return out
@GL.use_case(kind="command", route="POST /periods/{period_id}/run-accruals", permission=ACCRUAL_RUN, idempotent=True)
def run_accruals(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None) -> dict:
"""Release every active accrual and deferral's share for a period: one journal per registration keyed
derived_id('accrual', accrual_id, period_id) (Dr 6300 1,000.00 / Cr 1400 1,000.00 for a twelfth of a 12,000.00 premium). Run
twice it posts once. A terminated registration releases its remainder in the termination period and nothing after; a
registration that has released every period is completed."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_accrual.select().where(
t.gl_accrual.c.company_code == code, t.gl_accrual.c.status == "active").order_by(t.gl_accrual.c.accrual_code))]
released = []
for a in rows:
r = _release_one(caller, code, a["accrual_code"], period_id)
if r:
released.append(r)
return {"company_code": code, "period_id": period_id, "released": released}
@GL.use_case(kind="command", route="POST /accruals/{accrual_code}/release-accrual-period", permission=ACCRUAL_RUN, idempotent=True)
def release_accrual_period(caller: dict, accrual_code: Code, period_id: PeriodId, company_code: Optional[CompanyCode] = None) -> dict:
"""Release ONE registration's share for one period (also how the closing cockpit's task runs it): keyed
derived_id('accrual', accrual_id, period_id), so twice posts once; refused when the period is closed or not in its schedule."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
a = _accrual(repo, caller, company["company_code"], accrual_code)
if a["status"] != "active":
raise RuleBroken(f"{accrual_code} is {a['status']}, not active", status=409)
out = _release_one(caller, company["company_code"], accrual_code, period_id)
if out is None:
raise Refused(409, f"{accrual_code} has nothing to release in {period_id}")
return out
@GL.use_case(kind="command", route="POST /accruals/{accrual_code}/complete-accrual", permission=ACCRUAL_RUN, idempotent=True)
def complete_accrual(caller: dict, accrual_code: Code, company_code: Optional[CompanyCode] = None) -> dict:
"""Complete a registration once every period of its schedule is released: the released total equals the amount to the penny.
(A release that releases the last period completes it already; this answers the state and completes a registration that is
fully released but still active.) Refused while a period is unreleased."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
a = _accrual(repo, caller, company["company_code"], accrual_code)
if a["status"] == "completed":
return _accrual_view(a, _releases(repo, a["company_code"], accrual_code))
if a["status"] != "active":
raise RuleBroken(f"{accrual_code} is {a['status']}, not active", status=409)
rels = _releases(repo, a["company_code"], accrual_code)
left = a["amount"] - sum((r["amount"] for r in rels), ZERO)
if left != 0:
raise RuleBroken(f"{accrual_code} still has {left} to release", status=409)
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).set((a["company_code"], accrual_code), status="completed")
return _accrual_view(_accrual(repo, caller, a["company_code"], accrual_code), rels)
@GL.use_case(kind="command", route="POST /accruals/{accrual_code}/terminate-accrual", permission=ACCRUAL_WRITE)
def terminate_accrual(caller: dict, accrual_code: Code, reason: Reason, termination_period: Optional[PeriodId] = None,
company_code: Optional[CompanyCode] = None) -> dict:
"""End a registration early: its remainder (the amount less what was released) is released in the termination period -
posting 6,000.00 in 2026-07 for a 12,000.00 premium released to 2026-06 - and nothing after. The termination period must be
open and not already released."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
a = _accrual(repo, caller, code, accrual_code)
if a["status"] != "active":
raise RuleBroken(f"{accrual_code} is {a['status']}, not active", status=409)
period = termination_period or dates.period_id(today(repo))
if any(r["period_id"] >= period for r in _releases(repo, code, accrual_code)):
raise RuleBroken(f"{accrual_code} already released {period} or later: terminate in the next period", status=409)
p = repo.periods.get(code, period)
if p is None or p["status"] != "open":
raise refusal("period_open", f"the termination period {period} is not open", "terminate in an open period")
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).set((code, accrual_code), terminated_period=period)
out = _release_one(caller, code, accrual_code, period)
with repository() as repo:
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).set((code, accrual_code), status="terminated", reject_reason=reason)
return {**_accrual_view(_accrual(repo, caller, code, accrual_code), _releases(repo, code, accrual_code)), "final_release": out}
@GL.use_case(kind="query", route="GET /accruals", permission=PERIOD_READ)
def list_accruals(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""Registrations with their schedule, what is released and what remains, and the pending auto-reversals."""
with repository() as repo:
company = company_row(repo, caller, company_code)
code = company["company_code"]
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_accrual.select().where(t.gl_accrual.c.company_code == code).order_by(
t.gl_accrual.c.accrual_code))]
return {"company_code": code, "accruals": [_accrual_view(a, _releases(repo, code, a["accrual_code"])) for a in rows],
"pending_reversals": [str(j["journal_id"]) for j in repo.journals.pending_reversals(code)]}
_ = ledgerops