dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/use_cases.py
"""The use cases — what a person (or their agent, or a neighbour's service client) can do with the asset register.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere else: WHO
MAY (a PERMISSION on the declaration, never a role's name) and WHICH ROWS (`_sees_*`, from the caller: a row the caller
may not see is a 404, never a 403). The rules live in `domain/`; what several use cases share in `services.py`.
The module is split by what the use cases are about; every part registers on the one `FA` brick declared here:
uc_settings areas, keys, classes, account determination, rates, controls (FA-01, FA-02)
uc_assets the register, acquisitions, assets under construction, usage, AI suggestions (FA-03..FA-11)
uc_depreciation simulate, approve, post, reverse a run; the year close (FA-12..FA-15)
uc_transactions retirement, transfer, revaluation, impairment, approval and reversal (FA-16..FA-20)
uc_allowances the UK capital allowances computation (FA-21, FA-22)
uc_reports counts, the history sheet, reconciliation to gl, what gl asks (FA-23..FA-26)
uc_migration the Odoo slice (FA-30, FA-31)
An asset's values are NEVER stored: they are the SUM of its transactions and postings. The tax area never posts a journal.
A second person is a different email AND a different sub, for every approval here.
"""
from contextlib import contextmanager
from datetime import date, timedelta
from erp_platform import auth
from erp_platform.bricks import Brick, Refused
from sqlalchemy.exc import IntegrityError
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.infrastructure import seams
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared. Which people hold
# which is the composition's role catalogue — roles.yaml beside this package is the recommended default.
READ, READ_OWN, WRITE = "fa:asset:read", "fa:asset:read_own", "fa:asset:write"
ACQ_WRITE, AUC_SETTLE, USAGE = "fa:acquisition:write", "fa:auc:settle", "fa:usage:write"
SIMULATE, RUN, APPROVE_RUN = "fa:depreciation:simulate", "fa:depreciation:run", "fa:depreciation:approve"
TX_WRITE, TX_APPROVE = "fa:transaction:write", "fa:transaction:approve"
CA_WRITE, CA_READ, CA_APPROVE = "fa:capital_allowance:write", "fa:capital_allowance:read", "fa:capital_allowance:approve"
COUNT_MANAGE, COUNT_WRITE = "fa:count:manage", "fa:count:write"
REPORT, YEAR_CLOSE = "fa:report:read", "fa:year:close"
SETTINGS_WRITE, CONTROLS = "fa:settings:write", "fa:settings:controls"
MIG_READ, MIG_SIGN = "fa:migration:read", "fa:migration:sign"
# machine permissions: a service client holds them, never a person, and no wildcard reaches them
NUMBER_CLAIM, INTAKE, JOURNAL_POST = "fa:number_series:claim", "fa:acquisition:intake", "fa:journal:post"
DEP_READ, AI_SUGGEST = "fa:depreciation:read", "fa:asset:ai_suggest"
MIG_LOAD, MIG_RECONCILE = "fa:migration:load", "fa:migration:reconcile"
MACHINE = (NUMBER_CLAIM, INTAKE, JOURNAL_POST, DEP_READ, AI_SUGGEST, MIG_LOAD, MIG_RECONCILE)
# what a control governs: whoever holds one of these may not hold fa:settings:controls (a configuration error)
CONTROLLED = (WRITE, ACQ_WRITE, APPROVE_RUN, TX_WRITE, TX_APPROVE, CA_WRITE, CA_APPROVE, SIMULATE, RUN)
FA = Brick("fixed-assets", version="0.1.0", ids_as_strings=True, machine_permissions=MACHINE, instructions=(
"Fixed assets & depreciation: what the business owns, what it cost, what it is worth in the books, in IFRS and for "
"tax. An asset's values are the SUM of its transactions per depreciation area; nothing stores a net book value. "
"Depreciation is derived from the asset's values, its key and its life, cumulatively, so no period drifts by a "
"penny: simulate a period (simulate_depreciation), a second person approves it (approve_depreciation_run), the system "
"posts one journal per area to that area's gl ledger, and a reversal is a reversing run. A retirement, transfer, "
"revaluation or impairment is proposed by one person and approved by another (never the same email or sign-in). "
"The UK tax area (capital allowances) never posts. Purchasing hands a bill line over with POST /acquisitions."))
# The process view (BRICKS.md): the signed workflows as BPMN, GET /process/bpmn.
FA.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text_ = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text_ or "duplicate key" in text_ or "UNIQUE constraint" in text_
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a write conflict on a claim row can surface at COMMIT (the loser's COMMIT raises 23505), after the use case
has returned - so it is caught here, around the commit, and answered as the same retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_fixed_assets.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
repository = _repository_default # tests and a composition swap in their own
@contextmanager
def decide():
"""A broken business rule is a 409 with the rule's own words; a neighbour that is down says so; a lost race on a
claim is a 409 to retry. Through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
except IntegrityError:
raise Refused(409, "Someone else wrote the same record a moment ago — load it again and retry") from None
def retrying(unit, attempts: int = 6):
"""Run one unit of work (its own `with repository()`), again when a claim it made collided with a concurrent writer
(a plain INSERT of a key another transaction took: 23505, or a lost version claim). Each attempt is a NEW
transaction — an aborted one cannot go on — and re-reads what the winner committed."""
from erp_platform import repository as platform_repository
for attempt in range(attempts):
try:
return unit()
except (IntegrityError, platform_repository.Conflict) as lost:
if attempt == attempts - 1:
raise
_ = lost
def today(repo, record: bool = False) -> date:
"""Business today (the one business_timezone), never backwards. A command passes record=True."""
return repo.today(record)
def tomorrow(repo) -> date:
return today(repo) + timedelta(days=1)
def need(caller: dict, *perms: str) -> None:
"""Refuse (403) a caller holding none of `perms` — for the few reads two permissions open."""
if not any(auth.has_permission(caller, p) for p in perms):
raise Refused(403, f"missing permission: {perms[0]}")
def control_holder_ok(caller: dict) -> None:
"""The controls are never held by whoever they control: a person who can write assets, approve depreciation or
transactions, or prepare allowances and also holds fa:settings:controls is a configuration error, refused here."""
clash = [p for p in CONTROLLED if auth.has_permission(caller, p)]
if clash:
raise Refused(403, "a control is never held by whoever it controls: this sign-in also holds "
f"{clash[0]}, so fa:settings:controls is refused (give the owner role to someone else)")
def sees_all_assets(caller: dict) -> bool:
return auth.has_permission(caller, READ)
# isort: off
# uc_reports FIRST: it registers the STATIC routes GET /assets/balances and GET /reports/*, which must precede
# GET /assets/{asset_id} (uc_assets) or "balances" is looked up as an asset id and gl's reconciliation gets a 404.
from brick_fixed_assets.application import uc_reports # noqa: E402,F401
from brick_fixed_assets.application import ( # noqa: E402,F401 (each registers its use cases on FA)
uc_allowances,
uc_assets,
uc_depreciation,
uc_migration,
uc_settings,
uc_transactions,
)
# isort: on