dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/uc_depreciation.py
"""Depreciation (FA-12 .. FA-15): a run simulates a period, a second person approves it, the system posts one journal per
area to that area's ledger, and a reversal is a reversing run.
A run posts PLANNED-TO-DATE MINUS ALREADY-POSTED per asset and area, so a late invoice is a catch-up in the next run and a repeat
posts only the difference (zero when nothing changed). The plan is derived from rows (domain/depreciation.py); nothing
is stored until a run posts. Posting is claim-first: a plain INSERT of (company, period, area, post sequence) decides which
of two runs approved at once posts, and the loser re-reads and posts the difference — zero. Only then do journals go to gl
(their ids derive from the run, so a retry replays)."""
import json
from collections import defaultdict
from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import Limit
from brick_fixed_assets.application import services
from brick_fixed_assets.application import use_cases as hub
from brick_fixed_assets.application.inputs import Code, Period, Reason, Year
from brick_fixed_assets.domain import journals
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import (
ZERO,
fiscal_year,
fy_first_period,
fy_last_period,
met,
period_add,
period_end,
period_index,
period_of,
same_person,
unmet,
)
from brick_fixed_assets.infrastructure import seams
from brick_fixed_assets.infrastructure import tables as T
FA = hub.FA
def _run_id(company: str, period: str, run_no: int) -> str:
return f"{company}:{period}:{run_no}"
def _split_run_id(run_id: str) -> tuple:
try:
company, period, no = run_id.rsplit(":", 2)
return company, period, int(no)
except ValueError: # not a run id at all: there is no such run
raise Refused(404, f"no run {run_id}") from None
def compute_lines(repo, company_code: str, period: str) -> tuple:
"""([line], [Gate]) for a period: for each active asset and posting area, planned-to-date minus already posted."""
today = repo.today()
end = period_end(period)
areas = repo.posting_areas(today, company_code)
masters = repo.masters_on(end)
posted: dict = defaultdict(lambda: ZERO)
for p in repo.postings(company_code=company_code):
posted[(p["asset_id"], p["area_code"])] += p["amount"] or ZERO
lines, usage_missing, no_accounts = [], [], set()
roots = repo.asset_roots(company_code, statuses=["active"])
loaded = repo.assets(r["asset_id"] for r in roots)
repo.prefetch_plans(loaded) # every asset's posted rows and usage in two statements, not two per asset and area
try:
lines, usage_missing, no_accounts = _plan_lines(repo, roots, loaded, areas, masters, end, period, today, posted)
finally:
repo.clear_plans()
gates = [unmet("usage_missing", "no reading for the period: " + ", ".join(sorted(set(usage_missing))),
"the custodian records the units (zero is a reading)") if usage_missing else met("usage_missing"),
unmet("account_determination", "no account in force for: " + ", ".join(f"{c} in area {a}" for c, a in sorted(no_accounts)),
"Settings: account determination (owner)") if no_accounts else met("account_determination")]
journal_day = min(period_end(period), today)
gates += services.period_gates(journal_day, company_code, areas)
return lines, gates
def _plan_lines(repo, roots, loaded, areas, masters, end, period, today, posted) -> tuple:
lines, usage_missing, no_accounts = [], [], set()
for root in roots:
a = loaded[root["asset_id"]]
m = masters.get(a.asset_id) or a.master_on(end) or {}
for area in areas:
code = area["area_code"]
row = a.area_on(code, end)
if row is None:
continue
plan = services.build_plan(repo, a, code)
if plan is None:
continue
key = repo.key(row["key_code"], end)
if key and key["method"] == "units_of_production" and period not in repo.usage(a.asset_id):
usage_missing.append(a.asset_id)
planned = plan.cum(period)
amount = planned - posted[(a.asset_id, code)]
if amount == 0:
continue
acc = repo.account_map(a.root["class_code"], code, today)
if acc is None or acc["depreciation_expense_account_id"] is None or \
acc["accumulated_depreciation_account_id"] is None:
no_accounts.add((a.root["class_code"], code))
lines.append({"asset_id": a.asset_id, "area_code": code, "class_code": a.root["class_code"],
"cost_centre": m.get("cost_centre"), "planned_to_date": planned,
"posted_before": posted[(a.asset_id, code)], "amount": amount})
return lines, usage_missing, no_accounts
def _totals(lines: list) -> dict:
out: dict = defaultdict(lambda: ZERO)
for ln in lines:
out[ln["area_code"]] += ln["amount"]
return {k: str(v) for k, v in sorted(out.items())}
def _signature(lines: list) -> list:
return sorted((ln["asset_id"], ln["area_code"], str(ln["amount"])) for ln in lines)
def _load_run(repo, run_id: str) -> dict:
company, period, no = _split_run_id(run_id)
got = repo.load(T.DEP_RUN, company_code=company, period_id=period, run_no=no)
if got is None:
raise Refused(404, f"no depreciation run {run_id}")
root = got["fa_dep_run"][0]
return {"root": root, "lines": got["fa_dep_run_line"], "version": root["version"]}
def _save_run(repo, run: dict) -> int:
return T.DEP_RUN.save(repo.conn, {"fa_dep_run": [run["root"]], "fa_dep_run_line": run["lines"]},
expected_version=run["version"],
snapshot=repo.spend(T.DEP_RUN, run["root"]["company_code"], run["root"]["period_id"], run["root"]["run_no"]))
def _run_view(repo, run: dict, detail: bool = False) -> dict:
r = run["root"]
groups: dict = defaultdict(lambda: ZERO)
for ln in run["lines"]:
groups[(ln["area_code"], ln["class_code"], ln["cost_centre"])] += ln["amount"] or ZERO
out = {"run_id": r["run_id"], "company_code": r["company_code"], "period_id": r["period_id"], "run_no": r["run_no"],
"status": r["status"], "started_by_email": r["started_by_email"], "approved_by_email": r["approved_by_email"],
"approved_on": r["approved_on"].isoformat() if r["approved_on"] else None,
"posted_on": r["posted_on"].isoformat() if r["posted_on"] else None, "posted_by": r["posted_by"],
"reverses_run_id": r["reverses_run_id"], "reason": r["reason"], "totals": json.loads(r["totals"] or "{}"),
"gates": json.loads(r["gates"] or "[]"), "version": run["version"],
"groups": [{"area_code": a, "class_code": c, "cost_centre": cc, "amount": str(v)}
for (a, c, cc), v in sorted(groups.items(), key=lambda kv: tuple(str(x) for x in kv[0]))],
"assets": len({ln["asset_id"] for ln in run["lines"]})}
if detail:
out["lines"] = [{k: str(v) if isinstance(v, Decimal) else v for k, v in ln.items()
if k not in ("company_code", "period_id", "run_no", "created_at", "updated_at")}
for ln in sorted(run["lines"], key=lambda x: (x["area_code"], x["asset_id"]))]
return out
def _fill(run: dict, lines: list, gates: list) -> None:
r = run["root"]
run["lines"] = [{"company_code": r["company_code"], "period_id": r["period_id"], "run_no": r["run_no"], **ln}
for ln in lines]
r["totals"] = json.dumps(_totals(lines))
r["gates"] = json.dumps([g.as_dict() for g in gates])
def _runs_of(repo, company: str, period: str) -> list:
from sqlalchemy import select
rows = repo.conn.execute(select(T.fa_dep_run).where((T.fa_dep_run.c.company_code == company) &
(T.fa_dep_run.c.period_id == period)).order_by(T.fa_dep_run.c.run_no))
return [dict(r._mapping) for r in rows]
@FA.use_case(kind="command", route="POST /depreciation-runs", permission=hub.SIMULATE)
def simulate_depreciation(caller: dict, company_code: Code, period_id: Period) -> dict:
"""Simulate a period's depreciation for a company (FA-12): per area, class and cost centre, each asset's planned-to-date
minus already posted, with the gates of the run (usage_missing, account_determination, period_open per ledger).
NOTHING is posted. A second simulation of the same period returns the same figures until a row changes (it refreshes
the open run). A period after the current one is refused: no depreciation in advance."""
def unit():
with hub.repository() as repo:
today = hub.today(repo, True)
if period_id > period_of(today):
raise RuleBroken(f"{period_id} is after the current period {period_of(today)} (not_in_advance): no "
"depreciation in advance — run it in its own month")
services.check_year_open(repo, company_code, period_end(period_id))
runs = _runs_of(repo, company_code, period_id)
open_run = next((r for r in runs if r["status"] in ("simulated", "approved")), None)
lines, gates = compute_lines(repo, company_code, period_id)
if open_run:
run = _load_run(repo, open_run["run_id"])
if run["root"]["status"] == "approved":
return _run_view(repo, run)
else:
no = repo.first_free(r["run_no"] for r in runs)
repo.claim_slot("RUN", _run_id(company_code, period_id, no))
run = {"root": {"company_code": company_code, "period_id": period_id, "run_no": no,
"run_id": _run_id(company_code, period_id, no), "status": "simulated",
"started_by_email": services.email_of(caller), "started_by_sub": services.sub_of(caller),
"approved_by_email": None, "approved_by_sub": None, "approved_on": None, "posted_on": None,
"posted_by": None, "reverses_run_id": None, "reversed_by_email": None, "reason": None,
"totals": "{}", "gates": "[]"}, "lines": [], "version": 0}
_fill(run, lines, gates)
run["version"] = _save_run(repo, run)
return _run_view(repo, run)
with hub.decide():
return hub.retrying(unit)
@FA.use_case(kind="command", route="POST /depreciation-runs/{run_id}/resimulate", permission=hub.SIMULATE, idempotent=True)
def resimulate_depreciation(caller: dict, run_id: Code) -> dict:
"""Simulate a run again after a change (a late invoice, a corrected life): the figures are recomputed from the rows.
Only a run still in state simulated."""
with hub.repository() as repo, hub.decide():
run = _load_run(repo, run_id)
if run["root"]["status"] != "simulated":
raise RuleBroken(f"run {run_id} is {run['root']['status']}: only a simulated run is simulated again")
company, period, _ = _split_run_id(run_id)
lines, gates = compute_lines(repo, company, period)
_fill(run, lines, gates)
run["version"] = _save_run(repo, run)
return _run_view(repo, run)
@FA.use_case(kind="query", route="GET /depreciation-runs/{run_id}", permission=hub.REPORT)
def show_depreciation_run(caller: dict, run_id: Code, detail: bool = False) -> dict:
"""A run: its state, totals per area, the simulation grouped by area, class and cost centre, its gates, who started and
who approved it. `detail` adds every asset's line."""
with hub.repository() as repo:
return _run_view(repo, _load_run(repo, run_id), detail)
@FA.use_case(kind="query", route="GET /depreciation-runs", permission=hub.REPORT)
def list_depreciation_runs(caller: dict, company_code: Optional[Code] = None, period_id: Optional[Period] = None,
limit: Limit = 50) -> dict:
"""Depreciation runs, newest period first."""
with hub.repository() as repo:
from sqlalchemy import select
q = select(T.fa_dep_run).order_by(T.fa_dep_run.c.period_id.desc(), T.fa_dep_run.c.run_no.desc())
if company_code:
q = q.where(T.fa_dep_run.c.company_code == company_code)
if period_id:
q = q.where(T.fa_dep_run.c.period_id == period_id)
rows = [dict(r._mapping) for r in repo.conn.execute(q.limit(limit))]
return {"runs": [{"run_id": r["run_id"], "company_code": r["company_code"], "period_id": r["period_id"],
"status": r["status"], "totals": json.loads(r["totals"] or "{}")} for r in rows]}
def _gate_failures(gates: list) -> list:
return [g for g in gates if not g.met]
def _check_ready(gates: list, only: tuple = ("usage_missing", "account_determination")) -> None:
bad = [g for g in gates if not g.met and (g.gate in only)]
if bad:
raise RuleBroken("; ".join(f"{g.gate}: {g.reason} — {g.fix}" for g in bad))
def _period_gate_refusal(gates: list) -> None:
bad = [g for g in gates if not g.met and g.gate.startswith("period_open")]
if bad:
raise RuleBroken("a posting area's ledger period is closed (period_open): " + "; ".join(g.reason for g in bad)
+ " — reopen the period in gl, or post in the next run")
def _recheck(repo, run: dict) -> tuple:
"""Fresh lines and gates; the run must still say the same thing it said when it was simulated."""
company, period, _ = _split_run_id(run["root"]["run_id"])
lines, gates = compute_lines(repo, company, period)
if _signature(lines) != _signature([{k: ln[k] for k in ("asset_id", "area_code", "amount")} for ln in run["lines"]]):
raise RuleBroken("the figures changed since the simulation (a row changed): simulate the run again "
"(resimulate_depreciation) and approve what you see")
return lines, gates
@FA.use_case(kind="command", route="POST /depreciation-runs/{run_id}/approve", permission=hub.APPROVE_RUN, idempotent=True)
def approve_depreciation_run(caller: dict, run_id: Code) -> dict:
"""Approve a simulated run (FA-13). The approver is recorded (email, sign-in and today's date) and must not be who
started the run: the same email under another sign-in, or the same sign-in under another email, is the same person
(second_person). A usage reading missing, a class with no account in force, or a posting area whose ledger period is
closed (named) refuses it. On approval the system posts one journal per area to that area's ledger; approving again
changes nothing."""
actor = services.who(caller)
with hub.repository() as repo, hub.decide():
run = _load_run(repo, run_id)
r = run["root"]
if r["status"] in ("approved", "posted", "reversed"):
if r["status"] == "approved":
pass # resume the posting below
else:
return _run_view(repo, run)
else:
if same_person(services.email_of(caller), services.sub_of(caller), r["started_by_email"], r["started_by_sub"]):
raise RuleBroken("you started this run (second_person: same email or same sign-in): another approver "
"approves it")
_, gates = _recheck(repo, run)
_check_ready(gates)
_period_gate_refusal(gates)
r["status"] = "approved"
r["approved_by_email"], r["approved_by_sub"] = services.email_of(caller), services.sub_of(caller)
r["approved_on"] = hub.today(repo, True)
run["version"] = _save_run(repo, run)
return _post_run(run_id, actor, poster=None)
@FA.use_case(kind="command", route="POST /depreciation-runs/{run_id}/post", permission=hub.RUN, idempotent=True)
def post_unapproved_run(caller: dict, run_id: Code) -> dict:
"""Post a simulated run WITHOUT a second person — only where the owner has switched depreciation_needs_approval off
(warned: a wrong life or key is then caught by nobody else). The poster is recorded. As approval: usage readings,
account determination and open ledger periods are still required."""
actor = services.who(caller)
with hub.repository() as repo, hub.decide():
run = _load_run(repo, run_id)
r = run["root"]
if r["status"] in ("posted", "reversed"):
return _run_view(repo, run)
if repo.setting("depreciation_needs_approval", hub.today(repo)):
raise RuleBroken("the owner requires a second person to approve a depreciation run "
"(depreciation_needs_approval): ask another approver")
if r["status"] == "simulated":
_, gates = _recheck(repo, run)
_check_ready(gates)
_period_gate_refusal(gates)
r["status"] = "approved"
r["approved_on"] = hub.today(repo, True)
run["version"] = _save_run(repo, run)
return _post_run(run_id, actor, poster=actor)
@FA.use_case(kind="command", route="POST /depreciation-runs/{run_id}/post-journals", permission=hub.JOURNAL_POST,
idempotent=True)
def post_depreciation_run(caller: dict, run_id: Code) -> dict:
"""The system posts an APPROVED run: one journal per area to that area's ledger, keyed from the run and the ledger so a
retry replays instead of posting twice. Used to finish a posting that gl could not take the first time."""
return _post_run(run_id, services.who(caller), poster=None)
def _post_run(run_id: str, actor: str, poster: Optional[str]) -> dict:
"""Claim, record, send, finish — see the module docstring. Safe to repeat at every step."""
def phase_a():
with hub.repository() as repo:
from erp_platform.repository import upsert
from sqlalchemy import func, insert, select
run = _load_run(repo, run_id)
r = run["root"]
if r["status"] == "posted" and not _unsent(repo, run_id):
return "done"
if r["status"] not in ("approved", "posted"):
raise RuleBroken(f"run {run_id} is {r['status']}: only an approved run is posted")
have = repo.conn.execute(select(func.count()).select_from(T.fa_dep_posting).where(
T.fa_dep_posting.c.run_id == run_id)).scalar()
if have:
return "send"
today = hub.today(repo, True)
lines, gates = _recheck_for_post(repo, run)
_check_ready(gates)
_period_gate_refusal(gates)
company, period, _ = _split_run_id(run_id)
for area in sorted({ln["area_code"] for ln in lines}):
seq = (repo.conn.execute(select(func.max(T.fa_dep_post_claim.c.post_seq)).where(
(T.fa_dep_post_claim.c.company_code == company) & (T.fa_dep_post_claim.c.period_id == period) &
(T.fa_dep_post_claim.c.area_code == area))).scalar() or 0) + 1
repo.conn.execute(insert(T.fa_dep_post_claim).values(
company_code=company, period_id=period, area_code=area, post_seq=seq, run_id=run_id,
created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc)))
rows = [{"asset_id": ln["asset_id"], "area_code": ln["area_code"], "period_id": period, "run_id": run_id,
"amount": ln["amount"], "cost_centre": ln["cost_centre"], "class_code": ln["class_code"],
"journal_id": None, "kind": "run", "posted_on": today, "company_code": company} for ln in lines]
if rows:
upsert(repo.conn, T.fa_dep_posting, rows, ("asset_id", "area_code", "period_id", "run_id"))
run["lines"] = [{"company_code": company, "period_id": period, "run_no": r["run_no"], **ln} for ln in lines]
r["totals"] = json.dumps(_totals(lines))
if poster:
r["posted_by"] = poster
run["version"] = _save_run(repo, run)
return "send" if rows else "finish"
with hub.decide():
step = hub.retrying(phase_a)
if step == "send":
_send_run(run_id, actor)
with hub.repository() as repo:
run = _load_run(repo, run_id)
r = run["root"]
if r["status"] != "posted":
r["status"] = "posted"
r["posted_on"] = hub.today(repo, True)
run["version"] = _save_run(repo, run)
return _run_view(repo, run)
def _recheck_for_post(repo, run: dict) -> tuple:
"""At posting the difference is recomputed from COMMITTED postings: a concurrent run that posted first leaves this one
nothing (zero), never a double charge."""
company, period, _ = _split_run_id(run["root"]["run_id"])
return compute_lines(repo, company, period)
def _unsent(repo, run_id: str) -> int:
from sqlalchemy import func, select
return repo.conn.execute(select(func.count()).select_from(T.fa_dep_posting).where(
(T.fa_dep_posting.c.run_id == run_id) & (T.fa_dep_posting.c.journal_id.is_(None)) &
(T.fa_dep_posting.c.amount != 0))).scalar()
def _run_journals(repo, run_id: str, reverse: bool = False) -> list:
"""[(area, suffix, body, [posting keys])] from the run's postings: one journal per area, or per class past 5,000 lines."""
company, period, _ = _split_run_id(run_id)
today = repo.today()
areas = {a["area_code"]: a for a in repo.posting_areas(today)}
by_area: dict = defaultdict(list)
for p in repo.postings(company_code=company):
if p["run_id"] == run_id and p["amount"] and p["journal_id"] is None:
by_area[p["area_code"]].append(p)
out = []
jday = today if reverse else min(period_end(period), today)
for code, rows in sorted(by_area.items()):
area = areas.get(code)
if area is None:
raise RuleBroken(f"area {code} no longer posts to a ledger")
per_class: dict = defaultdict(list)
for p in rows:
acc = repo.account_map(p["class_code"], code, today)
L: list = []
journals.put(L, p["amount"], {"account_id": acc["depreciation_expense_account_id"]},
{"account_id": acc["accumulated_depreciation_account_id"]}, asset_id=p["asset_id"],
cost_centre=p["cost_centre"], movement="depreciation", memo=f"depreciation {period}")
per_class[p["class_code"]] += L
merged = {c: journals.merge(ls) for c, ls in per_class.items()}
for suffix, lines in journals.split_by_class(merged):
lines = journals.merge(lines) if suffix == "" else lines
journals.check_balanced(lines)
doc = f"fa:{run_id}{suffix}"
body = services.journal_body(doc, "reverse" if reverse else "depreciate", company, area["ledger_id"], jday,
area["currency"] or "GBP", f"depreciation {period}", lines)
keys = [p for p in rows if not suffix or p["class_code"] == suffix[1:]]
out.append((code, suffix, body, keys))
return out
def _send_run(run_id: str, actor: str, reverse: bool = False) -> None:
with hub.repository() as repo:
planned = _run_journals(repo, run_id, reverse)
sent = []
for code, suffix, body, keys in planned:
r = seams.post_journal(body, actor)
if not r.ok:
raise RuleBroken(f"gl refused the journal for area {code} ({r.gate or r.status}): {r.reason}"
+ (f" — {r.fix}" if r.fix else ""))
sent.append((keys, r.journal_id))
with hub.repository() as repo:
from erp_platform.repository import upsert
for keys, jid in sent:
upsert(repo.conn, T.fa_dep_posting,
[{"asset_id": k["asset_id"], "area_code": k["area_code"], "period_id": k["period_id"],
"run_id": k["run_id"], "journal_id": jid} for k in keys],
("asset_id", "area_code", "period_id", "run_id"))
@FA.use_case(kind="command", route="POST /depreciation-runs/{run_id}/reverse", permission=hub.APPROVE_RUN, destructive=True)
def reverse_depreciation_run(caller: dict, run_id: Code, reason: Reason) -> dict:
"""Reverse a posted run (FA-14): a reversing run with its own journals dated today in an open period, the posted amounts
per asset set back by reversing rows, and the original journals untouched. The next run plans the period again.
Refused for the starter of the run (second_person: it is an approval), once the asset year is closed, or when the
period of today is closed in a posting area's ledger."""
actor = services.who(caller)
def phase_a():
with hub.repository() as repo:
from erp_platform.repository import upsert
from sqlalchemy import func, insert, select
run = _load_run(repo, run_id)
r = run["root"]
existing = next((x for x in _runs_of(repo, r["company_code"], r["period_id"])
if x["reverses_run_id"] == run_id), None)
if existing:
return existing["run_id"]
if r["status"] != "posted":
raise RuleBroken(f"run {run_id} is {r['status']}: only a posted run is reversed")
if same_person(services.email_of(caller), services.sub_of(caller), r["started_by_email"], r["started_by_sub"]):
raise RuleBroken("you started this run (second_person: same email or same sign-in): another approver "
"reverses it")
today = hub.today(repo, True)
fy = fiscal_year(r["period_id"], services._fy_end(repo, r["company_code"]))
if repo.year_closed(r["company_code"], fy):
raise RuleBroken(f"the asset year {fy} is closed (year_open): correct it in the next year's run")
services.check_year_open(repo, r["company_code"], today)
gates = services.period_gates(today, r["company_code"], repo.posting_areas(today, r["company_code"]))
_period_gate_refusal(gates)
runs = _runs_of(repo, r["company_code"], r["period_id"])
no = repo.first_free(x["run_no"] for x in runs)
new_id = _run_id(r["company_code"], r["period_id"], no)
repo.claim_slot("RUN", new_id)
orig = [p for p in repo.postings(company_code=r["company_code"]) if p["run_id"] == run_id]
root = {"company_code": r["company_code"], "period_id": r["period_id"], "run_no": no, "run_id": new_id,
"status": "posted", "started_by_email": services.email_of(caller), "started_by_sub": services.sub_of(caller),
"approved_by_email": None, "approved_by_sub": None, "approved_on": None, "posted_on": today,
"posted_by": actor, "reverses_run_id": run_id, "reversed_by_email": None, "reason": reason,
"totals": "{}", "gates": "[]"}
for area in sorted({p["area_code"] for p in orig}):
seq = (repo.conn.execute(select(func.max(T.fa_dep_post_claim.c.post_seq)).where(
(T.fa_dep_post_claim.c.company_code == r["company_code"]) &
(T.fa_dep_post_claim.c.period_id == r["period_id"]) & (T.fa_dep_post_claim.c.area_code == area))).scalar()
or 0) + 1
repo.conn.execute(insert(T.fa_dep_post_claim).values(
company_code=r["company_code"], period_id=r["period_id"], area_code=area, post_seq=seq, run_id=new_id,
created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc)))
rows = [{"asset_id": p["asset_id"], "area_code": p["area_code"], "period_id": p["period_id"], "run_id": new_id,
"amount": -p["amount"], "cost_centre": p["cost_centre"], "class_code": p["class_code"],
"journal_id": None, "kind": "reversal", "posted_on": today, "company_code": p["company_code"]}
for p in orig]
if rows:
upsert(repo.conn, T.fa_dep_posting, rows, ("asset_id", "area_code", "period_id", "run_id"))
lines = [{"company_code": root["company_code"], "period_id": root["period_id"], "run_no": no,
"asset_id": p["asset_id"], "area_code": p["area_code"], "class_code": p["class_code"],
"cost_centre": p["cost_centre"], "planned_to_date": ZERO, "posted_before": p["amount"],
"amount": -p["amount"]} for p in orig]
root["totals"] = json.dumps(_totals([{"area_code": ln["area_code"], "amount": ln["amount"]} for ln in lines]))
T.DEP_RUN.save(repo.conn, {"fa_dep_run": [root], "fa_dep_run_line": lines}, expected_version=0)
return new_id
with hub.decide():
new_id = hub.retrying(phase_a)
_send_run(new_id, actor, reverse=True)
with hub.repository() as repo:
orig = _load_run(repo, run_id)
if orig["root"]["status"] != "reversed":
orig["root"]["status"] = "reversed"
orig["root"]["reversed_by_email"] = services.email_of(caller)
orig["root"]["reason"] = reason
orig["version"] = _save_run(repo, orig)
return _run_view(repo, _load_run(repo, new_id))
@FA.use_case(kind="command", route="POST /years/{company_code}/{fiscal_year}/close", permission=hub.YEAR_CLOSE,
idempotent=True)
def close_asset_year(caller: dict, company_code: Code, fiscal_year: Year) -> dict:
"""Close the asset year after its last run (FA-15): an asset transaction dated in it is refused afterwards, naming the
year, and the next year's opening values are the closing values (derived, nothing copied). Refused while a period of the
year has no posted run, or a transaction awaits approval — each named. A closed year is never reopened here."""
with hub.repository() as repo, hub.decide():
from erp_platform.repository import upsert
if repo.year_closed(company_code, fiscal_year):
return {"company_code": company_code, "fiscal_year": fiscal_year, "status": "closed"}
fy_end = services._fy_end(repo, company_code)
today = hub.today(repo, True)
first, last = fy_first_period(fiscal_year, fy_end), fy_last_period(fiscal_year, fy_end)
problems = []
for i in range(period_index(first), period_index(last) + 1):
p = period_add("0000-01", i)
runs = _runs_of(repo, company_code, p)
if not any(x["status"] == "posted" and not x["reverses_run_id"] for x in runs):
problems.append(f"period {p} has no posted depreciation run")
for t in repo.tx_rows(statuses=("proposed", "approved"), company_code=company_code):
if t["fiscal_year"] == fiscal_year:
problems.append(f"transaction {t['transaction_id']} ({t['kind']}) awaits approval or posting")
if problems:
raise RuleBroken("the year cannot close: " + "; ".join(problems))
upsert(repo.conn, T.fa_year, [{"company_code": company_code, "fiscal_year": fiscal_year, "status": "closed",
"closed_by_email": services.email_of(caller), "closed_by_sub": services.sub_of(caller),
"closed_on": today}], ("company_code", "fiscal_year"))
return {"company_code": company_code, "fiscal_year": fiscal_year, "status": "closed", "closed_on": today.isoformat()}
@FA.use_case(kind="query", route="GET /depreciation/status", permission=hub.DEP_READ, tool=False)
def get_depreciation_status(caller: dict, company_code: Code, period_id: Period, ledger_id: Optional[Code] = None) -> dict:
"""What gl's period close asks (its gate depreciation_posted): for each area posting to a ledger, whether its depreciation
for the period is posted (and by which run), how many asset transactions still await approval or posting, and the
readiness list. A period closes in a ledger only when every area posting to it has posted."""
with hub.repository() as repo:
today = hub.today(repo)
areas = [a for a in repo.posting_areas(today) if ledger_id is None or str(a["ledger_id"]) == str(ledger_id)]
runs = _runs_of(repo, company_code, period_id)
posted_runs = [x for x in runs if x["status"] == "posted" and not x["reverses_run_id"]]
out, gates = [], []
for a in areas:
run = posted_runs[-1] if posted_runs else None
ok = run is not None
out.append({"area_code": a["area_code"], "ledger_id": a["ledger_id"], "posted": ok,
"run_id": run["run_id"] if run else None})
gates.append(met(f"depreciation_posted:{a['area_code']}") if ok else
unmet(f"depreciation_posted:{a['area_code']}", f"area {a['area_code']} has no posted depreciation "
f"for {period_id}", "simulate, approve and post the run"))
pending = [t for t in repo.tx_rows(statuses=("proposed", "approved"), company_code=company_code)
if period_of(t["value_date"]) == period_id]
gates.append(met("transactions_posted") if not pending else
unmet("transactions_posted", f"{len(pending)} asset transaction(s) await approval or posting",
"approve and post them"))
return {"company_code": company_code, "period_id": period_id, "areas": out, "pending_transactions": len(pending),
"readiness": [g.as_dict() for g in gates]}