dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/closing.py
"""The closing cockpit (GL-57, GL-58): a month-end checklist where the system ticks what it can check and people tick the rest,
and a second person signs the close off.
A task list belongs to a company and a kind (month, or year for the last period). Starting a close run instantiates every task
with its owner role, its due business day (counted in the business time zone from the first day after the period) and its
dependencies. A SYSTEM task is met when its readiness gate is met - checked on read, never ticked by hand; a MANUAL task is done
with a note or evidence (an object key under raw/close/) or marked not applicable with a reason. The sign-off records who, when
and a digest of the trial balance; the preparer (the run's starter) and anyone who completed a task cannot sign (email AND sub),
and a posting into the period afterwards makes the sign-off stale.
"""
from datetime import date, timedelta
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from pydantic import Field
from brick_gl.application import gates as gate_mod
from brick_gl.application import ledgerops
from brick_gl.application.brick import (
CLOSE_CHECK,
CLOSE_MANAGE,
CLOSE_SIGNOFF,
CLOSE_TASK,
GL,
PERIOD_READ,
company_row,
decide,
real_now,
repository,
same_person,
sub_of,
today,
who,
)
from brick_gl.application.inputs import Code, CompanyCode, PeriodId, Reason, _In
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.readiness import refusal
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, utc
_list = lambda repo: _Keyed(repo.conn, t.gl_close_task_list, ("task_list_id", "task_id")) # noqa: E731
_run = lambda repo: _Keyed(repo.conn, t.gl_close_run, ("run_id",)) # noqa: E731
_task = lambda repo: _Keyed(repo.conn, t.gl_close_task, ("run_id", "task_id")) # noqa: E731
class TaskModel(_In):
task_id: Code
title: Annotated[str, Field(max_length=200)]
kind_of_task: Code = "manual"
gate: Optional[Code] = None
owner_role: Optional[Code] = None
due_business_day: Annotated[int, Field(ge=0, le=31)] = 1
depends_on: Annotated[list[Code], Field(max_length=50)] = []
required: bool = True
def business_day(after: date, n: int) -> date:
"""The n-th business day (Monday to Friday) on or after the day following `after`."""
d, count = after, 0
while count < max(n, 1):
d += timedelta(days=1)
if d.weekday() < 5:
count += 1
return d
@GL.use_case(kind="command", route="POST /close-task-lists", permission=CLOSE_MANAGE, idempotent=True)
def keep_close_task_list(caller: dict, task_list_id: Code, kind: Code, tasks: Annotated[list[TaskModel], Field(max_length=200)],
company_code: Optional[CompanyCode] = None) -> dict:
"""Define a company's month-end (kind month) or year-end (kind year) checklist: each task is a SYSTEM task (met when its readiness
gate is: recurring_posted, accruals_posted, allocations_posted, depreciation_posted, revalued, ic_matched,
receivable_reconciled, payable_reconciled, resplit_done ...) or a MANUAL one (an owner role and a due business day), with
dependencies. Saving the list again replaces it."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
if kind not in ("month", "year"):
raise RuleBroken("kind is month or year", status=422)
if not tasks:
raise RuleBroken("a task list has at least one task", status=422)
ids = {tk.task_id for tk in tasks}
if len(ids) != len(tasks):
raise RuleBroken("task ids are unique within a list", status=422)
for tk in tasks:
if tk.kind_of_task not in ("system", "manual"):
raise RuleBroken(f"{tk.task_id}: kind_of_task is system or manual", status=422)
if tk.kind_of_task == "system" and tk.gate not in gate_mod.GATE_NAMES:
raise RuleBroken(f"{tk.task_id}: a system task names one readiness gate: {', '.join(gate_mod.GATE_NAMES)}", status=422)
for d in tk.depends_on:
if d not in ids:
raise RuleBroken(f"{tk.task_id} depends on {d}, which is not in the list", status=422)
repo.conn.execute(t.gl_close_task_list.delete().where(t.gl_close_task_list.c.task_list_id == task_list_id))
for tk in tasks:
_list(repo).add({"task_list_id": task_list_id, "task_id": tk.task_id, "company_code": company["company_code"], "kind": kind,
"title": tk.title, "kind_of_task": tk.kind_of_task, "gate": tk.gate, "owner_role": tk.owner_role,
"due_business_day": tk.due_business_day, "depends_on": ",".join(tk.depends_on), "required": tk.required})
return {"task_list_id": task_list_id, "kind": kind, "tasks": len(tasks)}
def _run_row(repo, caller, run_id: str) -> dict:
r = _run(repo).get(run_id)
if r is None:
raise Refused(404, f"no close run {run_id}")
company_row(repo, caller, r["company_code"])
return r
def _evaluate(repo, run: dict) -> dict:
"""Compute the run's live state: each task (system ones from their gates), what is late, the digest and whether a sign-off is stale."""
company = repo.companies.get(run["company_code"])
period = repo.periods.get(run["company_code"], run["period_id"])
settings = repo.settings.in_force(today(repo), run["company_code"])
td = today(repo)
tasks = {r["task_id"]: r for r in (dict(x._mapping) for x in repo.conn.execute(t.gl_close_task.select().where(t.gl_close_task.c.run_id == run["run_id"])))}
view, all_done = [], True
for tid, tk in sorted(tasks.items()):
status, why = tk["status"], ""
if tk["kind_of_task"] == "system":
g = gate_mod.named(repo, company, period, settings, tk["gate"])
status = "done" if g.met else "open"
why = "" if g.met else g.reason
blocked = [d for d in (tk["depends_on"] or "").split(",") if d and tasks[d]["status"] not in ("done", "not_applicable") and not (
tasks[d]["kind_of_task"] == "system" and gate_mod.named(repo, company, period, settings, tasks[d]["gate"]).met)]
if status not in ("done", "not_applicable"):
all_done = False
view.append({"task_id": tid, "title": tk["title"], "kind": tk["kind_of_task"], "gate": tk["gate"], "owner_role": tk["owner_role"],
"due_on": tk["due_on"], "depends_on": [d for d in (tk["depends_on"] or "").split(",") if d], "status": status,
"blocked_by": blocked, "late": bool(tk["due_on"] and tk["due_on"] < td and status not in ("done", "not_applicable")),
"reason": why or tk["reason"], "done_by": tk["done_by"], "done_at": tk["done_at"], "note": tk["note"],
"evidence_key": tk["evidence_key"]})
digest = ledgerops.tb_digest(repo, company, period)
status, stale = run["status"], ""
if status in ("ready", "signed") and run["tb_digest"] != digest:
status, stale = "in_progress", "a posting landed in the period after the digest: finish again and ask for a new sign-off"
elif status == "in_progress" and all_done:
status = "ready"
return {"run": run, "tasks": view, "all_done": all_done, "status": status, "stale_reason": stale or run["stale_reason"] if status == "in_progress" else "",
"digest": digest, "completers": {tk["done_by"] for tk in tasks.values() if tk["done_by"]} | {tk["done_sub"] for tk in tasks.values() if tk["done_sub"]}}
def _view(repo, ev: dict) -> dict:
r = ev["run"]
return {"run_id": r["run_id"], "company_code": r["company_code"], "period_id": r["period_id"], "kind": r["kind"], "status": ev["status"],
"task_list_id": r["task_list_id"], "started_by": r["started_by"], "signed_by": r["signed_by"], "signed_at": r["signed_at"],
"tb_digest": r["tb_digest"], "stale_reason": ev["stale_reason"], "tasks": ev["tasks"],
"late": [tk["task_id"] for tk in ev["tasks"] if tk["late"]], "ready_to_sign": ev["all_done"] and ev["status"] in ("ready", "signed")}
@GL.use_case(kind="command", route="POST /periods/{period_id}/start-close-run", permission=CLOSE_MANAGE, idempotent=True)
def start_close_run(caller: dict, period_id: PeriodId, task_list_id: Optional[Code] = None, kind: Optional[Code] = None,
company_code: Optional[CompanyCode] = None) -> dict:
"""Start the close run of a company's period: every task of its checklist is instantiated with its owner, its due business day
(in business_timezone, counted from the day after the period ends) and its dependencies. The run for the LAST period of a year
takes the year-end list. Starting again answers the run that exists."""
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}")
last = repo.periods.of_company(code, p["fiscal_year"])[-1]["period_id"] == period_id
k = kind or ("year" if last else "month")
run_id = f"{code}:{period_id}:{k}"
if _run(repo).get(run_id):
return _view(repo, _evaluate(repo, _run(repo).get(run_id)))
lists = [dict(r._mapping) for r in repo.conn.execute(t.gl_close_task_list.select().where(
t.gl_close_task_list.c.company_code == code, t.gl_close_task_list.c.kind == k).order_by(t.gl_close_task_list.c.task_list_id, t.gl_close_task_list.c.task_id))]
if task_list_id:
lists = [r for r in lists if r["task_list_id"] == task_list_id]
if not lists:
raise refusal("task_list", f"no {k}-end task list exists for {code}", "Close cockpit: define one (keep_close_task_list)")
chosen = lists[0]["task_list_id"]
at = real_now()
_run(repo).add({"run_id": run_id, "company_code": code, "period_id": period_id, "kind": k, "status": "in_progress", "task_list_id": chosen,
"started_by": who(caller), "started_sub": sub_of(caller), "tb_digest": None, "signed_by": None, "signed_sub": None,
"signed_at": None, "stale_reason": None})
for r in [x for x in lists if x["task_list_id"] == chosen]:
_task(repo).add({"run_id": run_id, "task_id": r["task_id"], "title": r["title"], "kind_of_task": r["kind_of_task"], "gate": r["gate"],
"owner_role": r["owner_role"], "due_on": business_day(p["ends_on"], r["due_business_day"] or 1),
"depends_on": r["depends_on"], "status": "open", "done_by": None, "done_sub": None, "done_at": None, "note": None,
"evidence_key": None, "reason": None})
_ = at
return _view(repo, _evaluate(repo, _run(repo).get(run_id)))
@GL.use_case(kind="query", route="GET /close-runs/{run_id}", permission=PERIOD_READ)
def get_close_run(caller: dict, run_id: Annotated[str, Field(max_length=64)]) -> dict:
"""The close run: each task with its owner, due business day, dependencies, status and evidence; a system task shows its gate's
reason while open; what is late; whether the sign-off is stale (a posting landed in the period after the digest)."""
with repository() as repo:
return _view(repo, _evaluate(repo, _run_row(repo, caller, run_id)))
@GL.use_case(kind="query", route="GET /close-runs", permission=PERIOD_READ)
def list_close_runs(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""A company's close runs with their status."""
with repository() as repo:
company = company_row(repo, caller, company_code)
runs = [dict(r._mapping) for r in repo.conn.execute(t.gl_close_run.select().where(t.gl_close_run.c.company_code == company["company_code"]))]
out = []
for r in runs:
ev = _evaluate(repo, r)
out.append({"run_id": r["run_id"], "period_id": r["period_id"], "kind": r["kind"], "status": ev["status"], "signed_by": r["signed_by"]})
return {"runs": out}
@GL.use_case(kind="command", route="POST /close-runs/{run_id}/complete-close-task", permission=CLOSE_TASK)
def complete_close_task(caller: dict, run_id: Annotated[str, Field(max_length=64)], task_id: Code, note: Optional[Annotated[str, Field(max_length=2000)]] = None,
evidence_key: Optional[Annotated[str, Field(max_length=500)]] = None) -> dict:
"""Complete a MANUAL task with a note or evidence (an object key under raw/close/): who and when are recorded. A task cannot
complete before the tasks it depends on (gate dependencies_done); a system task is never ticked by hand."""
with repository() as repo, decide():
run = _run_row(repo, caller, run_id)
ev = _evaluate(repo, run)
if run["status"] == "signed" and ev["status"] == "signed":
raise RuleBroken("the close run is signed off: ask for it to be reopened by posting or reopening the period", status=409)
tk = _task(repo).get(run_id, task_id)
if tk is None:
raise Refused(404, f"no task {task_id} in {run_id}")
if tk["kind_of_task"] != "manual":
raise RuleBroken(f"{task_id} is a system task: it is met when its gate is, never ticked by hand", status=409)
if not note and not evidence_key:
raise RuleBroken("complete a task with a note or an evidence key", status=422)
if evidence_key and not evidence_key.startswith("raw/close/"):
raise RuleBroken("evidence is an object under raw/close/", status=422)
view = {v["task_id"]: v for v in ev["tasks"]}
blocked = view[task_id]["blocked_by"]
if blocked:
raise refusal("dependencies_done", f"a task it depends on is open: {', '.join(blocked)}", "finish that first")
_task(repo).set((run_id, task_id), status="done", done_by=who(caller), done_sub=sub_of(caller), done_at=utc(real_now()), note=note,
evidence_key=evidence_key)
return _view(repo, _evaluate(repo, run))
@GL.use_case(kind="command", route="POST /close-runs/{run_id}/skip-close-task", permission=CLOSE_MANAGE)
def skip_close_task(caller: dict, run_id: Annotated[str, Field(max_length=64)], task_id: Code, reason: Reason) -> dict:
"""Mark a task not applicable, with the reason."""
with repository() as repo, decide():
run = _run_row(repo, caller, run_id)
tk = _task(repo).get(run_id, task_id)
if tk is None:
raise Refused(404, f"no task {task_id} in {run_id}")
_task(repo).set((run_id, task_id), status="not_applicable", reason=reason, done_by=who(caller), done_sub=sub_of(caller), done_at=utc(real_now()))
return _view(repo, _evaluate(repo, run))
@GL.use_case(kind="command", route="POST /close-runs/{run_id}/mark-close-ready", permission=CLOSE_CHECK, idempotent=True)
def mark_close_ready(caller: dict, run_id: Annotated[str, Field(max_length=64)]) -> dict:
"""The system checks the run: when every task is done or not applicable and every system task's gate is met, the run is ready and
the trial balance digest is taken. A posting into the period afterwards makes a ready run in progress again."""
with repository() as repo, decide():
run = _run_row(repo, caller, run_id)
ev = _evaluate(repo, run)
if ev["all_done"] and run["status"] != "signed":
_run(repo).set((run_id,), status="ready", tb_digest=ev["digest"], stale_reason=None)
elif run["status"] in ("ready", "signed") and ev["status"] == "in_progress":
_run(repo).set((run_id,), status="in_progress", stale_reason=ev["stale_reason"])
return _view(repo, _evaluate(repo, _run(repo).get(run_id)))
@GL.use_case(kind="command", route="POST /close-runs/{run_id}/invalidate-close-signoff", permission=CLOSE_CHECK, idempotent=True)
def invalidate_close_signoff(caller: dict, run_id: Annotated[str, Field(max_length=64)]) -> dict:
"""The system marks a ready or signed run stale when a posting landed in the period after its digest: the run goes back to
in progress and asks for a new sign-off (the reads show it already; this records it)."""
with repository() as repo, decide():
run = _run_row(repo, caller, run_id)
ev = _evaluate(repo, run)
if run["status"] in ("ready", "signed") and ev["status"] == "in_progress" and ev["stale_reason"]:
_run(repo).set((run_id,), status="in_progress", stale_reason=ev["stale_reason"], signed_by=None, signed_sub=None, signed_at=None)
return _view(repo, _evaluate(repo, _run(repo).get(run_id)))
@GL.use_case(kind="command", route="POST /close-runs/{run_id}/sign-off-close", permission=CLOSE_SIGNOFF)
def sign_off_close(caller: dict, run_id: Annotated[str, Field(max_length=64)]) -> dict:
"""Sign off the close: another person than whoever prepared it (the run's starter) or completed one of its tasks, matched on email
AND sub (gate second_person). Records who, when and the digest of the trial balance. With close_cockpit on, close_period is
refused until the run is signed; a posting into the period after the sign-off makes it stale and asks for it again."""
with repository() as repo, decide():
run = _run_row(repo, caller, run_id)
ev = _evaluate(repo, run)
if run["status"] == "signed" and ev["status"] == "signed":
return _view(repo, ev)
if not ev["all_done"]:
open_ = [tk["task_id"] for tk in ev["tasks"] if tk["status"] not in ("done", "not_applicable")]
raise refusal("tasks_done", f"tasks are still open: {', '.join(open_)}", "finish them first", status=409)
tasks = [dict(x._mapping) for x in repo.conn.execute(t.gl_close_task.select().where(t.gl_close_task.c.run_id == run_id))]
people = [(run["started_by"], run["started_sub"])] + [(tk["done_by"], tk["done_sub"]) for tk in tasks if tk["status"] == "done" and tk["kind_of_task"] == "manual"]
if any(same_person(caller, a, b) for a, b in people):
raise refusal("second_person", "you started the run or completed a task in it", "ask another accountant")
_run(repo).set((run_id,), status="signed", tb_digest=ev["digest"], signed_by=who(caller), signed_sub=sub_of(caller), signed_at=utc(real_now()),
stale_reason=None)
return _view(repo, _evaluate(repo, _run(repo).get(run_id)))
_ = dates