dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/uc_transactions.py
"""Changes in value (FA-16 .. FA-20): retirement (all or part, with gain or loss), transfer between companies, revaluation
and impairment — each PROPOSED by one person, APPROVED by another (never the same email or the same sign-in), then posted per
area to its ledger, and reversed only by a reversing transaction.
The figures are computed at the proposal and STORED (a posted document keeps its computed amounts; reads never recompute them).
The asset's version is the claim: an approval refuses (asset_unchanged) once anything changed the asset since the proposal, and
two partial retirements together can never retire more than 100%."""
import json
from datetime import date
from decimal import Decimal
from fractions import Fraction
from math import ceil
from typing import Optional
from erp_platform import work
from erp_platform.bricks import Refused
from erp_platform.types import Limit, 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, Hex64, IsoDate, PosMoney, Reason, Units, day
from brick_fixed_assets.application.uc_assets import _asset, _visible
from brick_fixed_assets.domain import changes
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import ZERO, money, period_add, period_of, same_person
from brick_fixed_assets.infrastructure import tables as T
FA = hub.FA
def _moment() -> str:
return work.now().strftime("%Y%m%dT%H%M%S%f")
def _frac_text(f: Fraction) -> str:
return f"{f.numerator}/{f.denominator}"
def _share_retired(repo, asset_id: str) -> Fraction:
"""The share of the ORIGINAL asset that proposed, approved or posted retirements already claim."""
total = Fraction(0)
for t in repo.tx_rows(asset_id, statuses=("proposed", "approved", "posted")):
if t["kind"] == "retirement":
total += Fraction(json.loads(t["params"] or "{}").get("orig_share", "0/1"))
return total
def _reference_area(repo, a, today: date) -> str:
areas = a.area_codes()
for code in areas:
ar = repo.area(code, today)
if ar and ar["purpose"] in ("book", "ifrs", "ledger_specific") and ar["posting"] == "per_run":
return code
return areas[0]
def _current_gate(repo, a, code: str, when: date, plan) -> int:
"""The depreciation in the book areas must be up to date at a valuation date: planned to the period of `when` equals
what is posted. Returns the catch-up still to post (zero when current)."""
posted = services.posted_total(repo, a.asset_id, code)
return (plan.cum(period_of(when)) if plan else ZERO) - posted
def _view(repo, txid: str) -> dict:
tx = repo.transaction(txid)
if tx is None:
raise Refused(404, f"no transaction {txid}")
out = services.tx_view(tx["root"], tx["areas"])
legs = repo.group(tx["root"]["group_id"])
if len(legs) > 1:
out["legs"] = [{"transaction_id": t["transaction_id"], "asset_id": t["asset_id"], "company_code": t["company_code"],
"leg": t["leg"], "status": t["status"]} for t in legs]
return out
# ── retirement (FA-16) ────────────────────────────────────────────────────────────────────────
@FA.use_case(kind="command", route="POST /assets/{asset_id}/retirements", permission=hub.TX_WRITE)
def propose_retirement(caller: dict, asset_id: Code, retirement_date: IsoDate, percent: Optional[PosMoney] = None,
amount: Optional[PosMoney] = None, quantity: Optional[Units] = None, proceeds: PosMoney = ZERO,
sale_clearing_account_id: Optional[Code] = None, memo: Optional[Note] = None) -> dict:
"""Propose to sell or scrap an asset, all of it or part (by percent of the original, by cost amount or by quantity), and
have the gain or loss worked out in every area. Depreciation is charged up to the retirement by the key's retirement
convention (the catch-up), cost and accumulated depreciation leave proportionally, the gain or loss is proceeds minus the
net book value that left, and the journal debits asset sale clearing with the proceeds. The retirement date is the date it
LEFT: today or earlier, never later. More than what remains (including other proposals) is refused. NOTHING posts until
a second person approves (approve_asset_transaction)."""
def unit():
with hub.repository() as repo:
a = _visible(repo, caller, asset_id)
today = hub.today(repo, True)
when = day(retirement_date)
if when > today:
raise RuleBroken("a retirement cannot be dated after today (not_future): record it on the day it left")
if a.root["status"] != "active":
raise RuleBroken(f"the asset is {a.root['status']}: only an active asset is retired")
services.check_year_open(repo, a.root["company_code"], when)
remaining = Fraction(1) - _share_retired(repo, asset_id)
ref = _reference_area(repo, a, today)
ref_values = services.area_values(repo, asset_id, ref)
share_rem = changes.retirement_fraction(
ref_values, remaining, percent=percent, amount=amount, quantity=quantity,
total_quantity=a.root["quantity"])
orig_share = share_rem * remaining
clearing = sale_clearing_account_id or repo.setting("asset_sale_clearing_account_id", today)
if proceeds > 0 and clearing is None:
raise RuleBroken("name the asset sale clearing account (sale_clearing_account_id): the sale is invoiced "
"through ar to it")
deltas, targets = {}, {}
for code in a.area_codes():
area = repo.area(code, today)
v = services.area_values(repo, asset_id, code)
posting = bool(area and area["posting"] == "per_run" and area["ledger_id"] is not None)
catch = ZERO
if posting:
plan = services.build_plan(repo, a, code)
if plan is not None:
key = repo.key(a.area_on(code, when)["key_code"], when)
g = plan.retirement_weight(key["ret_conv"] or "full_month", when)
target = money(plan.cum_with(period_of(when), g))
targets[code] = str(target)
catch = target - services.posted_total(repo, asset_id, code)
deltas[code] = changes.retire(v, catch, share_rem, proceeds)
params = {"share_of_remaining": _frac_text(share_rem), "orig_share": _frac_text(orig_share),
"proceeds": str(proceeds), "retirement_date": when.isoformat(), "catch_up_target": targets,
"memo": memo, "offset": {"account_id": int(clearing)} if clearing is not None and proceeds > 0 else None}
txid = f"ret:{asset_id}:{_moment()}"
services.new_tx(repo, caller, transaction_id=txid, asset=a, kind="retirement", value_date=when, deltas=deltas,
params=params, status="proposed", asset_version=a.version, source_ref=f"retire:{asset_id}")
return _view(repo, txid)
with hub.decide():
return hub.retrying(unit, 1)
# ── revaluation and impairment (FA-19, FA-20) ─────────────────────────────────────────────────
def _evidence(sha: str) -> str:
return f"raw/fixed-assets/valuations/{sha}.pdf"
@FA.use_case(kind="command", route="POST /assets/{asset_id}/revaluations", permission=hub.TX_WRITE)
def propose_revaluation(caller: dict, asset_id: Code, area_code: Code, revalued_amount: PosMoney, valuation_date: IsoDate,
evidence_sha256: Hex64, memo: Optional[Note] = None) -> dict:
"""Propose to revalue an asset in ONE area to a valuer's figure, where that area's policy is the revaluation model (IAS 16.31;
FRS 102 17.15B): the accumulated depreciation is eliminated against cost (35(b)), an increase is credited to the
revaluation reserve (other comprehensive income), a decrease goes first against THIS asset's own reserve and the rest to
profit or loss (40). Other areas are untouched. The valuer's report is kept as the object
raw/fixed-assets/valuations/<sha256>.pdf (`evidence_sha256` names it; it is required). The valuation date is a fact: today or
earlier. Depreciation must be posted up to it first."""
def unit():
with hub.repository() as repo:
a = _visible(repo, caller, asset_id)
today = hub.today(repo, True)
when = day(valuation_date)
area = repo.area(area_code, today)
if area is None or area_code not in a.area_codes():
raise RuleBroken(f"the asset has no area {area_code}")
if area["value_model"] != "revaluation":
raise RuleBroken(f"area {area_code} keeps the cost model: a revaluation is for an area whose policy is the "
"revaluation model (Settings: areas)")
if when > today:
raise RuleBroken("the valuation date is a fact: today or earlier")
if a.root["status"] != "active":
raise RuleBroken(f"the asset is {a.root['status']}: only an active asset is revalued")
services.check_year_open(repo, a.root["company_code"], when)
plan = services.build_plan(repo, a, area_code)
behind = _current_gate(repo, a, area_code, when, plan)
if behind != 0:
raise RuleBroken(f"depreciation in area {area_code} is not up to date at {when.isoformat()} (it differs by "
f"{behind}): post the depreciation run for {period_of(when)} first (depreciation_current)")
v = services.area_values(repo, asset_id, area_code)
d = changes.revalue(v, revalued_amount)
params = {"offset": None, "memo": memo, "areas": {area_code: {"new_nbv": str(revalued_amount)}},
"revalued_amount": str(revalued_amount), "nbv_before": str(v.nbv)}
txid = f"rev-model:{asset_id}:{_moment()}"
services.new_tx(repo, caller, transaction_id=txid, asset=a, kind="revaluation", value_date=when,
deltas={area_code: d}, params=params, status="proposed", evidence_key=_evidence(evidence_sha256),
asset_version=a.version, source_ref=f"revalue:{asset_id}")
return _view(repo, txid)
with hub.decide():
return hub.retrying(unit, 1)
def _unreversed(repo, asset_id: str, code: str) -> list:
"""[(amount left, periods remaining when it was made)] of an asset's impairments, reversals taken off the oldest first."""
made, taken = [], ZERO
for t in repo.tx_rows(asset_id, statuses=("posted",)):
prm = json.loads(t["params"] or "{}")
ar = (prm.get("areas") or {}).get(code)
if not ar:
continue
if t["kind"] == "impairment":
made.append([Decimal(ar["amount"]), Fraction(ar.get("remaining", "0/1"))])
elif t["kind"] == "impairment_reversal":
taken += Decimal(ar["amount"])
out = []
for amount, rem in made:
use = min(amount, taken)
taken -= use
if amount - use > 0:
out.append((amount - use, rem))
return out
@FA.use_case(kind="command", route="POST /assets/{asset_id}/impairments", permission=hub.TX_WRITE)
def propose_impairment(caller: dict, asset_id: Code, amount: PosMoney, review_date: IsoDate, evidence_sha256: Hex64,
area_codes: Optional[list[Code]] = None, reversal: bool = False, memo: Optional[Note] = None) -> dict:
"""Propose an impairment (or its reversal when things recover, `reversal`) in the named areas (default: every area that
posts): the loss is charged first against the asset's revaluation reserve and the rest to impairment loss (IAS 36), and
future depreciation is on the lower carrying amount over the remaining life. An impairment above the net book value is
refused (not_above_nbv). A reversal is capped at what the carrying amount would have been without the impairment (IAS 36.117)
and shown separately in the history sheet. The review is kept as raw/fixed-assets/valuations/<sha256>.pdf. A second person
approves."""
def unit():
with hub.repository() as repo:
a = _visible(repo, caller, asset_id)
today = hub.today(repo, True)
when = day(review_date)
if when > today:
raise RuleBroken("the review date is a fact: today or earlier")
if a.root["status"] != "active":
raise RuleBroken(f"the asset is {a.root['status']}: only an active asset is impaired")
services.check_year_open(repo, a.root["company_code"], when)
codes = area_codes or [x["area_code"] for x in repo.posting_areas(today) if x["area_code"] in a.area_codes()]
if not codes:
raise RuleBroken("name an area that posts: an impairment is a book figure")
deltas, areas_prm = {}, {}
for code in codes:
if code not in a.area_codes():
raise RuleBroken(f"the asset has no area {code}")
plan = services.build_plan(repo, a, code)
behind = _current_gate(repo, a, code, when, plan)
if behind != 0:
raise RuleBroken(f"depreciation in area {code} is not up to date at {when.isoformat()} (it differs by "
f"{behind}): post the depreciation run for {period_of(when)} first (depreciation_current)")
v = services.area_values(repo, asset_id, code)
if reversal:
rem = plan.remaining_periods(period_of(when)) if plan else Fraction(0)
cap = changes.reversal_cap(_unreversed(repo, asset_id, code), rem)
d = changes.reverse_impairment(v, amount, cap)
areas_prm[code] = {"new_nbv": str(v.nbv + amount), "amount": str(amount), "cap": str(cap)}
else:
d = changes.impair(v, amount)
rem = plan.remaining_periods(period_of(when)) if plan else Fraction(0)
areas_prm[code] = {"new_nbv": str(v.nbv - amount), "amount": str(amount), "remaining": _frac_text(rem)}
deltas[code] = d
txid = f"imp:{asset_id}:{_moment()}"
services.new_tx(repo, caller, transaction_id=txid, asset=a, kind="impairment_reversal" if reversal else "impairment",
value_date=when, deltas=deltas, params={"offset": None, "memo": memo, "areas": areas_prm},
status="proposed", evidence_key=_evidence(evidence_sha256), asset_version=a.version,
source_ref=f"impair:{asset_id}")
return _view(repo, txid)
with hub.decide():
return hub.retrying(unit, 1)
# ── transfer (FA-18) ──────────────────────────────────────────────────────────────────────────
@FA.use_case(kind="command", route="POST /assets/{asset_id}/transfers", permission=hub.TX_WRITE)
def propose_asset_transfer(caller: dict, asset_id: Code, transfer_date: IsoDate, to_asset_id: Optional[Code] = None,
to_company_code: Optional[Code] = None, to_class_code: Optional[Code] = None,
variant: Optional[Code] = None, memo: Optional[Note] = None) -> dict:
"""Propose to move an asset to another company in the group (FA-18): ONE transaction with two legs — the sender retires it
(credit cost, debit accumulated depreciation, debit inter-company clearing with the net book value) and the receiver
acquires an asset with the same values (the other way round), in each posting area to each company's ledger, the
remaining life carried. Name the receiving asset (`to_asset_id`) or a company and class (a new asset is created for it).
Variant gross carries cost and accumulated depreciation, net carries the net book value as cost (default: the owner's
setting). Both companies' periods must be open at approval. A move between cost centres in one company is
a dated master change (change_asset_master), not a transfer."""
def unit():
with hub.repository() as repo:
a = _visible(repo, caller, asset_id)
today = hub.today(repo, True)
when = day(transfer_date)
if when > today:
raise RuleBroken("a transfer cannot be dated after today")
if a.root["status"] != "active":
raise RuleBroken(f"the asset is {a.root['status']}: only an active asset is transferred")
if _share_retired(repo, asset_id) > 0:
raise RuleBroken("part of this asset is already claimed by a retirement: transfer what remains after it posts")
services.check_year_open(repo, a.root["company_code"], when)
var = variant or repo.setting("transfer_variant_default", today)
ic = repo.setting("intercompany_clearing_account_id", today)
if ic is None:
raise RuleBroken("the owner has not set the inter-company clearing account "
"(intercompany_clearing_account_id): Settings: controls")
if to_asset_id:
b = _asset(repo, to_asset_id)
if b.root["company_code"] == a.root["company_code"]:
raise RuleBroken("a transfer is between companies: a move inside one company is a master change")
if b.root["status"] in ("retired", "deactivated"):
raise RuleBroken("the receiving asset is retired or deactivated")
else:
if not (to_company_code and to_class_code):
raise RuleBroken("name the receiving asset, or the receiving company and class")
if to_company_code == a.root["company_code"]:
raise RuleBroken("a transfer is between companies: a move inside one company is a master change")
b = _create_receiver(repo, caller, a, to_company_code, to_class_code, today)
out_d, in_d = {}, {}
areas_prm = {}
receiving = _receiving_areas(repo, a, b, today)
for code in a.area_codes():
v = services.area_values(repo, asset_id, code)
plan = services.build_plan(repo, a, code)
area = repo.area(code, today)
if plan is not None and area and area["posting"] == "per_run":
behind = _current_gate(repo, a, code, when, plan)
if behind != 0:
raise RuleBroken(f"depreciation in area {code} is not up to date at {when.isoformat()} (it differs by "
f"{behind}): post the depreciation run for {period_of(when)} first")
o, i = changes.transfer_values(v, var)
theirs = receiving[code]
out_d[code], in_d[code] = o, i
rem = plan.remaining_periods(period_of(when)) if plan else Fraction(0)
br = b.area_on(theirs, when)
scrap = Decimal(str(br["scrap_value"] or 0)) if br else ZERO
areas_prm[theirs] = {"depreciable": str(max(i.cost - i.acc_dep - scrap, ZERO)), "remaining": max(ceil(rem), 1),
"first_period": period_add(period_of(when), 1)}
in_d = {receiving[code]: d for code, d in in_d.items()}
head = f"xfer:{asset_id}:{_moment()}"
common = {"variant": var, "areas": areas_prm, "memo": memo, "from_asset_id": asset_id, "to_asset_id": b.asset_id,
"offset": {"account_id": int(ic)}}
services.new_tx(repo, caller, transaction_id=head, asset=a, kind="transfer", value_date=when, deltas=out_d,
params={**common, "direction": "out"}, status="proposed", asset_version=a.version,
source_ref=f"transfer:{asset_id}")
services.new_tx(repo, caller, transaction_id=f"{head}:in", asset=b, kind="transfer", value_date=when, deltas=in_d,
params={**common, "direction": "in"}, status="proposed", group_id=head, leg="in",
asset_version=b.version, source_ref=f"transfer:{asset_id}")
return _view(repo, head)
with hub.decide():
return hub.retrying(unit, 1)
def _receiving_areas(repo, a, b, today: date) -> dict:
"""The receiver's area for each of the sender's: an area names ONE ledger, which belongs to ONE company, so the company
receiving posts to ITS ledger through its own area of the same purpose (the same area when it serves both, as the tax
area does)."""
mine = {c: repo.area(c, today) for c in b.area_codes()}
out = {}
for code in a.area_codes():
area = repo.area(code, today)
same = mine.get(code)
if same is not None and same.get("company_code") in (None, b.root["company_code"]):
out[code] = code
continue
match = [c for c, x in sorted(mine.items()) if x and area and x["purpose"] == area["purpose"]
and x.get("company_code") in (None, b.root["company_code"])]
if not match:
raise RuleBroken(f"{b.root['company_code']} has no depreciation area of purpose {area['purpose'] if area else code} "
f"to receive area {code}: add the class's area for it (Settings: controls)")
out[code] = match[0]
return out
def _create_receiver(repo, caller: dict, a, company: str, class_code: str, today: date):
from sqlalchemy import insert
from brick_fixed_assets.application.uc_assets import _gate_class, _new_asset
k = _gate_class(repo, class_code, today)
fmt = k["number_format"] or repo.setting("asset_number_format", today)
request_id = f"xfer-new:{a.asset_id}:{_moment()}"
number = repo.claim_number(class_code, today.year, fmt, request_id)
m = a.master_on(today) or {}
from brick_fixed_assets.application.inputs import AssetAreaIn
receiving = {c["area_code"]: repo.area(c["area_code"], today) for c in repo.class_areas(class_code, today)}
def theirs(code: str) -> Optional[str]:
"""The receiving class's area for the sender's: the same code when it serves the receiving company, else the one of
the same purpose (each company posts to its own ledger through its own area)."""
mine = repo.area(code, today)
if code in receiving and receiving[code] and receiving[code].get("company_code") in (None, company):
return code
return next((c for c, x in sorted(receiving.items()) if x and mine and x["purpose"] == mine["purpose"]
and x.get("company_code") in (None, company)), None)
areas = [AssetAreaIn(area_code=theirs(r["area_code"]), key_code=r["key_code"], life_periods=r["life_periods"],
total_units=r["total_units"], scrap_value=r["scrap_value"])
for r in a.areas if theirs(r["area_code"])
and r["valid_from"] == max(x["valid_from"] for x in a.areas if x["area_code"] == r["area_code"])]
b = _new_asset(repo, caller, company_code=company, class_code=class_code, description=a.root["description"],
request_id=request_id, asset_no=number, sub_no=0, serial_no=a.root["serial_no"],
inventory_no=a.root["inventory_no"], bp_id=a.root["bp_id"], quantity=a.root["quantity"],
cost_centre=None, location_id=None, custodian_email=m.get("custodian_email"), is_car=a.root["is_car"],
new_and_unused=False, ca_pool=a.root["ca_pool"], areas=areas, today=today)
repo.conn.execute(insert(T.fa_asset_request).values(request_id=request_id, asset_id=b.asset_id, company_code=company))
b.version = repo.save_asset(b)
return b
# ── approve, reject, post, reverse ────────────────────────────────────────────────────────────
@FA.use_case(kind="command", route="POST /asset-transactions/{transaction_id}/approve", permission=hub.TX_APPROVE,
idempotent=True)
def approve_asset_transaction(caller: dict, transaction_id: Code) -> dict:
"""Approve a proposed retirement, transfer, revaluation or impairment (FA-17). Refused for whoever proposed it — the same
email under another sign-in, or the same sign-in under another email, is the same person (second_person) — and once the
asset changed since the proposal (asset_unchanged: propose again from the current values). The approval records who (email
and sign-in) and today's date; the system then posts the journals per area to each ledger (every company's and area's
period must be open). An approved transaction is never edited: a correction is a reversing transaction."""
actor = services.who(caller)
def unit():
with hub.repository() as repo:
from brick_fixed_assets.application.uc_assets import _post_gates
tx = repo.transaction(transaction_id)
if tx is None:
raise Refused(404, f"no transaction {transaction_id}")
r = tx["root"]
if r["group_id"] != transaction_id:
raise Refused(404, f"no transaction {transaction_id}")
if r["status"] in ("approved", "posted", "reversed"):
return transaction_id
if r["status"] != "proposed":
raise RuleBroken(f"the transaction is {r['status']}: only a proposed one is approved")
if same_person(services.email_of(caller), services.sub_of(caller), r["proposed_by_email"], r["proposed_by_sub"]):
raise RuleBroken("you proposed this (second_person: same email or same sign-in): another approver approves it")
legs = repo.group(transaction_id)
assets = repo.assets(l["asset_id"] for l in legs) # one round trip for every leg's asset and one for its transaction
txs = repo.transactions(l["transaction_id"] for l in legs)
for leg in legs:
a = assets[leg["asset_id"]]
if leg["asset_version"] is not None and leg["asset_version"] != a.version:
raise RuleBroken(f"asset {a.asset_id} changed since the proposal (asset_unchanged): propose again from "
"the current values")
today = hub.today(repo, True)
for company in sorted({leg["company_code"] for leg in legs}):
services.check_year_open(repo, company, r["value_date"])
_post_gates(repo, company, r["value_date"], actor)
for leg in legs:
t = txs[leg["transaction_id"]]
if leg["kind"] == "retirement":
prm = json.loads(leg["params"] or "{}")
for code, target in (prm.get("catch_up_target") or {}).items():
row = t["areas"].get(code)
if row is not None:
row["catch_up"] = Decimal(target) - services.posted_total(repo, leg["asset_id"], code)
t["root"]["status"] = "approved"
t["root"]["approved_by_email"], t["root"]["approved_by_sub"] = services.email_of(caller), services.sub_of(caller)
t["root"]["approved_on"] = today
repo.save_transaction(t["root"], t["areas"], t["version"])
a = assets[leg["asset_id"]]
a.version = repo.save_asset(a) # the claim: a racing approval or change loses here
return transaction_id
with hub.decide():
head = hub.retrying(unit, 1)
with hub.repository() as repo:
state = repo.transaction(head)["root"]["status"]
if state == "approved":
services.post_group_now(hub.repository, head, actor)
with hub.repository() as repo:
return _view(repo, head)
@FA.use_case(kind="command", route="POST /asset-transactions/{transaction_id}/reject", permission=hub.TX_APPROVE,
idempotent=True)
def reject_asset_transaction(caller: dict, transaction_id: Code, reason: Reason) -> dict:
"""Reject a proposal with a reason: it keeps the reason and posts nothing."""
with hub.repository() as repo, hub.decide():
tx = repo.transaction(transaction_id)
if tx is None or tx["root"]["group_id"] != transaction_id:
raise Refused(404, f"no transaction {transaction_id}")
if tx["root"]["status"] == "rejected":
return _view(repo, transaction_id)
if tx["root"]["status"] != "proposed":
raise RuleBroken(f"the transaction is {tx['root']['status']}: only a proposed one is rejected")
for leg in repo.group(transaction_id):
t = repo.transaction(leg["transaction_id"])
t["root"]["status"] = "rejected"
t["root"]["rejected_reason"] = reason
repo.save_transaction(t["root"], t["areas"], t["version"])
return _view(repo, transaction_id)
@FA.use_case(kind="command", route="POST /asset-transactions/{transaction_id}/post", permission=hub.JOURNAL_POST,
idempotent=True)
def post_asset_transaction(caller: dict, transaction_id: Code) -> dict:
"""The system posts an APPROVED transaction: one journal per company and posting area, keyed from the transaction so a
retry replays instead of posting twice; the tax area is recorded, never posted. Used to finish a posting gl could not
take the first time."""
with hub.repository() as repo, hub.decide():
tx = repo.transaction(transaction_id)
if tx is None:
raise Refused(404, f"no transaction {transaction_id}")
state = tx["root"]["status"]
if state == "posted":
return _view(repo, transaction_id)
if state != "approved":
raise RuleBroken(f"the transaction is {state}: only an approved one is posted")
with hub.decide():
services.post_group_now(hub.repository, transaction_id, services.who(caller))
with hub.repository() as repo:
return _view(repo, transaction_id)
@FA.use_case(kind="command", route="POST /asset-transactions/{transaction_id}/reverse", permission=hub.TX_APPROVE,
destructive=True)
def reverse_asset_transaction(caller: dict, transaction_id: Code, reason: Reason) -> dict:
"""Reverse a posted transaction with a reversing transaction dated today (never an edit): every value negated and its
journals the other way round, the original left untouched. Refused for whoever proposed it (second_person), once the year is
closed, or when something later on the asset depends on it (a depreciation run after an acquisition, a revaluation or an
impairment; a later transaction)."""
actor = services.who(caller)
def unit():
with hub.repository() as repo:
from brick_fixed_assets.application.uc_assets import _post_gates
tx = repo.transaction(transaction_id)
if tx is None or tx["root"]["group_id"] != transaction_id:
raise Refused(404, f"no transaction {transaction_id}")
r = tx["root"]
if f"rev:{transaction_id}" in {t["transaction_id"] for t in repo.group(f"rev:{transaction_id}")}:
return f"rev:{transaction_id}"
if r["kind"] == "reversal":
raise RuleBroken("a reversal is not reversed: post the original again")
if r["status"] != "posted":
raise RuleBroken(f"the transaction is {r['status']}: only a posted one is reversed")
if same_person(services.email_of(caller), services.sub_of(caller), r["proposed_by_email"], r["proposed_by_sub"]):
raise RuleBroken("you proposed this (second_person: same email or same sign-in): another approver reverses it")
today = hub.today(repo, True)
legs = repo.group(transaction_id)
for company in sorted({leg["company_code"] for leg in legs}):
services.check_year_open(repo, company, today)
_post_gates(repo, company, today, actor)
for leg in legs:
later = [t for t in repo.tx_rows(leg["asset_id"], statuses=("proposed", "approved", "posted"))
if t["transaction_id"] not in {x["transaction_id"] for x in legs} and
(t["value_date"], t["transaction_id"]) > (leg["value_date"], leg["transaction_id"])]
if later:
raise RuleBroken(f"{later[0]['kind']} {later[0]['transaction_id']} on asset {leg['asset_id']} came later "
"and depends on it: reverse that first")
if leg["kind"] not in ("retirement",) and (leg["kind"] != "transfer" or leg["leg"] == "in"):
deps = [p for p in repo.postings(leg["asset_id"]) if p["kind"] == "run" and
p["period_id"] >= period_of(leg["value_date"]) and p["posted_on"] >= (leg["posted_on"] or leg["value_date"])]
if deps:
raise RuleBroken(f"depreciation has been posted on asset {leg['asset_id']} since (period "
f"{deps[0]['period_id']}): reverse the run first, nothing later may depend on it")
head = services.reverse_group(repo, caller, transaction_id, today, reason)
for leg in repo.group(head):
t = repo.transaction(leg["transaction_id"])
t["root"]["approved_by_email"], t["root"]["approved_by_sub"] = services.email_of(caller), services.sub_of(caller)
t["root"]["approved_on"] = today
repo.save_transaction(t["root"], t["areas"], t["version"])
return head
with hub.decide():
head = hub.retrying(unit, 1)
with hub.repository() as repo:
state = repo.transaction(head)["root"]["status"]
if state != "posted":
services.post_group_now(hub.repository, head, actor)
with hub.repository() as repo:
return _view(repo, head)
@FA.use_case(kind="query", route="GET /asset-transactions/{transaction_id}", permission=hub.REPORT)
def show_asset_transaction(caller: dict, transaction_id: Code) -> dict:
"""One asset transaction with its values per area (cost, accumulated depreciation, reserve, gain or loss, proceeds, the
depreciation catch-up), its journals and ledgers, the evidence kept, who proposed and who approved."""
with hub.repository() as repo:
return _view(repo, transaction_id)
@FA.use_case(kind="query", route="GET /asset-transactions", permission=hub.REPORT)
def list_asset_transactions(caller: dict, asset_id: Optional[Code] = None, status: Optional[Code] = None,
kind: Optional[Code] = None, limit: Limit = 50) -> dict:
"""Asset transactions (acquisitions, settlements, transfers, retirements, revaluations, impairments, takeovers, reversals),
oldest first, filtered by asset, status or kind."""
with hub.repository() as repo:
statuses = (status,) if status else ("proposed", "approved", "posted", "rejected", "reversed")
rows = repo.tx_rows(asset_id, statuses=statuses, kind=kind)[:limit]
return {"transactions": [{"transaction_id": t["transaction_id"], "asset_id": t["asset_id"], "kind": t["kind"],
"status": t["status"], "value_date": t["value_date"].isoformat(),
"company_code": t["company_code"]} for t in rows]}