dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/budgets.py
"""Budgets (GL-55): a plan per account, profit centre and month, approved and FROZEN, so plan against actual means the same thing
all year. A budget never posts to the ledger.
draft -> submitted (the submitter recorded, email and sub) -> approved by someone else (frozen: lines refused 409). A revision is
version + 1, approved the same way, and the plan/actual report names the version. Amounts are bounded NUMERIC(18,2) (422 above);
a plan amount is in the account's NORMAL sign (revenue and costs both positive).
"""
from datetime import date
from decimal import Decimal
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId
from pydantic import Field
from sqlalchemy import func, select
from brick_gl.application.brick import (
BUDGET_APPROVE,
BUDGET_WRITE,
GL,
REPORT_READ,
company_row,
decide,
ledger_for,
repository,
same_person,
sub_of,
who,
)
from brick_gl.application.inputs import (
Code,
CompanyCode,
FiscalYearName,
IsoDate,
LedgerCode,
Money,
PeriodId,
Reason,
_In,
)
from brick_gl.application.reports import _window, trial_balance_data
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.money import ZERO, q2
from brick_gl.domain.readiness import refusal
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed
_bud = lambda repo: _Keyed(repo.conn, t.gl_budget, ("budget_id", "version")) # noqa: E731
class BudgetLineModel(_In):
account_id: Optional[BigId] = None
account_code: Optional[Code] = None
profit_centre: Optional[Code] = None
cost_centre: Optional[Code] = None
period_id: PeriodId
amount: Money
def _get(repo, caller, budget_id: str, version: Optional[int] = None) -> dict:
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_budget.select().where(t.gl_budget.c.budget_id == budget_id).order_by(t.gl_budget.c.version))]
if not rows:
raise Refused(404, f"no budget {budget_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 {budget_id}")
company_row(repo, caller, row["company_code"])
return row
def _lines(repo, budget_id: str, version: int) -> list:
return [dict(r._mapping) for r in repo.conn.execute(t.gl_budget_line.select().where(
t.gl_budget_line.c.budget_id == budget_id, t.gl_budget_line.c.version == version))]
def _view(repo, b: dict, with_lines: bool = True) -> dict:
out = {k: b[k] for k in ("budget_id", "version", "company_code", "ledger_id", "fiscal_year", "status", "submitted_by", "approved_by", "return_reason")}
if with_lines:
out["lines"] = [{k: ln[k] for k in ("account_id", "profit_centre", "cost_centre", "period_id", "amount")} for ln in _lines(repo, b["budget_id"], b["version"])]
return out
def _resolve(repo, company: dict, models: list, fiscal_year: str) -> list:
rows, seen = [], set()
periods = {p["period_id"] for p in repo.periods.of_company(company["company_code"], fiscal_year)}
for m in models:
if m.account_id is not None:
a = repo.accounts.get(int(m.account_id))
else:
a = repo.accounts.by_code(company["chart_id"], m.account_code or "")
if a is None or a["chart_id"] != company["chart_id"]:
raise RuleBroken(f"there is no account {m.account_code or m.account_id} in chart {company['chart_id']}", status=422)
if periods and m.period_id not in periods:
raise RuleBroken(f"{m.period_id} is not a period of {fiscal_year}", status=422)
key = (a["account_id"], m.profit_centre or "", m.cost_centre or "", m.period_id)
if key in seen:
raise RuleBroken(f"the budget names {a['code']} {m.period_id} twice", status=422)
seen.add(key)
rows.append({"account_id": a["account_id"], "profit_centre": m.profit_centre or "", "cost_centre": m.cost_centre or "",
"period_id": m.period_id, "amount": q2(m.amount)})
return rows
def _put_lines(repo, b: dict, rows: list) -> None:
repo.conn.execute(t.gl_budget_line.delete().where(t.gl_budget_line.c.budget_id == b["budget_id"], t.gl_budget_line.c.version == b["version"]))
K = _Keyed(repo.conn, t.gl_budget_line, ("budget_id", "version", "account_id", "profit_centre", "cost_centre", "period_id"))
for r in rows:
K.add({"budget_id": b["budget_id"], "version": b["version"], **r})
@GL.use_case(kind="command", route="POST /budgets", permission=BUDGET_WRITE)
def draft_budget(caller: dict, budget_id: Code, fiscal_year: FiscalYearName,
lines: Annotated[list[BudgetLineModel], Field(max_length=50000)], company_code: Optional[CompanyCode] = None,
ledger_id: Optional[LedgerCode] = None, csv: Optional[Annotated[str, Field(max_length=2000000)]] = None) -> dict:
"""Draft a budget (version 1) for a company, ledger and financial year: lines by account, profit centre, cost centre and period,
typed or imported from a CSV of the same columns (account_code,profit_centre,cost_centre,period_id,amount). Amounts are bounded
NUMERIC(18,2) (422 above). Drafting an existing draft replaces its lines; an approved budget is frozen (409)."""
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 Refused(404, f"no financial year {fiscal_year} of {code}")
lg = ledger_for(repo, code, ledger_id, dates.fy_number(fiscal_year))
models = list(lines)
if csv:
for i, ln in enumerate(csv.strip().splitlines()):
parts = [p.strip() for p in ln.split(",")]
if i == 0 and parts[:1] == ["account_code"]:
continue
if len(parts) != 5:
raise Refused(422, f"csv line {i + 1}: expected account_code,profit_centre,cost_centre,period_id,amount")
try:
models.append(BudgetLineModel(account_code=parts[0], profit_centre=parts[1] or None, cost_centre=parts[2] or None,
period_id=parts[3], amount=parts[4]))
except ValueError as e:
raise Refused(422, f"csv line {i + 1}: {e}") from None
rows = _resolve(repo, company, models, fiscal_year)
existing = [dict(r._mapping) for r in repo.conn.execute(t.gl_budget.select().where(t.gl_budget.c.budget_id == budget_id).order_by(t.gl_budget.c.version))]
if existing:
last = existing[-1]
if last["status"] != "draft":
raise RuleBroken(f"{budget_id} version {last['version']} is {last['status']}: revise_budget drafts the next version, an "
"approved budget's lines are frozen", status=409)
b = last
else:
b = {"budget_id": budget_id, "version": 1, "company_code": code, "ledger_id": lg, "fiscal_year": fiscal_year, "status": "draft",
"submitted_by": None, "submitted_sub": None, "approved_by": None, "approved_sub": None, "return_reason": None}
_bud(repo).add(b)
_put_lines(repo, b, rows)
return _view(repo, b)
@GL.use_case(kind="command", route="POST /budgets/{budget_id}/submit-budget", permission=BUDGET_WRITE)
def submit_budget(caller: dict, budget_id: Code, version: Optional[int] = None) -> dict:
"""Submit a drafted budget for approval: the submitter is recorded (email and sub). Every amount is already bounded."""
with repository() as repo, decide():
b = _get(repo, caller, budget_id, version)
if b["status"] != "draft":
raise RuleBroken(f"{budget_id} version {b['version']} is {b['status']}, not a draft", status=409)
if not _lines(repo, budget_id, b["version"]):
raise RuleBroken("a budget with no lines is not submitted", status=422)
_bud(repo).set((budget_id, b["version"]), status="submitted", submitted_by=who(caller), submitted_sub=sub_of(caller))
return _view(repo, _get(repo, caller, budget_id, b["version"]), False)
@GL.use_case(kind="command", route="POST /budgets/{budget_id}/approve-budget", permission=BUDGET_APPROVE)
def approve_budget(caller: dict, budget_id: Code, version: Optional[int] = None) -> dict:
"""Approve a submitted budget: it is frozen (its lines are refused 409) and plan/actual reads it. The approver is not the
submitter, matched on email AND sub (gate second_person)."""
with repository() as repo, decide():
b = _get(repo, caller, budget_id, version)
if b["status"] != "submitted":
raise RuleBroken(f"{budget_id} version {b['version']} is {b['status']}, not submitted", status=409)
if same_person(caller, b["submitted_by"], b["submitted_sub"]):
raise refusal("second_person", "you submitted it", "ask another accountant")
_bud(repo).set((budget_id, b["version"]), status="approved", approved_by=who(caller), approved_sub=sub_of(caller))
return _view(repo, _get(repo, caller, budget_id, b["version"]), False)
@GL.use_case(kind="command", route="POST /budgets/{budget_id}/return-budget", permission=BUDGET_APPROVE)
def return_budget(caller: dict, budget_id: Code, reason: Reason, version: Optional[int] = None) -> dict:
"""Return a submitted budget to its drafter with a reason."""
with repository() as repo, decide():
b = _get(repo, caller, budget_id, version)
if b["status"] != "submitted":
raise RuleBroken(f"{budget_id} version {b['version']} is {b['status']}, not submitted", status=409)
_bud(repo).set((budget_id, b["version"]), status="draft", return_reason=reason)
return _view(repo, _get(repo, caller, budget_id, b["version"]), False)
@GL.use_case(kind="command", route="POST /budgets/{budget_id}/revise-budget", permission=BUDGET_WRITE)
def revise_budget(caller: dict, budget_id: Code, reason: Reason) -> dict:
"""Revise an approved budget: it is superseded (plan/actual run on it keep naming it) and version + 1 is drafted as a copy, to be
submitted and approved the same way."""
with repository() as repo, decide():
b = _get(repo, caller, budget_id)
if b["status"] != "approved":
raise RuleBroken(f"{budget_id} version {b['version']} is {b['status']}: only an approved budget is revised", status=409)
_bud(repo).set((budget_id, b["version"]), status="superseded", return_reason=reason)
new = {**{k: b[k] for k in ("budget_id", "company_code", "ledger_id", "fiscal_year")}, "version": b["version"] + 1, "status": "draft",
"submitted_by": None, "submitted_sub": None, "approved_by": None, "approved_sub": None, "return_reason": None}
_bud(repo).add(new)
for ln in _lines(repo, budget_id, b["version"]):
_Keyed(repo.conn, t.gl_budget_line, ("budget_id", "version", "account_id", "profit_centre", "cost_centre", "period_id")).add(
{k: ln[k] for k in ("budget_id", "account_id", "profit_centre", "cost_centre", "period_id", "amount")} | {"version": new["version"]})
return _view(repo, new)
@GL.use_case(kind="query", route="GET /budgets/{budget_id}", permission=REPORT_READ)
def get_budget(caller: dict, budget_id: Code, version: Optional[int] = None) -> dict:
"""A budget version with its lines, status and who submitted and approved it."""
with repository() as repo:
return _view(repo, _get(repo, caller, budget_id, version))
def plan_by_account(repo, company: dict, ledger_id: str, budget_id: Optional[str], version: Optional[int], start: date, end: date) -> tuple:
"""({account_id: plan in the account's normal sign over the window}, the approved budget used). Only an APPROVED version is read."""
if not budget_id:
raise Refused(422, "name the budget (budget_id) to compare with")
q = [dict(r._mapping) for r in repo.conn.execute(t.gl_budget.select().where(
t.gl_budget.c.budget_id == budget_id, t.gl_budget.c.company_code == company["company_code"], t.gl_budget.c.status == "approved").order_by(
t.gl_budget.c.version))]
if version is not None:
q = [b for b in q if b["version"] == version]
if not q:
raise Refused(404, f"no approved version of budget {budget_id}")
b = q[-1]
plan: dict = {}
sp, ep = dates.period_id(start), dates.period_id(end)
for ln in _lines(repo, budget_id, b["version"]):
if sp <= ln["period_id"] <= ep:
plan[ln["account_id"]] = plan.get(ln["account_id"], ZERO) + ln["amount"]
return plan, b
@GL.use_case(kind="query", route="GET /reports/plan-actual", permission=REPORT_READ)
def get_plan_actual(caller: dict, budget_id: Code, period_id: Optional[PeriodId] = None, as_at: Optional[IsoDate] = None,
version: Optional[int] = None, company_code: Optional[CompanyCode] = None, by: Optional[Code] = None) -> dict:
"""Plan against actual from an approved budget version for a period (or the year to a date): per account actual, budget,
variance and variance %, and per profit centre when by=profit_centre. The report names the budget version; a budget never
posts to the ledger. Actual is in the account's normal sign, like the plan."""
with repository() as repo:
company = company_row(repo, caller, company_code)
year, start, end = _window(repo, company, as_at, period_id)
b_ledger = repo.conn.execute(select(t.gl_budget.c.ledger_id).where(t.gl_budget.c.budget_id == budget_id)).scalar()
lg = b_ledger or ledger_for(repo, company["company_code"], None, year)
if not period_id:
start = carry_start(company, year)
plan, b = plan_by_account(repo, company, lg, budget_id, version, start, end)
tb = trial_balance_data(repo, company, lg, year, start, end, by=by if by == "profit_centre" else None)
actual: dict = {}
for r in tb["rows"]:
a = repo.accounts.get(r["account_id"])
if a["type"] in ("income", "expense"):
val = r["debit"] - r["credit"]
actual[(r["account_id"], r.get("profit_centre", ""))] = val if a["normal_balance"] == "debit" else -val
rows = []
accounts = {aid for aid, _ in actual} | set(plan)
for aid in sorted(accounts, key=lambda x: repo.accounts.get(x)["code"]):
a = repo.accounts.get(aid)
act = sum((v for (k, _), v in actual.items() if k == aid), ZERO)
bud = plan.get(aid, ZERO)
var = act - bud
rows.append({"account_id": aid, "code": a["code"], "name": a["name"], "actual": act, "budget": bud, "variance": var,
"variance_pct": (var / bud * 100).quantize(Decimal("0.01")) if bud else None})
out = {"company_code": company["company_code"], "ledger_id": lg, "budget_id": budget_id, "budget_version": b["version"], "from": start,
"to": end, "rows": rows, "total_actual": sum((r["actual"] for r in rows), ZERO), "total_budget": sum((r["budget"] for r in rows), ZERO)}
if by == "profit_centre":
pcs = sorted({pc for (_, pc) in actual if pc})
out["by_profit_centre"] = [{"profit_centre": pc, "actual": sum((v for (_, p), v in actual.items() if p == pc), ZERO)} for pc in pcs]
return out
def carry_start(company: dict, year: int) -> date:
return dates.fy_bounds(year, int(company["financial_year_end_month"]))[0]
_ = func