dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/uc_reports.py
"""Seeing it and checking it (FA-23 .. FA-26): the asset history sheet, the register reconciled to gl per class account and
ledger, what gl asks (GET /assets/balances), and the physical count of assets.
Every figure of the history sheet is a SUM of transactions and postings in rows, so the closing of one year equals the opening
of the next; the reconciliation compares the register's figure with gl's per class account to the penny."""
import csv
import io
import json
from collections import defaultdict
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import Limit, Name, Note
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,
CountResultIn,
In,
IsoDate,
Money,
Period,
Seq,
Year,
day,
)
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import (
RESULTS,
ZERO,
fy_first_period,
fy_last_period,
ids_of,
period_end,
period_start,
)
from brick_fixed_assets.infrastructure import tables as T
FA = hub.FA
# ── the asset history sheet (FA-23) ───────────────────────────────────────────────────────────
COLUMNS = ("cost_start", "additions", "retirements", "transfers_in", "transfers_out", "revaluations", "cost_end",
"depreciation_start", "charge", "impairments", "impairment_reversals", "depreciation_on_retirements",
"depreciation_on_transfers", "depreciation_on_revaluations", "depreciation_end", "nbv_start", "nbv_end")
def history(repo, company_code: str, area_code: str, start: date, end: date) -> list:
"""[{class_code, ...COLUMNS}] — each column the SUM of transactions (and postings) in rows."""
classes: dict = {}
def row(class_code: str) -> dict:
return classes.setdefault(class_code, {c: ZERO for c in COLUMNS})
klass_of = {r["asset_id"]: r["class_code"] for r in repo.asset_roots(company_code)}
for t in repo.area_values_rows(company_code=company_code, area_code=area_code):
cls = klass_of.get(t["asset_id"])
if cls is None:
continue
r = row(cls)
cost, acc, vd, kind = t["cost"] or ZERO, t["accumulated_depreciation"] or ZERO, t["value_date"], t["kind"]
if kind == "takeover":
r["cost_start"] += cost # carried in: the opening of the year it arrives in, and of every later one
diy = t["depreciation_in_year"] or ZERO # the part of `acc` charged in the year of the cutover
if vd < start:
r["depreciation_start"] += acc
elif vd <= end:
r["depreciation_start"] += acc - diy
r["charge"] += diy
else:
r["cost_start"] -= cost
continue
if vd < start:
r["cost_start"] += cost
r["depreciation_start"] += acc
elif vd <= end:
if kind in ("acquisition", "grant", "settlement"):
r["additions"] += cost
elif kind == "retirement":
r["retirements"] += cost
r["depreciation_on_retirements"] += acc
elif kind == "transfer":
if cost > 0:
r["transfers_in"] += cost
else:
r["transfers_out"] += cost
r["depreciation_on_transfers"] += acc
elif kind == "revaluation":
r["revaluations"] += cost
r["depreciation_on_revaluations"] += acc
elif kind == "impairment":
r["impairments"] += acc
elif kind == "impairment_reversal":
r["impairment_reversals"] += acc
elif kind == "reversal":
prm = json.loads(t["params"] or "{}")
of = prm.get("of_kind")
if of in ("acquisition", "grant", "settlement"):
r["additions"] += cost
elif of == "retirement":
r["retirements"] += cost
r["depreciation_on_retirements"] += acc
elif of == "transfer":
if cost > 0:
r["transfers_in"] += cost
else:
r["transfers_out"] += cost
r["depreciation_on_transfers"] += acc
elif of == "revaluation":
r["revaluations"] += cost
r["depreciation_on_revaluations"] += acc
elif of == "impairment":
r["impairments"] += acc
elif of == "impairment_reversal":
r["impairment_reversals"] += acc
for p in repo.postings(area_code=area_code, company_code=company_code):
cls = klass_of.get(p["asset_id"])
if cls is None:
continue
r = row(cls)
pend = period_end(p["period_id"])
if pend < start:
r["depreciation_start"] += p["amount"] or ZERO
elif pend <= end:
r["charge"] += p["amount"] or ZERO
out = []
for cls in sorted(classes):
r = classes[cls]
r["cost_end"] = (r["cost_start"] + r["additions"] + r["retirements"] + r["transfers_in"] + r["transfers_out"]
+ r["revaluations"])
r["depreciation_end"] = (r["depreciation_start"] + r["charge"] + r["impairments"] + r["impairment_reversals"]
+ r["depreciation_on_retirements"] + r["depreciation_on_transfers"]
+ r["depreciation_on_revaluations"])
r["nbv_start"] = r["cost_start"] - r["depreciation_start"]
r["nbv_end"] = r["cost_end"] - r["depreciation_end"]
out.append({"class_code": cls, **r})
return out
@FA.use_case(kind="query", route="GET /reports/history", permission=hub.REPORT)
def show_asset_history(caller: dict, company_code: Code, area_code: Code, fiscal_year: Optional[Year] = None,
from_period: Optional[Period] = None, to_period: Optional[Period] = None,
format: Code = "json") -> dict:
"""The asset history sheet (the fixed assets note, IAS 16.73(e)): per class and area, cost at the start of the year, additions,
retirements, transfers in and out, revaluations, cost at the end; accumulated depreciation at the start, the charge for the
year, impairments and reversals (separately), depreciation on retirements, transfers and revaluations, at the end; net
book value at the start and the end. Every column is the SUM of transactions in rows, so one year's closing is the next
year's opening. Give a fiscal_year or a period range; format csv returns it as CSV."""
with hub.repository() as repo:
if fiscal_year is not None:
fy_end = services._fy_end(repo, company_code)
first, last = fy_first_period(fiscal_year, fy_end), fy_last_period(fiscal_year, fy_end)
elif from_period and to_period:
first, last = from_period, to_period
else:
raise Refused(422, "give a fiscal_year, or from_period and to_period")
rows = history(repo, company_code, area_code, period_start(first), period_end(last))
totals = {c: str(sum((r[c] for r in rows), ZERO)) for c in COLUMNS}
shown = [{**{k: (str(v) if isinstance(v, Decimal) else v) for k, v in r.items()}} for r in rows]
out = {"company_code": company_code, "area_code": area_code, "from_period": first, "to_period": last,
"columns": list(COLUMNS), "classes": shown, "totals": totals}
if format == "csv":
buf = io.StringIO()
w = csv.writer(buf)
w.writerow(["class_code", *COLUMNS])
for r in shown:
w.writerow([r["class_code"], *[r[c] for c in COLUMNS]])
w.writerow(["TOTAL", *[totals[c] for c in COLUMNS]])
out["csv"] = buf.getvalue()
return out
# ── the register against gl (FA-24) ───────────────────────────────────────────────────────────
def register_accounts(repo, company_code: str, ledger_id, as_of: date) -> list:
"""[{account_id, role (cost | accumulated_depreciation), class_code, balance}] — the register's figure per class account in
the area that posts to the ledger, as of a date."""
today = repo.today()
area = next((a for a in repo.posting_areas(today) if ledger_id is None or str(a["ledger_id"]) == str(ledger_id)), None)
if area is None:
raise RuleBroken(f"no area posts to ledger {ledger_id}")
code = area["area_code"]
klass_of = {r["asset_id"]: r["class_code"] for r in repo.asset_roots(company_code)}
cost: dict = defaultdict(lambda: ZERO)
acc: dict = defaultdict(lambda: ZERO)
for t in repo.area_values_rows(company_code=company_code, area_code=code):
if t["asset_id"] in klass_of and t["value_date"] <= as_of:
cost[klass_of[t["asset_id"]]] += t["cost"] or ZERO
acc[klass_of[t["asset_id"]]] += t["accumulated_depreciation"] or ZERO
for p in repo.postings(area_code=code, company_code=company_code):
if p["asset_id"] in klass_of and period_end(p["period_id"]) <= as_of:
acc[klass_of[p["asset_id"]]] += p["amount"] or ZERO
out: dict = {}
for cls in sorted(set(cost) | set(acc)):
m = repo.account_map(cls, code, today)
if m is None:
continue
for role, field, val in (("cost", "cost_account_id", cost[cls]),
("accumulated_depreciation", "accumulated_depreciation_account_id", acc[cls])):
key = (m[field], role)
slot = out.setdefault(key, {"account_id": m[field], "role": role, "class_code": cls, "balance": ZERO,
"classes": []})
slot["balance"] += val
slot["classes"].append(cls)
rows = []
for (aid, role), s in sorted(out.items(), key=lambda kv: (str(kv[0][0]), kv[0][1])):
rows.append({"account_id": aid, "role": role, "class_code": ",".join(sorted(set(s["classes"]))),
"balance": s["balance"]})
return rows
def open_acquisitions(repo, company_code: str) -> list:
from sqlalchemy import select
out = []
for c in repo.conn.execute(select(T.fa_acquisition_claim)):
t = repo.transaction(c.transaction_id)
if t and t["root"]["company_code"] == company_code and t["root"]["status"] != "posted":
out.append({"acquisition_ref": c.acquisition_ref, "amount": str(c.amount)})
return out
@FA.use_case(kind="query", route="GET /assets/balances", permission=hub.READ)
def get_asset_balances(caller: dict, company_code: Code, as_of: IsoDate, ledger_id: Optional[Code] = None) -> dict:
"""What gl reconciles each asset account against (the way it reconciles payables against purchasing): for a company and
ledger, the register's balance per class account (cost and accumulated depreciation) as of a date, and the acquisitions
handed over but not yet journalled (asset clearing: zero when every asset bill was handed over)."""
with hub.repository() as repo, hub.decide():
when = day(as_of)
accounts = register_accounts(repo, company_code, ledger_id, when)
return {"company_code": company_code, "ledger_id": ledger_id, "as_of": as_of,
"accounts": [{"account_id": a["account_id"], "role": a["role"], "class_code": a["class_code"],
"balance": str(a["balance"])} for a in accounts],
"clearing": {"account_purpose": "asset_clearing", "open_acquisitions": open_acquisitions(repo, company_code)}}
class GlFigure(In):
account_id: Code
balance: Money
@FA.use_case(kind="query", route="GET /reports/reconciliation", permission=hub.REPORT)
def show_asset_reconciliation(caller: dict, company_code: Code, as_of: IsoDate, ledger_id: Optional[Code] = None) -> dict:
"""The register against the ledger per class account (FA-24): the register's figure for each account, and the open
hand-overs behind asset clearing. gl compares its own balance with these (get_asset_balances) and names a break: a
difference is 'posted to an asset account outside the register: reverse it and post a manual acquisition'. Asset clearing is
zero when every asset bill was handed over."""
with hub.repository() as repo, hub.decide():
when = day(as_of)
accounts = register_accounts(repo, company_code, ledger_id, when)
opens = open_acquisitions(repo, company_code)
return {"company_code": company_code, "ledger_id": ledger_id, "as_of": as_of,
"accounts": [{"account_id": a["account_id"], "role": a["role"], "class_code": a["class_code"],
"register": str(a["balance"]),
"fix_if_gl_differs": "posted to an asset account outside the register: reverse it and post a "
"manual acquisition"} for a in accounts],
"clearing": {"account_purpose": "asset_clearing", "open_acquisitions": opens,
"balance": str(sum((Decimal(o["amount"]) for o in opens), ZERO))}}
@FA.use_case(kind="command", route="POST /reports/reconcile", permission=hub.REPORT, idempotent=True, read_only=True)
def compare_asset_balances(caller: dict, company_code: Code, as_of: IsoDate, gl_balances: list[GlFigure],
ledger_id: Optional[Code] = None) -> dict:
"""Compare gl's balances (give them per account_id) with the register's, per class account: the register's figure, gl's
figure, the difference and its fix — a difference is named 'posted to an asset account outside the register: reverse it and
post a manual acquisition'. Nothing is changed."""
with hub.repository() as repo, hub.decide():
when = day(as_of)
mine = register_accounts(repo, company_code, ledger_id, when)
gl = {str(g.account_id): g.balance for g in gl_balances}
rows, equal = [], True
for a in mine:
theirs = gl.get(str(a["account_id"]))
diff = None if theirs is None else theirs - a["balance"]
ok = diff == 0
equal &= ok
rows.append({"account_id": a["account_id"], "role": a["role"], "class_code": a["class_code"],
"register": str(a["balance"]), "gl": None if theirs is None else str(theirs),
"difference": None if diff is None else str(diff), "equal": ok,
"fix": "" if ok else ("gl has no figure for this account" if theirs is None else
"posted to an asset account outside the register: reverse it and post a manual "
"acquisition")})
return {"company_code": company_code, "as_of": as_of, "equal": equal, "accounts": rows}
# ── asset counts (FA-25, FA-26) ───────────────────────────────────────────────────────────────
def _load_count(repo, location: str, count_date: date) -> dict:
got = repo.load(T.COUNT, location_id=location, count_date=count_date)
if got is None:
raise Refused(404, f"no count for {location} on {count_date}")
root = got["fa_count"][0]
return {"root": root, "lines": sorted(got["fa_count_line"], key=lambda r: r["line_no"]), "version": root["version"]}
def _save_count(repo, c: dict) -> int:
return T.COUNT.save(repo.conn, {"fa_count": [c["root"]], "fa_count_line": c["lines"]}, expected_version=c["version"],
snapshot=repo.spend(T.COUNT, c["root"]["location_id"], c["root"]["count_date"]))
def _own_count(caller: dict, c: dict, manage: bool) -> None:
"""A custodian sees only their own location's counts: another's is not found."""
if manage:
return
mine = ids_of(caller.get("email"), caller.get("sub"))
if not mine or (c["root"]["custodian_email"] or "").strip().lower() not in mine:
raise Refused(404, f"no count for {c['root']['location_id']} on {c['root']['count_date']}")
def _count_view(c: dict) -> dict:
r = c["root"]
return {"location_id": r["location_id"], "count_date": r["count_date"].isoformat(), "status": r["status"],
"custodian_email": r["custodian_email"], "opened_by_email": r["opened_by_email"], "version": c["version"],
"lines": [{"line_no": ln["line_no"], "asset_id": ln["asset_id"], "tag": ln["tag"],
"description": ln["description"], "serial_no": ln["serial_no"], "result": ln["result"],
"counted_by_email": ln["counted_by_email"],
"counted_on": ln["counted_on"].isoformat() if ln["counted_on"] else None, "decision": ln["decision"],
"decision_ref": ln["decision_ref"], "note": ln["note"], "moved_in": bool(ln["moved_in"])}
for ln in c["lines"]],
"expected": len([ln for ln in c["lines"] if not ln["moved_in"] and ln["asset_id"]])}
@FA.use_case(kind="command", route="POST /asset-counts", permission=hub.COUNT_MANAGE, idempotent=True)
def open_asset_count(caller: dict, location_id: Code, count_date: IsoDate, custodian_email: Name) -> dict:
"""Open a count of the assets at a location (or in a cost centre) on a date, assigned to its custodian (FA-25). The count lists
the active assets the register has there (tag, description, serial number): that list is FIXED when the count opens. It is
keyed (location, count date): a second open returns it. The count date is a fact: today or earlier."""
def unit():
with hub.repository() as repo:
when = day(count_date)
today = hub.today(repo, True)
if when > today:
raise RuleBroken("a count cannot be dated after today: the count date is a fact")
got = T.COUNT.load(repo.conn, {"location_id": location_id, "count_date": when})
if got:
return _count_view(_load_count(repo, location_id, when))
masters = repo.masters_on(today)
lines = []
for root in repo.asset_roots(statuses=["active"]):
m = masters.get(root["asset_id"]) or {}
if location_id in ((m.get("location_id") or ""), (m.get("cost_centre") or "")):
lines.append({"location_id": location_id, "count_date": when, "line_no": len(lines) + 1,
"asset_id": root["asset_id"], "tag": root["inventory_no"],
"description": root["description"], "serial_no": root["serial_no"], "result": None,
"counted_by_email": None, "counted_on": None, "decision": None, "decision_ref": None,
"note": None, "moved_in": False})
root = {"location_id": location_id, "count_date": when, "custodian_email": custodian_email.strip().lower(),
"status": "open", "opened_by_email": services.who(caller), "version": 0}
c = {"root": root, "lines": lines, "version": 0}
c["version"] = _save_count(repo, c)
return _count_view(c)
with hub.decide():
return hub.retrying(unit)
@FA.use_case(kind="command", route="POST /asset-counts/{location_id}/{count_date}/results", permission=hub.COUNT_WRITE)
def count_asset(caller: dict, location_id: Code, count_date: IsoDate, results: list[CountResultIn]) -> dict:
"""Record what is found (FA-26): each line a result — found, not_found, damaged or found_unlisted (something on site that
is not listed; an asset of the register that was moved in during the count is listed separately) — dated the count date, with
who counted. Name a line by line_no, asset_id or the tag scanned (inventory_no). The count date is a fact: today or earlier.
A custodian sees and counts only their own location's count."""
with hub.repository() as repo, hub.decide():
when = day(count_date)
c = _load_count(repo, location_id, when)
_own_count(caller, c, hub.auth.has_permission(caller, hub.COUNT_MANAGE))
today = hub.today(repo, True)
if when > today:
raise RuleBroken("a count cannot be dated after today: the count date is a fact")
if c["root"]["status"] != "open":
raise RuleBroken(f"the count is {c['root']['status']}: results are recorded while it is open")
for res in results:
if res.result not in RESULTS:
raise RuleBroken(f"result is one of {', '.join(RESULTS)}")
line = None
for ln in c["lines"]:
if (res.line_no and ln["line_no"] == res.line_no) or (res.asset_id and ln["asset_id"] == res.asset_id) or \
(res.inventory_no and ln["tag"] == res.inventory_no):
line = ln
break
if res.result == "found_unlisted" or line is None:
if res.result != "found_unlisted":
raise RuleBroken("that is not on the count's list: record it as found_unlisted")
known = repo.asset(res.asset_id) if res.asset_id else None
if line is None:
line = {"location_id": location_id, "count_date": when, "line_no": len(c["lines"]) + 1,
"asset_id": known.asset_id if known else None, "tag": res.inventory_no,
"description": res.description or (known.root["description"] if known else None),
"serial_no": None, "result": None, "counted_by_email": None, "counted_on": None,
"decision": None, "decision_ref": None, "note": None, "moved_in": bool(known)}
c["lines"].append(line)
line["result"] = res.result
line["counted_by_email"] = services.who(caller)
line["counted_on"] = when
line["note"] = res.note or line["note"]
c["version"] = _save_count(repo, c)
return _count_view(c)
@FA.use_case(kind="command", route="POST /asset-counts/{location_id}/{count_date}/finish", permission=hub.COUNT_WRITE,
idempotent=True)
def finish_asset_count(caller: dict, location_id: Code, count_date: IsoDate) -> dict:
"""Hand the count back to the accountant: refused while a listed asset has no result (all_counted) — each one named; count
it, or record not_found."""
with hub.repository() as repo, hub.decide():
c = _load_count(repo, location_id, day(count_date))
_own_count(caller, c, hub.auth.has_permission(caller, hub.COUNT_MANAGE))
if c["root"]["status"] != "open":
return _count_view(c)
missing = [ln["asset_id"] or ln["tag"] or f"line {ln['line_no']}" for ln in c["lines"] if ln["result"] is None]
if missing:
raise RuleBroken("lines without a result (all_counted): " + ", ".join(missing) + " — count them, or record not_found")
c["root"]["status"] = "counted"
c["version"] = _save_count(repo, c)
return _count_view(c)
DECISIONS = ("retire", "transfer", "new_asset", "note")
@FA.use_case(kind="command", route="POST /asset-counts/{location_id}/{count_date}/decisions", permission=hub.COUNT_MANAGE)
def decide_count_difference(caller: dict, location_id: Code, count_date: IsoDate, line_no: Seq, decision: Code,
note: Optional[Note] = None, class_code: Optional[Code] = None,
company_code: Optional[Code] = None) -> dict:
"""Decide one difference of a counted count: not_found or damaged becomes a retirement PROPOSAL for the accountant to approve
(decision `retire`, scrapping at the count date), a transfer or a note; something found that is not listed becomes a DRAFT
asset (`new_asset`, needs class_code and company_code) or a note. Nothing is posted."""
when = day(count_date)
with hub.repository() as repo, hub.decide():
c = _load_count(repo, location_id, when)
if c["root"]["status"] != "counted":
raise RuleBroken(f"the count is {c['root']['status']}: differences are decided once the custodian hands it back")
ln = next((x for x in c["lines"] if x["line_no"] == line_no), None)
if ln is None:
raise Refused(404, f"no line {line_no}")
if ln["result"] not in ("not_found", "damaged", "found_unlisted"):
raise RuleBroken("only a difference (not_found, damaged, found_unlisted) is decided")
if decision not in DECISIONS:
raise RuleBroken(f"decision is one of {', '.join(DECISIONS)}")
if decision == "retire" and not ln["asset_id"]:
raise RuleBroken("there is no asset to retire on this line")
if decision == "new_asset" and not (class_code and company_code):
raise RuleBroken("a new asset needs class_code and company_code")
ref = None
if decision == "retire":
from brick_fixed_assets.application.uc_transactions import propose_retirement
ref = propose_retirement(caller, ln["asset_id"], count_date, percent=Decimal("100"), memo=note or "not found in count")
ref = ref["transaction_id"]
elif decision == "new_asset":
from brick_fixed_assets.application.uc_assets import create_asset
made = create_asset(caller, f"count:{location_id}:{count_date}:{line_no}", company_code, class_code,
ln["description"] or "found in count", inventory_no=ln["tag"], location_id=location_id)
ref = made["asset_id"]
with hub.repository() as repo, hub.decide():
c = _load_count(repo, location_id, when)
line = next(x for x in c["lines"] if x["line_no"] == line_no)
line["decision"], line["decision_ref"] = decision, ref
line["note"] = note or line["note"]
c["version"] = _save_count(repo, c)
return _count_view(c)
@FA.use_case(kind="command", route="POST /asset-counts/{location_id}/{count_date}/close", permission=hub.COUNT_MANAGE,
idempotent=True)
def close_asset_count(caller: dict, location_id: Code, count_date: IsoDate) -> dict:
"""Close the count: refused while a difference has no decision (differences_decided). The last-counted date is set on every
asset found (or damaged but there)."""
with hub.repository() as repo, hub.decide():
when = day(count_date)
c = _load_count(repo, location_id, when)
if c["root"]["status"] == "closed":
return _count_view(c)
if c["root"]["status"] != "counted":
raise RuleBroken(f"the count is {c['root']['status']}: the custodian hands it back first")
open_ = [ln["asset_id"] or ln["tag"] or f"line {ln['line_no']}" for ln in c["lines"]
if ln["result"] in ("not_found", "damaged", "found_unlisted") and not ln["decision"]]
if open_:
raise RuleBroken("a difference has no decision (differences_decided): " + ", ".join(open_))
counted = repo.assets(ln["asset_id"] for ln in c["lines"] if ln["result"] in ("found", "damaged", "found_unlisted") and ln["asset_id"])
for ln in c["lines"]:
if ln["result"] in ("found", "damaged", "found_unlisted") and ln["asset_id"]:
a = counted.get(ln["asset_id"])
if a is not None:
a.root["last_counted_on"] = when
repo.save_asset(a)
c["root"]["status"] = "closed"
c["version"] = _save_count(repo, c)
return _count_view(c)
@FA.use_case(kind="query", route="GET /asset-counts/{location_id}/{count_date}", permission=hub.COUNT_WRITE)
def show_asset_count(caller: dict, location_id: Code, count_date: IsoDate) -> dict:
"""One count with its lines (tag, description, serial number, result, decision). A custodian sees only their own
location's counts; another's is not found."""
with hub.repository() as repo:
c = _load_count(repo, location_id, day(count_date))
_own_count(caller, c, hub.auth.has_permission(caller, hub.COUNT_MANAGE))
return _count_view(c)
@FA.use_case(kind="query", route="GET /asset-counts", permission=hub.COUNT_WRITE)
def list_asset_counts(caller: dict, status: Optional[Code] = None, limit: Limit = 50) -> dict:
"""The counts you may see (a custodian: their own), newest first."""
with hub.repository() as repo:
from sqlalchemy import select
q = select(T.fa_count).order_by(T.fa_count.c.count_date.desc())
if status:
q = q.where(T.fa_count.c.status == status)
mine = ids_of(caller.get("email"), caller.get("sub"))
manage = hub.auth.has_permission(caller, hub.COUNT_MANAGE)
rows = [dict(r._mapping) for r in repo.conn.execute(q)]
rows = [r for r in rows if manage or (mine and (r["custodian_email"] or "").strip().lower() in mine)][:limit]
return {"counts": [{"location_id": r["location_id"], "count_date": r["count_date"].isoformat(),
"status": r["status"], "custodian_email": r["custodian_email"]} for r in rows]}
@FA.use_case(kind="query", route="GET /manage/asset-counts/{location_id}/{count_date}", permission=hub.COUNT_MANAGE)
def show_managed_asset_count(caller: dict, location_id: Code, count_date: IsoDate) -> dict:
"""One count with its lines, for whoever manages counts (the accountant): any location's, to decide its differences and
close it. A custodian reads their own with show_asset_count."""
with hub.repository() as repo:
return _count_view(_load_count(repo, location_id, day(count_date)))
@FA.use_case(kind="query", route="GET /manage/asset-counts", permission=hub.COUNT_MANAGE)
def list_managed_asset_counts(caller: dict, status: Optional[Code] = None, limit: Limit = 50) -> dict:
"""Every count, newest first, for whoever manages counts."""
with hub.repository() as repo:
from sqlalchemy import select
q = select(T.fa_count).order_by(T.fa_count.c.count_date.desc())
if status:
q = q.where(T.fa_count.c.status == status)
rows = [dict(r._mapping) for r in repo.conn.execute(q)][:limit]
return {"counts": [{"location_id": r["location_id"], "count_date": r["count_date"].isoformat(),
"status": r["status"], "custodian_email": r["custodian_email"]} for r in rows]}