dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/uc_settings.py
"""Settings: the controls on the people who value assets (FA-01, FA-02). Every row is DATED and never edited: a change is a
later row, starting after today (the first row of a key may start today). The controls — the depreciation areas and the
ledger each posts to, the keys, account determination per class, the capitalisation threshold, number formats, whether a
depreciation posting needs a second person — are `fa:settings:controls`, refused to whoever they control and to
`fa:settings:write` (day-to-day: an AI step's mode)."""
import json
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, 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 ClassAreaIn, Code, Currency, LedgerCode, Life, Pct, Rate, day
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import (
AREA_PURPOSES,
CONVENTIONS,
METHODS,
POOLS,
VALUE_MODELS,
bound_life,
bound_rate,
)
from brick_fixed_assets.infrastructure import seams
from brick_fixed_assets.infrastructure import tables as T
from brick_fixed_assets.infrastructure.tables import ALL_SETTINGS, CONTROL_SETTINGS, DAY_TO_DAY_SETTINGS
FA = hub.FA
def _when(repo, valid_from: Optional[str], existing: list) -> date:
"""The date a new dated row starts: as given, else today for the first row of its key and tomorrow for a later one."""
if valid_from:
try:
return day(valid_from)
except ValueError:
raise Refused(422, "valid_from is a date like 2026-11-01") from None
return hub.today(repo) if not existing else hub.tomorrow(repo)
def _view(row: dict) -> dict:
return {k: (v.isoformat() if isinstance(v, date) else (str(v) if isinstance(v, Decimal) else v))
for k, v in row.items() if k not in ("created_at", "updated_at")}
@FA.use_case(kind="command", route="POST /settings/areas", permission=hub.CONTROLS, idempotent=True)
def set_depreciation_area(caller: dict, area_code: Code, name: Name, purpose: Code, valid_from: Optional[IsoDate] = None,
ledger_id: Optional[LedgerCode] = None, company_code: Optional[Code] = None,
posting: Optional[Code] = None, value_model: Code = "cost", values_from: Optional[Code] = None,
currency: Optional[Currency] = None) -> dict:
"""Define a depreciation area from a date: one way an asset is valued (book, IFRS, a ledger-specific area, or the UK
tax area) and the gl ledger it posts to. A book or IFRS area names ONE ledger (checked against gl's ledgers of
`company_code`); the tax area (purpose 'tax') never posts and is refused a ledger; no two areas post to one ledger.
A row is never edited: a change is a later valid_from."""
hub.control_holder_ok(caller)
with hub.repository() as repo, hub.decide():
if purpose not in AREA_PURPOSES:
raise RuleBroken(f"purpose is one of {', '.join(AREA_PURPOSES)}")
if value_model not in VALUE_MODELS:
raise RuleBroken(f"value_model is one of {', '.join(VALUE_MODELS)}")
if purpose == "tax":
if ledger_id is not None:
raise RuleBroken("the tax area never posts to a ledger: it is refused a ledger_id")
posting = "none"
elif posting is None:
posting = "per_run" if ledger_id is not None else "none"
if posting == "per_run" and ledger_id is None:
raise RuleBroken("an area that posts needs the gl ledger it posts to (ledger_id)")
if posting == "none" and ledger_id is not None:
raise RuleBroken("an area that posts nothing has no ledger")
today = hub.today(repo, True)
existing = repo.rows(T.fa_area, area_code=area_code)
start = _when(repo, valid_from, existing)
if ledger_id is not None:
if not company_code:
raise RuleBroken("company_code is needed to check the ledger against gl's ledgers of that company")
got = seams.company_ledgers(company_code, services.who(caller))
ids = [str(x["ledger_id"]) for x in got.get("ledgers", [])]
if ledger_id not in ids:
raise RuleBroken(f"gl has no ledger {ledger_id} for company {company_code} (it has {ids}): an area "
"naming a ledger gl does not have is refused")
for other in repo.areas(start):
if other["area_code"] != area_code and other["ledger_id"] == ledger_id:
raise RuleBroken(f"area {other['area_code']} already posts to ledger {ledger_id}: one area per ledger")
row, created = repo.put_dated(T.fa_area, {"area_code": area_code}, start,
{"name": name, "purpose": purpose, "ledger_id": ledger_id, "posting": posting,
"value_model": value_model, "values_from": values_from, "currency": currency,
"company_code": company_code if ledger_id is not None else None},
today, services.who(caller))
return {**_view(row), "created": created}
@FA.use_case(kind="command", route="POST /settings/keys", permission=hub.CONTROLS, idempotent=True)
def set_depreciation_key(caller: dict, key_code: Code, method: Code, valid_from: Optional[IsoDate] = None,
rate_pct: Optional[Pct] = None, life_periods: Optional[Life] = None, base: Code = "net_book_value",
acquisition_convention: Code = "full_month", retirement_convention: Code = "full_month",
stop_at: Code = "scrap") -> dict:
"""Define a depreciation method once, as a key (straight line over 5 years from the month of purchase means exactly
one thing everywhere): method, rate or life, base (cost | net_book_value), acquisition and retirement convention
(full_month | next_month | mid_month | daily | half_year) and whether it stops at zero or at scrap. A key an asset
uses can be superseded by a new dated row but never edited; the change applies prospectively (IAS 8 change in
estimate)."""
hub.control_holder_ok(caller)
with hub.repository() as repo, hub.decide():
if method not in METHODS:
raise RuleBroken(f"method is one of {', '.join(METHODS)}")
for c in (acquisition_convention, retirement_convention):
if c not in CONVENTIONS:
raise RuleBroken(f"a convention is one of {', '.join(CONVENTIONS)}")
if base not in ("cost", "net_book_value") or stop_at not in ("zero", "scrap"):
raise RuleBroken("base is cost | net_book_value and stop_at is zero | scrap")
bound_rate(rate_pct, "rate_pct")
bound_life(life_periods)
if method in ("declining_balance", "declining_then_straight") and rate_pct is None:
raise RuleBroken("a declining-balance key needs rate_pct")
today = hub.today(repo, True)
start = _when(repo, valid_from, repo.rows(T.fa_dep_key, key_code=key_code))
row, created = repo.put_dated(T.fa_dep_key, {"key_code": key_code}, start,
{"method": method, "rate_pct": rate_pct, "life_periods": life_periods, "base": base,
"acq_conv": acquisition_convention, "ret_conv": retirement_convention,
"stop_at": stop_at}, today, services.who(caller))
return {**_view(row), "created": created}
@FA.use_case(kind="command", route="POST /settings/classes", permission=hub.CONTROLS, idempotent=True)
def set_asset_class(caller: dict, class_code: Code, name: Name, valid_from: Optional[IsoDate] = None,
is_auc: bool = False, is_low_value: bool = False, default_ca_pool: Code = "main",
number_format: Optional[Code] = None, areas: Optional[list[ClassAreaIn]] = None) -> dict:
"""Define a kind of asset from a date: its defaults per area (the key and life an asset created in it takes) and the
capital allowance pool it defaults to. An asset created from tomorrow takes these; a row dated today or earlier is
refused (a first row of a class may start today). An area must exist and a key must exist."""
hub.control_holder_ok(caller)
with hub.repository() as repo, hub.decide():
if default_ca_pool not in POOLS:
raise RuleBroken(f"default_ca_pool is one of {', '.join(POOLS)}")
today = hub.today(repo, True)
start = _when(repo, valid_from, repo.rows(T.fa_class, class_code=class_code))
for ar in areas or []:
if repo.area(ar.area_code, start) is None:
raise RuleBroken(f"area {ar.area_code} is not defined on {start.isoformat()}: define it first")
if repo.key(ar.key_code, start) is None:
raise RuleBroken(f"depreciation key {ar.key_code} is not defined on {start.isoformat()}: define it first")
row, created = repo.put_dated(T.fa_class, {"class_code": class_code}, start,
{"name": name, "is_auc": is_auc, "is_low_value": is_low_value,
"default_ca_pool": default_ca_pool, "number_format": number_format},
today, services.who(caller))
out = {**_view(row), "created": created, "areas": []}
for ar in areas or []:
r, _ = repo.put_dated(T.fa_class_area, {"class_code": class_code, "area_code": ar.area_code}, start,
{"key_code": ar.key_code, "life_periods": ar.life_periods}, today, services.who(caller))
out["areas"].append(_view(r))
return out
@FA.use_case(kind="command", route="POST /settings/account-determination", permission=hub.CONTROLS, idempotent=True)
def set_account_determination(caller: dict, class_code: Code, area_code: Code, valid_from: Optional[IsoDate] = None,
cost_account_id: Optional[BigId] = None,
accumulated_depreciation_account_id: Optional[BigId] = None,
depreciation_expense_account_id: Optional[BigId] = None, gain_account_id: Optional[BigId] = None,
loss_account_id: Optional[BigId] = None, revaluation_reserve_account_id: Optional[BigId] = None,
impairment_account_id: Optional[BigId] = None,
impairment_reversal_account_id: Optional[BigId] = None) -> dict:
"""Which gl accounts a class posts to in one area, from a date (cost, accumulated depreciation, depreciation expense,
gain, loss, revaluation reserve, impairment, impairment reversal). The accounts are gl's account_ids; the tax area
has none (it never posts)."""
hub.control_holder_ok(caller)
with hub.repository() as repo, hub.decide():
today = hub.today(repo, True)
area = repo.area(area_code, today) or (repo.rows(T.fa_area, area_code=area_code) or [None])[-1]
if area is None:
raise RuleBroken(f"area {area_code} is not defined: define it first")
if area["purpose"] == "tax":
raise RuleBroken("the tax area never posts: it has no accounts")
start = _when(repo, valid_from, repo.rows(T.fa_account_map, class_code=class_code, area_code=area_code))
row, created = repo.put_dated(
T.fa_account_map, {"class_code": class_code, "area_code": area_code}, start,
{"cost_account_id": cost_account_id, "accumulated_depreciation_account_id": accumulated_depreciation_account_id,
"depreciation_expense_account_id": depreciation_expense_account_id, "gain_account_id": gain_account_id,
"loss_account_id": loss_account_id, "revaluation_reserve_account_id": revaluation_reserve_account_id,
"impairment_account_id": impairment_account_id, "impairment_reversal_account_id": impairment_reversal_account_id},
today, services.who(caller))
return {**_view(row), "created": created}
@FA.use_case(kind="command", route="POST /settings/ca-rates", permission=hub.CONTROLS, idempotent=True)
def set_ca_rate(caller: dict, rate_key: Code, value: Rate, legislation_ref: Optional[Note] = None,
valid_from: Optional[IsoDate] = None) -> dict:
"""Set a capital allowances rate or limit from a date, with its legislation reference: aia_limit, wda_main_pct,
wda_special_pct, sba_pct, small_pool_limit, fya_full_expensing_pct, fya_50_pct. They are dated owner-controlled rows,
not code."""
hub.control_holder_ok(caller)
with hub.repository() as repo, hub.decide():
if rate_key not in ("aia_limit", "wda_main_pct", "wda_special_pct", "sba_pct", "small_pool_limit",
"fya_full_expensing_pct", "fya_50_pct"):
raise RuleBroken("rate_key is aia_limit | wda_main_pct | wda_special_pct | sba_pct | small_pool_limit | "
"fya_full_expensing_pct | fya_50_pct")
if rate_key.endswith("_pct") and value > 100:
raise RuleBroken("a percentage rate is from 0 to 100")
today = hub.today(repo, True)
start = _when(repo, valid_from, repo.rows(T.fa_ca_rate, rate_key=rate_key))
row, created = repo.put_dated(T.fa_ca_rate, {"rate_key": rate_key}, start,
{"value": value, "legislation_ref": legislation_ref}, today, services.who(caller))
return {**_view(row), "created": created}
def _put_setting(caller: dict, key: str, value, valid_from: Optional[str], allowed: dict) -> dict:
with hub.repository() as repo, hub.decide():
if key not in allowed:
if key in ALL_SETTINGS:
what = "a control (fa:settings:controls)" if key in CONTROL_SETTINGS else "a day-to-day setting"
raise Refused(403, f"{key} is {what}: this sign-in may not change it")
raise RuleBroken(f"no such setting: {key}")
kind, _ = ALL_SETTINGS[key]
raw = _coerce(key, kind, value)
today = hub.today(repo, True)
start = _when(repo, valid_from, repo.rows(T.fa_settings, setting_key=key))
row, created = repo.put_dated(T.fa_settings, {"setting_key": key}, start, {"value": json.dumps(raw)}, today,
services.who(caller))
return {"setting_key": key, "value": raw, "valid_from": row["valid_from"].isoformat(), "created": created}
def _coerce(key: str, kind: str, value):
try:
if kind == "decimal":
d = Decimal(str(value))
if not d.is_finite() or d < 0 or d >= Decimal("1e15") or d.as_tuple().exponent < -2:
raise ValueError
return str(d)
if kind == "bool":
if not isinstance(value, bool):
raise ValueError
return value
if kind == "int":
return None if value is None else int(value)
v = str(value)
if key == "transfer_variant_default" and v not in ("gross", "net"):
raise ValueError
if key == "asset_number_format" and "{seq" not in v:
raise ValueError
if key.startswith("ai.") and v not in ("off", "suggest"):
raise ValueError
return v[:200]
except (ValueError, ArithmeticError):
raise RuleBroken(f"{key}: {value!r} is not a valid value") from None
@FA.use_case(kind="command", route="POST /settings/controls", permission=hub.CONTROLS, idempotent=True)
def set_fa_control(caller: dict, setting_key: Code, value: str, valid_from: Optional[IsoDate] = None) -> dict:
"""Set a control from a date: capitalisation_threshold (money), depreciation_needs_approval (true | false, warned: a
wrong life or key is caught by a second reader before it is in the books), asset_number_format (needs {seq:4}),
transfer_variant_default (gross | net), aia_group_allocation, intercompany_clearing_account_id and
asset_sale_clearing_account_id (gl account ids). A threshold changed mid-year makes two similar purchases land
differently: change it from a new year."""
hub.control_holder_ok(caller)
parsed = _parse(value, ALL_SETTINGS.get(setting_key, ("str", None))[0])
out = _put_setting(caller, setting_key, parsed, valid_from, CONTROL_SETTINGS)
if setting_key == "depreciation_needs_approval" and parsed is False:
out["warning"] = "a depreciation posting no longer needs a second person: a wrong life or key is caught only by you"
return out
def _parse(value: str, kind: str):
if kind == "bool":
if value.lower() in ("true", "false"):
return value.lower() == "true"
raise Refused(422, "value is true or false")
return value
@FA.use_case(kind="command", route="POST /settings/day-to-day", permission=hub.SETTINGS_WRITE, idempotent=True)
def set_fa_setting(caller: dict, setting_key: Code, value: str, valid_from: Optional[IsoDate] = None) -> dict:
"""Change a day-to-day setting from a date: an AI step's mode (ai.ai_suggest_asset_class.mode: off | suggest). A
control (area, key, account determination, threshold, approval switch, number format) is refused here: it is
fa:settings:controls and never held by whoever it controls."""
return _put_setting(caller, setting_key, value, valid_from, DAY_TO_DAY_SETTINGS)
@FA.use_case(kind="query", route="GET /settings", permission=hub.READ)
def show_fa_settings(caller: dict, on: Optional[IsoDate] = None) -> dict:
"""Every dated setting row in force on a date (default today): the areas and their ledgers, the keys, the classes with
their defaults and accounts, the capital allowance rates, and the settings. The controls section is shown to the
owner only (fa:settings:controls)."""
with hub.repository() as repo:
when = day(on) if on else hub.today(repo)
owner = hub.auth.has_permission(caller, hub.CONTROLS)
out = {"on": when.isoformat(), "today": hub.today(repo).isoformat(),
"settings": repo.settings_view(when, controls=owner)}
out["classes"] = [{**_view(c), "areas": [_view(a) for a in repo.class_areas(c["class_code"], when)]}
for c in repo.classes(when)]
if owner:
out["areas"] = [_view(a) for a in repo.areas(when)]
codes = sorted({r["key_code"] for r in repo.rows(T.fa_dep_key)})
out["keys"] = [_view(k) for k in (repo.key(c, when) for c in codes) if k]
out["account_determination"] = [
_view(r) for r in (repo.account_map(c["class_code"], a["area_code"], when)
for c in repo.classes(when) for a in repo.areas(when)) if r]
out["ca_rates"] = {k: str(v) for k, v in repo.ca_rates(when).items()}
return out
@FA.use_case(kind="query", route="GET /asset-classes", permission=hub.READ)
def list_asset_classes(caller: dict, on: Optional[IsoDate] = None) -> dict:
"""The asset classes in force on a date (default today), with the capitalisation threshold: the order line's asset
picker in purchasing (a new asset per line by class, or an existing asset_id)."""
with hub.repository() as repo:
when = day(on) if on else hub.today(repo)
thr = repo.setting("capitalisation_threshold", when)
return {"classes": [{"class_code": c["class_code"], "name": c["name"], "is_auc": bool(c["is_auc"]),
"is_low_value": bool(c["is_low_value"]), "capitalisation_threshold": str(thr)}
for c in repo.classes(when)]}