dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/periods.py
"""Periods and financial years: open a year, close a month, reopen it, close the year (GL-07, GL-13, GL-24, GL-32).
A period is closed PER COMPANY (DE01's 2026-09 stays open until its own close). The lock is a guard on the one write path:
every posting writes the period row in its own transaction (ledgerops.assert_period_still_open), and so does the close, so a
close racing a posting conflicts on that row and the loser sees closed. Closing needs every gate met; reopening is a
deliberate act with a reason, by someone other than the closer where the owner asks for it, and it is recorded.
"""
from typing import Optional
from erp_platform.bricks import Refused
from sqlalchemy import update
from brick_gl.application import carry, gates, ledgerops
from brick_gl.application.brick import (
BALANCE_DERIVE,
CARRY_DERIVE,
GL,
PERIOD_CLOSE,
PERIOD_READ,
PERIOD_REOPEN,
company_row,
decide,
real_now,
repository,
same_person,
sub_of,
today,
who,
)
from brick_gl.application.inputs import CompanyCode, FiscalYearName, IsoDate, PeriodId, Reason
from brick_gl.application.posting import post_due_reversals
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.readiness import GatesUnmet, met, refusal, unmet
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import utc
def _period(repo, company_code: str, period_id: str) -> dict:
p = repo.periods.get(company_code, period_id)
if p is None:
raise Refused(404, f"no period {period_id} of {company_code}")
return p
def _period_view(p: dict) -> dict:
return {k: p[k] for k in ("company_code", "period_id", "fiscal_year", "starts_on", "ends_on", "status", "closed_by", "closed_at",
"statements_stale")}
# ── financial years ──────────────────────────────────────────────────────────────────────────────────────────────────
def _open_year_rows(repo, company: dict, year: int, at) -> bool:
"""Create the year and its twelve open periods when they do not exist: True when anything was created."""
code, endm = company["company_code"], int(company["financial_year_end_month"])
name = dates.fy_name(year)
start, end = dates.fy_bounds(year, endm)
made = False
if repo.years.get(code, name) is None:
repo.years.add({"company_code": code, "fiscal_year": name, "starts_on": start, "ends_on": end, "status": "open",
"closed_by": None, "closed_at": None})
made = True
have = {p["period_id"] for p in repo.periods.of_company(code, name)}
for pid, s, e in dates.fy_periods(year, endm):
if pid not in have:
repo.periods.add({"company_code": code, "period_id": pid, "fiscal_year": name, "starts_on": s, "ends_on": e,
"status": "open", "closed_by": None, "closed_sub": None, "closed_at": None,
"last_posted_at": None, "statements_stale": False})
made = True
return made
@GL.use_case(kind="command", route="POST /fiscal-years/open-year", permission=PERIOD_CLOSE, idempotent=True)
def open_year(caller: dict, fiscal_year: Optional[FiscalYearName] = None, company_code: Optional[CompanyCode] = None) -> dict:
"""Open a company's financial year (named by the year it ends in: FY2026): twelve open periods (period_id YYYY-MM) that follow
the company's previous year end, and the carry-forward from the previous year derived (no closing journal). Without a
year it opens the one after the last (the first year must be named). Opening again changes nothing. Pending reversals dated
in the new periods post (GL-24)."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
years = repo.years.of_company(code)
if fiscal_year is None:
if not years:
raise RuleBroken("name the first financial year to open (FY2026)", status=422)
year = dates.fy_number(years[-1]["fiscal_year"]) + 1
else:
year = dates.fy_number(fiscal_year)
name = dates.fy_name(year)
if years and repo.years.get(code, name) is None:
last = dates.fy_number(years[-1]["fiscal_year"])
if year != last + 1 and year > last:
raise refusal("year_sequence", f"{name} does not follow {years[-1]['fiscal_year']}: a company's years run without a gap",
f"open {dates.fy_name(last + 1)} first")
if year < dates.fy_number(years[0]["fiscal_year"]):
raise refusal("year_sequence", f"{name} is before the company's first year {years[0]['fiscal_year']}",
"a company's history is a migration")
at = real_now()
_open_year_rows(repo, company, year, at)
carried = carry.derive(repo, company, year, at)
out = {"company_code": code, "fiscal_year": name, "periods": [p["period_id"] for p in repo.periods.of_company(code, name)] or
[pid for pid, _, _ in dates.fy_periods(year, int(company["financial_year_end_month"]))],
"carry_forward_rows": carried}
post_due_reversals(code)
return out
@GL.use_case(kind="command", route="POST /periods/open-periods", permission=PERIOD_CLOSE, idempotent=True)
def open_periods(caller: dict, fiscal_year: FiscalYearName, company_code: Optional[CompanyCode] = None) -> dict:
"""Open the periods of an existing financial year that are missing and post the pending reversals dated in them: one per
month, period_id YYYY-MM, per company. A second call, or a retry, posts nothing twice (each reversal is keyed
derived_id(journal_id, 'reversal'))."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
if repo.years.get(code, fiscal_year) is None:
raise refusal("year_open", f"{fiscal_year} is not opened for {code}", "open_year first")
made = _open_year_rows(repo, company, dates.fy_number(fiscal_year), real_now())
posted = post_due_reversals(code)
return {"company_code": code, "fiscal_year": fiscal_year, "opened": made, "reversals_posted": posted}
@GL.use_case(kind="query", route="GET /fiscal-years", permission=PERIOD_READ)
def list_fiscal_years(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""A company's financial years with their dates and whether each is closed to adjustments."""
with repository() as repo:
company = company_row(repo, caller, company_code)
return {"company_code": company["company_code"],
"years": [{k: y[k] for k in ("fiscal_year", "starts_on", "ends_on", "status", "closed_by", "closed_at")}
for y in repo.years.of_company(company["company_code"])]}
@GL.use_case(kind="query", route="GET /periods", permission=PERIOD_READ)
def list_periods(caller: dict, company_code: Optional[CompanyCode] = None, fiscal_year: Optional[FiscalYearName] = None) -> dict:
"""A company's periods with their status, who closed each and when, the pending reversals waiting for the next period to
open, and the close and reopen history."""
with repository() as repo:
company = company_row(repo, caller, company_code)
code = company["company_code"]
periods = repo.periods.of_company(code, fiscal_year)
pending = repo.journals.pending_reversals(code)
out = []
for p in periods:
v = _period_view(p)
v["history"] = [{"event": e["event"], "reason": e["reason"], "by": e["by_email"], "at": e["event_at"]}
for e in repo.periods.events(code, p["period_id"])]
v["pending_reversals"] = [str(j["journal_id"]) for j in pending
if dates.period_id(dates.first_of_next_month(j["journal_date"])) == p["period_id"]]
out.append(v)
return {"company_code": code, "periods": out}
@GL.use_case(kind="query", route="GET /periods/open", permission=PERIOD_READ)
def check_period_open(caller: dict, date: IsoDate, company_code: Optional[CompanyCode] = None,
ledger_id: Optional[str] = None) -> dict:
"""Is a date open for posting (invoicing, payments, inventory, purchasing and fixed-assets ask before they issue a document,
so they refuse before issuing rather than after)? {date, company_code, period_id, open, status: open|closed|none, reason,
fix}; status none and open false when no period covers the date. The default company when none is named."""
with repository() as repo:
company = company_row(repo, caller, company_code)
on = dates.parse(date)
p = repo.periods.for_date(company["company_code"], on)
code = company["company_code"]
if p is None:
return {"date": on, "company_code": code, "period_id": None, "open": False, "status": "none",
"reason": f"no period of {code} covers {on}", "fix": "ask the accountant to open the financial year"}
if p["status"] == "open":
return {"date": on, "company_code": code, "period_id": p["period_id"], "open": True, "status": "open", "reason": "", "fix": ""}
return {"date": on, "company_code": code, "period_id": p["period_id"], "open": False, "status": "closed",
"reason": f"the period {p['period_id']} of {code} is closed", "fix": "date it in an open period, or ask the accountant to reopen"}
# ── closing and reopening a month ────────────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="query", route="GET /periods/{period_id}/readiness", permission=PERIOD_READ)
def get_period_readiness(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None) -> dict:
"""Every gate of closing a period as {gate, met, reason, fix}: the previous period closed, no manual journal waiting, the
receivable and payables control accounts reconciled (when payments / purchasing are installed), balances and open items
revalued, recurring entries, accruals and allocations posted, depreciation posted (fixed-assets installed), intercompany
matched, open items re-split, pending reversals posted, the carry-forward current, the close signed off (cockpit on) and the
trial balance at zero. The Periods screen and the close read the same list."""
with repository() as repo:
company = company_row(repo, caller, company_code)
p = _period(repo, company["company_code"], period_id)
settings = repo.settings.in_force(today(repo), company["company_code"])
gs = gates.period_gates(repo, company, p, settings)
return {"company_code": company["company_code"], "period_id": period_id, "status": p["status"],
"ready": all(g.met for g in gs), "gates": [g.as_dict() for g in gs]}
@GL.use_case(kind="command", route="POST /periods/{period_id}/close-period", permission=PERIOD_CLOSE, idempotent=True)
def close_period(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None) -> dict:
"""Close a month of a company so nobody can change history after it has been reported: it needs every gate of
get_period_readiness met, and is refused with the unmet gates' own words. Operational postings dated in it are then
refused naming the period (a year-end audit adjustment is the exception); other companies' periods are untouched.
Closing twice changes nothing."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
p = _period(repo, code, period_id)
if p["status"] == "closed":
return _period_view(p)
settings = repo.settings.in_force(today(repo), code)
gs = gates.period_gates(repo, company, p, settings)
if any(not g.met for g in gs):
raise GatesUnmet(f"close {period_id}", gs)
at = real_now()
res = repo.conn.execute(update(t.gl_period).where(t.gl_period.c.company_code == code, t.gl_period.c.period_id == period_id,
t.gl_period.c.status == "open").values(
status="closed", closed_by=who(caller), closed_sub=sub_of(caller), closed_at=utc(at), updated_at=utc(at)))
if res.rowcount == 0:
return _period_view(repo.periods.get(code, period_id))
repo.periods.log(code, period_id, "close", "", who(caller), sub_of(caller), at)
from brick_gl.application.reports import materialise
materialise(repo, company, p, at)
return _period_view(repo.periods.get(code, period_id))
@GL.use_case(kind="command", route="POST /periods/{period_id}/reopen-period", permission=PERIOD_REOPEN)
def reopen_period(caller: dict, period_id: PeriodId, reason: Reason, company_code: Optional[CompanyCode] = None) -> dict:
"""Reopen a closed month, deliberately: with a reason, only while the financial year is open, and by someone other than
the closer where the owner asks for it (reopen_second_person; matched on email AND sub). Recorded; the materialised
statements and any close sign-off go stale."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
p = _period(repo, code, period_id)
if p["status"] != "closed":
raise RuleBroken(f"{period_id} is not closed", status=409)
year = repo.years.get(code, p["fiscal_year"])
if year and year["status"] == "closed":
raise refusal("year_open", f"the financial year {p['fiscal_year']} is closed", "post a correction in the current year")
settings = repo.settings.in_force(today(repo), code)
if settings.reopen_second_person and same_person(caller, p["closed_by"], p["closed_sub"]):
raise refusal("second_person", f"you closed {period_id}, and the owner asks for a second person to reopen it",
"ask another accountant")
at = real_now()
repo.conn.execute(update(t.gl_period).where(t.gl_period.c.company_code == code, t.gl_period.c.period_id == period_id).values(
status="open", statements_stale=True, closed_by=None, closed_sub=None, closed_at=None, updated_at=utc(at)))
repo.periods.log(code, period_id, "reopen", reason, who(caller), sub_of(caller), at)
repo.conn.execute(update(t.gl_close_run).where(
t.gl_close_run.c.company_code == code, t.gl_close_run.c.period_id == period_id,
t.gl_close_run.c.status.in_(("ready", "signed"))).values(status="in_progress", stale_reason=f"{period_id} was reopened",
updated_at=utc(at)))
return _period_view(repo.periods.get(code, period_id))
# ── closing the year, the carry-forward ──────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="command", route="POST /fiscal-years/{fiscal_year}/close-year", permission=PERIOD_CLOSE, idempotent=True)
def close_year(caller: dict, fiscal_year: FiscalYearName, company_code: Optional[CompanyCode] = None) -> dict:
"""Close a financial year: every period closed, no audit adjustment waiting, retained_earnings mapped. The carry-forward into
the next year is DERIVED for every ledger of the company (each balance-sheet account opens with its closing balance, per
profit centre, segment and partner company; the year's income and expense sum opens on retained earnings) - no closing
journal is posted. The year is then closed to adjustments and the next year opened if it is not. Closing twice derives
the same rows."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
y = repo.years.get(code, fiscal_year)
if y is None:
raise Refused(404, f"no financial year {fiscal_year} of {code}")
year = dates.fy_number(fiscal_year)
open_p = [p["period_id"] for p in repo.periods.of_company(code, fiscal_year) if p["status"] != "closed"]
gs = [unmet("periods_closed", f"periods of the year are still open: {', '.join(open_p)}", "close them (GL-07)")
if open_p else met("periods_closed")]
waiting = [j for p in repo.periods.of_company(code, fiscal_year) for j in repo.journals.proposed_in(code, p["period_id"])
if j["adjustment"]]
gs.append(unmet("no_proposed_adjustment", f"{len(waiting)} audit adjustment(s) are waiting for approval", "approve or reject it (GL-56)")
if waiting else met("no_proposed_adjustment"))
if repo.account_map.account(company["chart_id"], "retained_earnings", "", y["ends_on"]) is None:
gs.append(unmet("retained_earnings_mapped", f"no retained_earnings account is in the map on {y['ends_on']}", "Account map (GL-12)"))
else:
gs.append(met("retained_earnings_mapped"))
from brick_gl.domain.readiness import require
require(f"close {fiscal_year}", gs)
at = real_now()
nxt = year + 1
_open_year_rows(repo, company, nxt, at)
carried = carry.derive(repo, company, nxt, at)
if y["status"] != "closed":
repo.years.set((code, fiscal_year), status="closed", closed_by=who(caller), closed_at=utc(at))
return {"company_code": code, "fiscal_year": fiscal_year, "status": "closed", "next_year": dates.fy_name(nxt),
"carry_forward_rows": carried}
@GL.use_case(kind="command", route="POST /periods/{period_id}/rederive-balances", permission=BALANCE_DERIVE, idempotent=True)
def rederive_balances(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None) -> dict:
"""The system re-derives every account's balance of a company's period from its posted split lines, in every ledger the company
keeps: a SUM, never an increment, so running it again (or after any number of concurrent postings) writes the same rows."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code, scoped=False)
code = company["company_code"]
if repo.periods.get(code, period_id) is None:
raise Refused(404, f"no period {period_id} of {code}")
from sqlalchemy import select
ids = [r[0] for r in repo.conn.execute(select(t.gl_split_line.c.account_id).where(
t.gl_split_line.c.company_code == code, t.gl_split_line.c.period_id == period_id).distinct())]
return {"company_code": code, "period_id": period_id, "balances": ledgerops.derive_balances(repo, code, period_id, ids)}
@GL.use_case(kind="command", route="POST /fiscal-years/{fiscal_year}/derive-carry-forward", permission=CARRY_DERIVE, idempotent=True)
def derive_carry_forward(caller: dict, fiscal_year: FiscalYearName, company_code: Optional[CompanyCode] = None) -> dict:
"""The system derives a year's opening balances from the previous year's closing ones: balance-sheet accounts open with
their closing balance (per ledger, profit centre, segment, partner company), income and expense open at zero and their sum
opens on retained earnings. Written with every value in VALUES, so deriving again writes the same rows - also after an
audit adjustment or a reopened period moved the previous year."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code, scoped=False)
year = dates.fy_number(fiscal_year)
if repo.years.get(company["company_code"], fiscal_year) is None:
raise Refused(404, f"no financial year {fiscal_year} of {company['company_code']}")
wrote = carry.derive(repo, company, year, real_now())
return {"company_code": company["company_code"], "fiscal_year": fiscal_year, "rows": wrote}
__all__ = ["ledgerops"]