"""Hardware-asset financial lifecycle — service layer. Aggregation + reporting that turns raw ``AssetItem`` rows into the finance-persona view of the hardware fleet: * current net book value, by category / department / office, with currency-exact breakdowns (mirrors the software ``CostSummaryResponse`` shape so finance reads the two cost axes identically); * disposal P&L (proceeds vs. frozen book value); * the monthly book-value trend (read from materialised snapshots, current month computed live); * per-asset financial timeline (acquisition → depreciation schedule → repair investment → disposal → total cost of ownership); * depreciation policy resolution + CRUD; * the monthly snapshot cron worker. Depreciation arithmetic lives in :mod:`app.domains.asset.depreciation` (pure, no DB). This module is the DB-aware orchestration on top. """ from __future__ import annotations import calendar import logging from datetime import UTC, date, datetime from decimal import Decimal from uuid import UUID from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.fx_rate import get_usd_per_cny from app.domains.asset import depreciation as dep from app.domains.asset.models import ( AssetBookValueSnapshot, AssetDepreciationPolicy, AssetItem, AssetOfficeAddress, ) logger = logging.getLogger(__name__) _ZERO = Decimal("0") _DISPOSED_STATUSES = ("disposed", "disposal_pending") # --------------------------------------------------------------------------- # Policy resolution # --------------------------------------------------------------------------- async def load_policy_map( db: AsyncSession, ) -> tuple[dict[str, dep.DepreciationPolicy], dep.DepreciationPolicy]: """Load active depreciation policies into ``(by_category, default)``. The single ``asset_category IS NULL`` row is the default fallback; if it's missing (fresh DB before the seed migration ran) the module default policy is used. """ rows = ( await db.execute( select(AssetDepreciationPolicy).where( AssetDepreciationPolicy.is_active.is_(True) ) ) ).scalars().all() by_category: dict[str, dep.DepreciationPolicy] = {} default = dep.DepreciationPolicy() for row in rows: policy = dep.DepreciationPolicy( method=row.method, useful_life_months=row.useful_life_months, residual_rate=row.residual_rate, declining_rate=row.declining_rate, ).normalised() if row.asset_category is None: default = policy else: by_category[row.asset_category] = policy return by_category, default def resolve_policy( asset_category: str | None, by_category: dict[str, dep.DepreciationPolicy], default: dep.DepreciationPolicy, ) -> dep.DepreciationPolicy: """Pick the category's policy, else the platform default. The policy is authoritative for method / residual / life — the depreciation is explainable purely from "category X depreciates over N months", with no per-unit surprises. """ if asset_category and asset_category in by_category: return by_category[asset_category] return default # --------------------------------------------------------------------------- # Currency helpers # --------------------------------------------------------------------------- def _add_currency(bucket: dict[str, Decimal], currency: str | None, amount: Decimal) -> None: cur = (currency or "CNY").upper() bucket[cur] = bucket.get(cur, _ZERO) + amount # --------------------------------------------------------------------------- # Current financial stats (live) # --------------------------------------------------------------------------- class _Bucket: __slots__ = ( "key", "label", "asset_count", "undepreciable_count", "original_cost", "accumulated_depreciation", "net_book_value", "repair_total_cost", "by_currency_original", "by_currency_nbv", ) def __init__(self, key: str, label: str | None = None) -> None: self.key = key self.label = label self.asset_count = 0 self.undepreciable_count = 0 self.original_cost = _ZERO self.accumulated_depreciation = _ZERO self.net_book_value = _ZERO self.repair_total_cost = _ZERO self.by_currency_original: dict[str, Decimal] = {} self.by_currency_nbv: dict[str, Decimal] = {} def add(self, *, cost: Decimal, accumulated: Decimal, nbv: Decimal, repair: Decimal, currency: str | None, undated: bool) -> None: self.asset_count += 1 if undated: self.undepreciable_count += 1 self.original_cost += cost self.accumulated_depreciation += accumulated self.net_book_value += nbv self.repair_total_cost += repair _add_currency(self.by_currency_original, currency, cost) _add_currency(self.by_currency_nbv, currency, nbv) def to_dict(self) -> dict: return { "key": self.key, "label": self.label, "asset_count": self.asset_count, "undepreciable_count": self.undepreciable_count, "original_cost": self.original_cost, "accumulated_depreciation": self.accumulated_depreciation, "net_book_value": self.net_book_value, "repair_total_cost": self.repair_total_cost, "by_currency_original": self.by_currency_original, "by_currency_nbv": self.by_currency_nbv, } async def compute_financial_stats( db: AsyncSession, *, scope: str, person_ids: list[UUID] | None = None, department: str | None = None, as_of: date | None = None, ) -> dict: """Live hardware financial summary for a scope. ``scope`` ∈ {mine, department, subordinates, all}. ``person_ids`` narrows to a person set (mine / subordinates); ``department`` narrows to an org subtree. Disposal P&L is reported for the same person/dept scope (all-time). """ from app.core.org_scope import department_membership_subtree_clause from app.identity.person_models import IdentityOrgMembership, IdentityPerson as_of = as_of or datetime.now(UTC).date() by_category, default_policy = await load_policy_map(db) stmt = ( select( AssetItem.id, AssetItem.category, AssetItem.purchase_cost, AssetItem.purchase_date, AssetItem.currency, AssetItem.repair_total_cost, AssetItem.holder_office_id, IdentityOrgMembership.org_node_name.label("department"), ) .outerjoin(IdentityPerson, AssetItem.assigned_to == IdentityPerson.id) .outerjoin( IdentityOrgMembership, and_( IdentityPerson.id == IdentityOrgMembership.person_id, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ), ) .where(AssetItem.deleted_at.is_(None)) .where(AssetItem.status.notin_(_DISPOSED_STATUSES)) ) if scope in ("mine", "subordinates"): # Fail CLOSED: a ``mine``/``subordinates`` scope with no person # set (None or empty) must return an empty book, NOT silently # fall through to a whole-fleet aggregate. All current HTTP # callers pass an explicit list, so this only closes the latent # fail-open if a future caller omits it. if not person_ids: return _empty_stats(scope, as_of) stmt = stmt.where(AssetItem.assigned_to.in_(person_ids)) elif scope == "department" and department: stmt = stmt.where(department_membership_subtree_clause(department)) rows = (await db.execute(stmt)).all() total = _Bucket("__total__") cat_buckets: dict[str, _Bucket] = {} dept_buckets: dict[str, _Bucket] = {} office_ids: set[UUID] = set() office_buckets: dict[str, _Bucket] = {} for row in rows: policy = resolve_policy(row.category, by_category, default_policy) bv = dep.compute_book_value(row.purchase_cost, row.purchase_date, policy, as_of) if bv is None: # No usable cost — count it but contributes nothing financial. total.asset_count += 1 cat_buckets.setdefault(row.category or "uncategorized", _Bucket(row.category or "uncategorized")).asset_count += 1 continue repair = row.repair_total_cost or _ZERO kwargs = dict( cost=bv.original_cost, accumulated=bv.accumulated_depreciation, nbv=bv.net_book_value, repair=repair, currency=row.currency, undated=bv.undated, ) total.add(**kwargs) cat_key = row.category or "uncategorized" cat_buckets.setdefault(cat_key, _Bucket(cat_key)).add(**kwargs) if row.department: dept_buckets.setdefault(row.department, _Bucket(row.department, row.department)).add(**kwargs) if row.holder_office_id: office_ids.add(row.holder_office_id) office_buckets.setdefault(str(row.holder_office_id), _Bucket(str(row.holder_office_id))).add(**kwargs) # Resolve office labels. if office_ids: offices = ( await db.execute( select(AssetOfficeAddress.id, AssetOfficeAddress.city, AssetOfficeAddress.office_name) .where(AssetOfficeAddress.id.in_(office_ids)) ) ).all() label_by_id = {str(o.id): (o.office_name or o.city) for o in offices} for key, bucket in office_buckets.items(): bucket.label = label_by_id.get(key, key) disposal = await _disposal_pnl( db, scope=scope, person_ids=person_ids, department=department, ) distinct_currencies = len([v for v in total.by_currency_original.values() if v != _ZERO]) single_currency = distinct_currencies <= 1 return { "as_of": as_of.isoformat(), "scope": scope, "asset_count": total.asset_count, "undepreciable_count": total.undepreciable_count, # total_* rollups are meaningful only in a single currency; the # by_currency_* maps are always authoritative. "total_original_cost": total.original_cost if single_currency else None, "total_accumulated_depreciation": total.accumulated_depreciation if single_currency else None, "total_net_book_value": total.net_book_value if single_currency else None, "total_repair_cost": total.repair_total_cost if single_currency else None, "by_currency_original": total.by_currency_original, "by_currency_nbv": total.by_currency_nbv, "currency_warning": distinct_currencies > 1, "by_category": _sorted_buckets(cat_buckets), "by_department": _sorted_buckets(dept_buckets), "by_office": _sorted_buckets(office_buckets), "disposal_pnl": disposal, } def _sorted_buckets(buckets: dict[str, _Bucket]) -> list[dict]: return [ b.to_dict() for b in sorted(buckets.values(), key=lambda x: x.net_book_value, reverse=True) ] def _empty_stats(scope: str, as_of: date) -> dict: return { "as_of": as_of.isoformat(), "scope": scope, "asset_count": 0, "undepreciable_count": 0, "total_original_cost": _ZERO, "total_accumulated_depreciation": _ZERO, "total_net_book_value": _ZERO, "total_repair_cost": _ZERO, "by_currency_original": {}, "by_currency_nbv": {}, "currency_warning": False, "by_category": [], "by_department": [], "by_office": [], "disposal_pnl": { "disposed_count": 0, "total_proceeds": _ZERO, "total_book_value_at_disposal": _ZERO, "total_gain_loss": _ZERO, "by_currency_proceeds": {}, "currency_warning": False, }, } async def _disposal_pnl( db: AsyncSession, *, scope: str, person_ids: list[UUID] | None, department: str | None, ) -> dict: """All-time disposal P&L for the scope. Uses the frozen ``disposal_book_value`` + ``disposal_proceeds`` captured at disposal. ``from_user_id`` on the disposal operation preserved the prior owner, but the asset row's ``assigned_to`` is cleared on disposal — so person/department scoping of disposed assets is best-effort via the (now-cleared) assignee and only meaningful for the ``all`` scope. For mine/department scope we still report all-time company disposal P&L attributed to assets currently in scope is not possible post-clear, so non-``all`` scopes report an empty P&L (honest: disposed assets have no live owner to scope by). """ stmt = ( select( AssetItem.disposal_proceeds, AssetItem.disposal_book_value, AssetItem.disposal_currency, AssetItem.currency, ) .where(AssetItem.deleted_at.is_(None)) .where(AssetItem.status == "disposed") ) # Disposal clears assigned_to, so person/department scoping can't # attribute disposed assets reliably. Only the global (all / admin) # scope reports disposal P&L; scoped views show an empty block. if scope != "all": return { "disposed_count": 0, "total_proceeds": _ZERO, "total_book_value_at_disposal": _ZERO, "total_gain_loss": _ZERO, "by_currency_proceeds": {}, "currency_warning": False, } rows = (await db.execute(stmt)).all() count = 0 total_proceeds = _ZERO total_bv = _ZERO by_currency: dict[str, Decimal] = {} by_currency_bv: dict[str, Decimal] = {} for row in rows: if row.disposal_proceeds is None and row.disposal_book_value is None: continue # disposed before this feature existed — no financials count += 1 proceeds = row.disposal_proceeds or _ZERO bv = row.disposal_book_value or _ZERO cur = row.disposal_currency or row.currency total_proceeds += proceeds total_bv += bv _add_currency(by_currency, cur, proceeds) _add_currency(by_currency_bv, cur, bv) # Scalar proceeds / book-value / gain-loss are meaningful only in a # single currency — mirror the headline block (which nulls its # total_* on mixed currency). Consider BOTH proceeds AND book-value # currencies: the old ``currency_warning`` inspected only the proceeds # map, so a set that was single-currency in proceeds but multi- # currency in book value returned a meaningless CNY+USD-blended # gain/loss with ``currency_warning=False`` (e.g. ¥3,000-¥9,000 and # $1,500-$2,000 → total_gain_loss=-6,500 presented as one figure). nonzero_currencies = { c for c, v in by_currency.items() if v != _ZERO } | { c for c, v in by_currency_bv.items() if v != _ZERO } single_currency = len(nonzero_currencies) <= 1 return { "disposed_count": count, "total_proceeds": total_proceeds if single_currency else None, "total_book_value_at_disposal": total_bv if single_currency else None, "total_gain_loss": ( total_proceeds - total_bv if single_currency else None ), "by_currency_proceeds": by_currency, "by_currency_book_value": by_currency_bv, "currency_warning": not single_currency, } # --------------------------------------------------------------------------- # Disposal financials (used by AssetService.dispose_asset) # --------------------------------------------------------------------------- async def compute_disposal_financials( db: AsyncSession, asset: AssetItem, proceeds: Decimal | None, disposed_on: date, proceeds_currency: str | None = None, ) -> dict: """Freeze the disposal financials: net book value AT disposal + proceeds + gain/loss. Returned as a dict for snapshotting into the disposal operation detail and the asset disposal_* columns. """ by_category, default_policy = await load_policy_map(db) policy = resolve_policy(asset.category, by_category, default_policy) bv = dep.compute_book_value(asset.purchase_cost, asset.purchase_date, policy, disposed_on) book_value = bv.net_book_value if bv is not None else None gain_loss: Decimal | None = None disposal_currency = (proceeds_currency or asset.currency or "CNY").upper() # A cross-currency sale has no meaningful native gain/loss until finance # posts an FX journal; never subtract e.g. USD from a CNY book value. if proceeds is not None and book_value is not None and disposal_currency == (asset.currency or "CNY").upper(): gain_loss = proceeds - book_value return { "disposal_proceeds": str(proceeds) if proceeds is not None else None, "disposal_book_value": str(book_value) if book_value is not None else None, "disposal_gain_loss": str(gain_loss) if gain_loss is not None else None, "disposal_currency": disposal_currency, "_book_value": book_value, # Decimal for column write (popped by caller) } # --------------------------------------------------------------------------- # Per-asset financial profile # --------------------------------------------------------------------------- async def get_asset_financial_profile( db: AsyncSession, asset_id: UUID, *, as_of: date | None = None, ) -> dict | None: asset = ( await db.execute( select(AssetItem).where( AssetItem.id == asset_id, AssetItem.deleted_at.is_(None) ) ) ).scalar_one_or_none() if asset is None: return None as_of = as_of or datetime.now(UTC).date() by_category, default_policy = await load_policy_map(db) policy = resolve_policy(asset.category, by_category, default_policy) bv = dep.compute_book_value(asset.purchase_cost, asset.purchase_date, policy, as_of) repair_cost = asset.repair_total_cost or _ZERO gain_loss: Decimal | None = None if asset.disposal_proceeds is not None and asset.disposal_book_value is not None: gain_loss = asset.disposal_proceeds - asset.disposal_book_value # TCO = purchase + repairs − proceeds (single-currency view; assets # are single-currency by row so this is always exact). tco: Decimal | None = None if asset.purchase_cost is not None: tco = asset.purchase_cost + repair_cost - (asset.disposal_proceeds or _ZERO) schedule = dep.build_schedule( asset.purchase_cost, asset.purchase_date, policy, as_of, ) return { "asset_id": asset.id, "asset_tag": asset.asset_tag, "asset_name": asset.asset_name, "category": asset.category, "currency": asset.currency, "purchase_date": asset.purchase_date, "purchase_cost": asset.purchase_cost, "purchase_entity": asset.purchase_entity, "policy_method": policy.method, "useful_life_months": policy.useful_life_months, "residual_rate": policy.residual_rate, "elapsed_months": bv.elapsed_months if bv else 0, "residual_value": bv.residual_value if bv else None, "accumulated_depreciation": bv.accumulated_depreciation if bv else None, "net_book_value": bv.net_book_value if bv else None, "fully_depreciated": bv.fully_depreciated if bv else False, "undepreciable": bv.undated if bv else (asset.purchase_cost is None), "schedule": schedule, "repair_count": asset.repair_count or 0, "repair_total_cost": repair_cost, "status": asset.status, "disposal_type": asset.disposal_type, "disposed_at": asset.disposed_at, "disposal_proceeds": asset.disposal_proceeds, "disposal_book_value": asset.disposal_book_value, "disposal_gain_loss": gain_loss, "total_cost_of_ownership": tco, } # --------------------------------------------------------------------------- # Monthly book-value snapshot (cron) + trend (read) # --------------------------------------------------------------------------- def _month_end(period: str) -> date: year, month = (int(x) for x in period.split("-")) last = calendar.monthrange(year, month)[1] return date(year, month, last) def _prev_period(today: date) -> str: """The just-closed month relative to ``today`` (YYYY-MM).""" if today.month == 1: return f"{today.year - 1}-12" return f"{today.year}-{today.month - 1:02d}" async def snapshot_book_values(db: AsyncSession, period: str) -> dict: """Materialise the fleet net book value for ``period`` (YYYY-MM). Includes assets that existed at month-end and weren't disposed before it. Costs normalised to CNY: CNY as-is, USD via the platform FX rate; other currencies (or USD when FX is unavailable) are excluded from the CNY totals and counted in ``skipped_fx``. Idempotent upsert on ``(period, asset_category)``. """ as_of = _month_end(period) as_of_end = datetime(as_of.year, as_of.month, as_of.day, 23, 59, 59, tzinfo=UTC) by_category, default_policy = await load_policy_map(db) usd_per_cny = await get_usd_per_cny(db) cny_per_usd: Decimal | None = None if usd_per_cny: cny_per_usd = (Decimal("1") / Decimal(str(usd_per_cny))).quantize(Decimal("0.000001")) # "Existed at month-end" is anchored on the acquisition date # (purchase_date) so historical backfill works on bulk-imported # fleets (whose Helix ``created_at`` is the import day, not the buy # date). Fall back to ``created_at`` when purchase_date is unknown. # And the asset must not have been disposed before month-end. from sqlalchemy import and_ as _and from sqlalchemy import or_ as _or rows = ( await db.execute( select( AssetItem.category, AssetItem.purchase_cost, AssetItem.purchase_date, AssetItem.currency, ) .where(AssetItem.deleted_at.is_(None)) .where( _or( AssetItem.purchase_date <= as_of, _and( AssetItem.purchase_date.is_(None), AssetItem.created_at <= as_of_end, ), ) ) .where( (AssetItem.disposed_at.is_(None)) | (AssetItem.disposed_at > as_of_end) ) ) ).all() agg: dict[str | None, dict] = {} skipped_fx = 0 # #1798: track USD assets skipped specifically because the CNY/USD rate # was unavailable, separately from genuinely-unsupported currencies # (EUR/JPY/…). The former is an FX-outage that silently under-reports # the whole CNY total for the month — the handler RAISES on it so the # run lands 'failed' and re-runs once FX is restored. The latter is a # long-standing "we never had a rate for this currency" and must NOT # fail the cron every month. skipped_usd_no_fx = 0 for row in rows: policy = resolve_policy(row.category, by_category, default_policy) bv = dep.compute_book_value(row.purchase_cost, row.purchase_date, policy, as_of) if bv is None: continue cur = (row.currency or "CNY").upper() if cur == "CNY": factor = Decimal("1") elif cur == "USD" and cny_per_usd is not None: factor = cny_per_usd else: skipped_fx += 1 if cur == "USD": # reached only when cny_per_usd is None skipped_usd_no_fx += 1 continue bucket = agg.setdefault(row.category, { "original": _ZERO, "accumulated": _ZERO, "nbv": _ZERO, "count": 0, "undated": 0, }) bucket["original"] += (bv.original_cost * factor) bucket["accumulated"] += (bv.accumulated_depreciation * factor) bucket["nbv"] += (bv.net_book_value * factor) bucket["count"] += 1 if bv.undated: bucket["undated"] += 1 written = 0 for category, vals in agg.items(): existing = ( await db.execute( select(AssetBookValueSnapshot).where( AssetBookValueSnapshot.period == period, ( AssetBookValueSnapshot.asset_category.is_(None) if category is None else AssetBookValueSnapshot.asset_category == category ), ) ) ).scalar_one_or_none() cents = Decimal("0.01") if existing is None: db.add(AssetBookValueSnapshot( period=period, asset_category=category, original_cost_cny=vals["original"].quantize(cents), accumulated_depreciation_cny=vals["accumulated"].quantize(cents), net_book_value_cny=vals["nbv"].quantize(cents), asset_count=vals["count"], undepreciable_count=vals["undated"], )) else: existing.original_cost_cny = vals["original"].quantize(cents) existing.accumulated_depreciation_cny = vals["accumulated"].quantize(cents) existing.net_book_value_cny = vals["nbv"].quantize(cents) existing.asset_count = vals["count"] existing.undepreciable_count = vals["undated"] written += 1 await db.flush() return { "period": period, "categories_written": written, "skipped_fx": skipped_fx, "skipped_usd_no_fx": skipped_usd_no_fx, "fx_available": cny_per_usd is not None, } async def book_value_trend(db: AsyncSession, *, months: int = 24) -> dict: """Read the materialised monthly trend + compute the current month live.""" rows = ( await db.execute( select( AssetBookValueSnapshot.period, AssetBookValueSnapshot.original_cost_cny, AssetBookValueSnapshot.accumulated_depreciation_cny, AssetBookValueSnapshot.net_book_value_cny, AssetBookValueSnapshot.asset_count, AssetBookValueSnapshot.undepreciable_count, ) ) ).all() # Roll up per period (sum across categories). per_period: dict[str, dict] = {} for row in rows: b = per_period.setdefault(row.period, { "original": _ZERO, "accumulated": _ZERO, "nbv": _ZERO, "count": 0, "undated": 0, }) b["original"] += row.original_cost_cny or _ZERO b["accumulated"] += row.accumulated_depreciation_cny or _ZERO b["nbv"] += row.net_book_value_cny or _ZERO b["count"] += row.asset_count or 0 b["undated"] += row.undepreciable_count or 0 periods = sorted(per_period.keys())[-months:] points = [ { "period": p, "original_cost_cny": per_period[p]["original"], "accumulated_depreciation_cny": per_period[p]["accumulated"], "net_book_value_cny": per_period[p]["nbv"], "asset_count": per_period[p]["count"], "undepreciable_count": per_period[p]["undated"], } for p in periods ] # Current month, live (CNY-normalised, same convention as snapshot). today = datetime.now(UTC).date() current_period = f"{today.year}-{today.month:02d}" live = await _live_period_total(db, today) current = { "period": current_period, "original_cost_cny": live["original"], "accumulated_depreciation_cny": live["accumulated"], "net_book_value_cny": live["nbv"], "asset_count": live["count"], "undepreciable_count": live["undated"], } return { "points": points, "current": current, "currency": "CNY", "fx_available": live["fx_available"], } async def _live_period_total(db: AsyncSession, as_of: date) -> dict: by_category, default_policy = await load_policy_map(db) usd_per_cny = await get_usd_per_cny(db) cny_per_usd: Decimal | None = None if usd_per_cny: cny_per_usd = (Decimal("1") / Decimal(str(usd_per_cny))).quantize(Decimal("0.000001")) rows = ( await db.execute( select( AssetItem.category, AssetItem.purchase_cost, AssetItem.purchase_date, AssetItem.currency, ) .where(AssetItem.deleted_at.is_(None)) .where(AssetItem.status.notin_(_DISPOSED_STATUSES)) ) ).all() original = accumulated = nbv = _ZERO count = undated = 0 for row in rows: policy = resolve_policy(row.category, by_category, default_policy) bv = dep.compute_book_value(row.purchase_cost, row.purchase_date, policy, as_of) if bv is None: continue cur = (row.currency or "CNY").upper() if cur == "CNY": factor = Decimal("1") elif cur == "USD" and cny_per_usd is not None: factor = cny_per_usd else: continue original += bv.original_cost * factor accumulated += bv.accumulated_depreciation * factor nbv += bv.net_book_value * factor count += 1 if bv.undated: undated += 1 cents = Decimal("0.01") return { "original": original.quantize(cents), "accumulated": accumulated.quantize(cents), "nbv": nbv.quantize(cents), "count": count, "undated": undated, "fx_available": cny_per_usd is not None, } # --------------------------------------------------------------------------- # Depreciation policy CRUD # --------------------------------------------------------------------------- async def list_policies(db: AsyncSession) -> list[AssetDepreciationPolicy]: return list(( await db.execute( select(AssetDepreciationPolicy).order_by( AssetDepreciationPolicy.asset_category.is_(None).desc(), AssetDepreciationPolicy.asset_category, ) ) ).scalars().all()) async def upsert_policy( db: AsyncSession, *, asset_category: str | None, method: str, useful_life_months: int, residual_rate: Decimal, declining_rate: Decimal | None, is_active: bool, notes: str | None, ) -> AssetDepreciationPolicy: """Create or update the policy for a category (NULL = default). Single-default invariant (one NULL row) enforced here, since MySQL unique indexes permit multiple NULLs. """ if asset_category is None: existing = ( await db.execute( select(AssetDepreciationPolicy).where( AssetDepreciationPolicy.asset_category.is_(None) ) ) ).scalar_one_or_none() else: existing = ( await db.execute( select(AssetDepreciationPolicy).where( AssetDepreciationPolicy.asset_category == asset_category ) ) ).scalar_one_or_none() if existing is None: existing = AssetDepreciationPolicy(asset_category=asset_category) db.add(existing) existing.method = method existing.useful_life_months = useful_life_months existing.residual_rate = residual_rate existing.declining_rate = declining_rate existing.is_active = is_active existing.notes = notes await db.flush() return existing async def delete_policy(db: AsyncSession, policy_id: UUID) -> bool: """Delete a per-category policy. The default (NULL) row cannot be deleted (it's the fallback) — only deactivated/edited.""" policy = await db.get(AssetDepreciationPolicy, policy_id) if policy is None: return False if policy.asset_category is None: raise ValueError("cannot_delete_default_policy") await db.delete(policy) await db.flush() return True