dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/infrastructure/tables.py
"""The tables of the fixed-assets brick (module `fixed-assets`, prefix `fa_`): one column model, shared by the tests
(SQLite) and schema.sql / the migrations (DataK3 over the pg wire; tests/test_schema.py keeps them equal).
Every table has a natural primary key (never SERIAL); a part's key starts with its root's, so the repository can find
stale rows at any depth. Money is NUMERIC(18,2), units and rates NUMERIC(19,4). Configuration is DATED rows (a row is
never edited: a change is a later row). An asset's values are NEVER stored: they are the SUM of its transactions
(fa_transaction_area) and posted depreciation (fa_dep_posting).
"""
from erp_platform.repository import Aggregate, Part
from sqlalchemy import TIMESTAMP, BigInteger, Boolean, Column, Date, Integer, MetaData, Numeric, String, Table
md = MetaData()
S, D, INT, BI, B, TS = String, Date, Integer, BigInteger, Boolean, TIMESTAMP
N2, N4 = Numeric(18, 2), Numeric(19, 4)
def table(name: str, cols: str, *, audit: bool = True) -> Table:
"""`cols`: space-separated `name:Type` — a trailing `*` marks a primary key column. Types: S D I BI B TS N2 N4."""
kinds = {"S": S, "D": D, "I": INT, "BI": BI, "B": B, "TS": TS, "N2": N2, "N4": N4}
columns = []
for spec in cols.split():
col, _, kind = spec.partition(":")
pk = kind.endswith("*")
columns.append(Column(col, kinds[kind.rstrip("*")], primary_key=pk))
if audit:
columns += [Column("created_at", TS), Column("updated_at", TS)]
return Table(name, md, *columns)
# ── configuration: dated rows, never edited ───────────────────────────────────────────────────
fa_settings = table("fa_settings", "setting_key:S* valid_from:D* value:S set_by:S")
fa_area = table("fa_area", "area_code:S* valid_from:D* name:S purpose:S ledger_id:S posting:S value_model:S "
"values_from:S currency:S company_code:S set_by:S")
fa_dep_key = table("fa_dep_key", "key_code:S* valid_from:D* method:S rate_pct:N4 life_periods:I base:S acq_conv:S "
"ret_conv:S stop_at:S set_by:S")
fa_class = table("fa_class", "class_code:S* valid_from:D* name:S is_auc:B is_low_value:B default_ca_pool:S "
"number_format:S set_by:S")
fa_class_area = table("fa_class_area", "class_code:S* area_code:S* valid_from:D* key_code:S life_periods:I set_by:S")
fa_account_map = table("fa_account_map", "class_code:S* area_code:S* valid_from:D* cost_account_id:BI "
"accumulated_depreciation_account_id:BI depreciation_expense_account_id:BI gain_account_id:BI "
"loss_account_id:BI revaluation_reserve_account_id:BI impairment_account_id:BI "
"impairment_reversal_account_id:BI set_by:S")
fa_ca_rate = table("fa_ca_rate", "rate_key:S* valid_from:D* value:N4 legislation_ref:S set_by:S")
fa_number_series = table("fa_number_series", "series_id:S* year:I* format:S high_water_mark:I")
fa_number_claim = table("fa_number_claim", "asset_no:S* series_id:S year:I seq:I request_id:S")
fa_clock = table("fa_clock", "clock_key:S* last_today:D")
# ── the register ──────────────────────────────────────────────────────────────────────────────
fa_asset_request = table("fa_asset_request", "request_id:S* asset_id:S company_code:S")
fa_asset = table("fa_asset", "asset_id:S* company_code:S asset_no:S sub_no:I class_code:S description:S serial_no:S "
"inventory_no:S bp_id:BI capitalised_on:D status:S last_counted_on:D quantity:N4 "
"request_id:S is_auc:B is_car:B new_and_unused:B ca_pool:S deactivated_reason:S "
"created_by:S version:I")
fa_asset_master = table("fa_asset_master", "asset_id:S* valid_from:D* cost_centre:S location_id:S custodian_email:S "
"set_by_email:S")
fa_asset_area = table("fa_asset_area", "asset_id:S* area_code:S* valid_from:D* key_code:S life_periods:I "
"total_units:N4 scrap_value:N2 dep_start:S")
ASSET = Aggregate(Part(fa_asset, ("asset_id",)), Part(fa_asset_master, ("asset_id", "valid_from")),
Part(fa_asset_area, ("asset_id", "area_code", "valid_from")), version="version")
fa_transaction = table("fa_transaction", "transaction_id:S* asset_id:S company_code:S group_id:S leg:S kind:S value_date:D "
"posting_period_id:S fiscal_year:I status:S proposed_by_email:S proposed_by_sub:S "
"approved_by_email:S approved_by_sub:S approved_on:D posted_on:D rejected_reason:S "
"reverses_transaction_id:S evidence_key:S source_ref:S params:S asset_version:I version:I")
fa_transaction_area = table("fa_transaction_area", "transaction_id:S* area_code:S* cost:N2 accumulated_depreciation:N2 "
"depreciation_in_year:N2 revaluation_reserve:N2 gain_loss:N2 proceeds:N2 catch_up:N2 "
"journal_id:BI ledger_id:S")
TRANSACTION = Aggregate(Part(fa_transaction, ("transaction_id",)), Part(fa_transaction_area, ("transaction_id", "area_code")),
version="version")
fa_acquisition_claim = table("fa_acquisition_claim", "acquisition_ref:S* bill_id:S line_no:I asset_id:S amount:N2 "
"transaction_id:S taken_at:TS")
fa_auc_settlement = table("fa_auc_settlement", "auc_asset_id:S* settlement_date:D* seq:I* status:S rule:S "
"settled_by_email:S settled_on:D reversed_reason:S")
fa_settlement_claim = table("fa_settlement_claim", "auc_asset_id:S* line_item_transaction_id:S* receiver_asset_id:S* "
"settlement_seq:I settlement_date:D percent:N4 amount:N2")
fa_settlement_line_seq = table("fa_settlement_line_seq", "auc_asset_id:S* line_item_transaction_id:S* n:I* "
"settlement_seq:I")
fa_usage = table("fa_usage", "asset_id:S* period_id:S* seq:I* units:N4 recorded_by_email:S reason:S")
# ── depreciation ──────────────────────────────────────────────────────────────────────────────
fa_dep_run = table("fa_dep_run", "company_code:S* period_id:S* run_no:I* run_id:S status:S started_by_email:S "
"started_by_sub:S approved_by_email:S approved_by_sub:S approved_on:D posted_on:D posted_by:S "
"reverses_run_id:S reversed_by_email:S reason:S totals:S gates:S version:I")
fa_dep_run_line = table("fa_dep_run_line", "company_code:S* period_id:S* run_no:I* asset_id:S* area_code:S* class_code:S "
"cost_centre:S planned_to_date:N2 posted_before:N2 amount:N2")
DEP_RUN = Aggregate(Part(fa_dep_run, ("company_code", "period_id", "run_no")),
Part(fa_dep_run_line, ("company_code", "period_id", "run_no", "asset_id", "area_code")),
version="version")
fa_dep_posting = table("fa_dep_posting", "asset_id:S* area_code:S* period_id:S* run_id:S* amount:N2 cost_centre:S "
"class_code:S journal_id:BI kind:S posted_on:D company_code:S")
fa_dep_post_claim = table("fa_dep_post_claim", "company_code:S* period_id:S* area_code:S* post_seq:I* run_id:S")
fa_year = table("fa_year", "company_code:S* fiscal_year:I* status:S closed_by_email:S closed_by_sub:S closed_on:D")
# ── capital allowances (the tax area) ─────────────────────────────────────────────────────────
fa_ca_computation = table("fa_ca_computation", "company_code:S* period_start:D* period_end:D* status:S months:I "
"aia_limit_allocated:N2 opening:S result:S prepared_by_email:S prepared_by_sub:S "
"approved_by_email:S approved_by_sub:S return_reference:S returned_reason:S version:I")
fa_ca_claim = table("fa_ca_claim", "company_code:S* period_start:D* period_end:D* asset_id:S* claim:S pool:S "
"amount:N2 disposal_value:N2")
CA = Aggregate(Part(fa_ca_computation, ("company_code", "period_start", "period_end")),
Part(fa_ca_claim, ("company_code", "period_start", "period_end", "asset_id")), version="version")
# ── counts ────────────────────────────────────────────────────────────────────────────────────
fa_count = table("fa_count", "location_id:S* count_date:D* custodian_email:S status:S opened_by_email:S version:I")
fa_count_line = table("fa_count_line", "location_id:S* count_date:D* line_no:I* asset_id:S tag:S description:S "
"serial_no:S result:S counted_by_email:S counted_on:D decision:S decision_ref:S note:S moved_in:B")
COUNT = Aggregate(Part(fa_count, ("location_id", "count_date")),
Part(fa_count_line, ("location_id", "count_date", "line_no")), version="version")
# ── AI steps ──────────────────────────────────────────────────────────────────────────────────
fa_ai_finding = table("fa_ai_finding", "subject_ref:S* step:S* run_no:I* finding_no:I* model:S inputs_digest:S field:S "
"value:S evidence:S decision:S decided_by_email:S decided_at:TS")
fa_task = table("fa_task", "task_id:S* step:S subject_ref:S status:S capability:S input_json:S plan:S reason:S "
"destructive:B summary:S output_json:S created_by:S resolved_by:S note:S")
# ── migration (Odoo) ──────────────────────────────────────────────────────────────────────────
fa_migration_run = table("fa_migration_run", "run_id:S* odoo_db:S cutover_date:D status:S doors:S checks:S "
"checksums:S loaded_by:S signed_by_email:S signed_by_sub:S signed_on:D version:I")
fa_migration_map = table("fa_migration_map", "odoo_db:S* odoo_model:S* odoo_id:S* natural_key:S run_id:S")
MIGRATION_RUN = Aggregate(Part(fa_migration_run, ("run_id",)), version="version")
# ═══ the repository: reading and writing the rows, one per connection (what `use_cases.repository()` yields) ═══
# The caller owns the transaction. Keyed writes go through erp_platform.repository (`upsert`, `Aggregate`) — never a
# hand-written upsert, never `session.merge()`. A plain INSERT is used where a second writer MUST collide on the key (a
# claim: an asset number, a posting sequence, a settlement line), because DataK3 rejects a second committed insert of a
# primary key (23505) and gives no read-your-writes inside an open transaction. (It lives in this module because the
# scenario runner finds a brick's `Repository` here.)
import json # noqa: E402
import re # noqa: E402
import sys # noqa: E402
from datetime import date, datetime, timezone # noqa: E402
from decimal import Decimal # noqa: E402
from typing import Optional # noqa: E402
from erp_platform.repository import upsert # noqa: E402
from sqlalchemy import and_, func, insert, select # noqa: E402
from brick_fixed_assets.domain.errors import RuleBroken # noqa: E402
from brick_fixed_assets.infrastructure import business_time # noqa: E402
T = sys.modules[__name__]
EPOCH = date(1900, 1, 1)
FOREVER = date(2199, 12, 31)
CONTROL_SETTINGS = {
"capitalisation_threshold": ("decimal", "500.00"),
"depreciation_needs_approval": ("bool", True),
"asset_number_format": ("str", "{yyyy}-{class}-{seq:4}"),
"transfer_variant_default": ("str", "gross"),
"aia_group_allocation": ("str", "all to the first company"),
"intercompany_clearing_account_id": ("int", None),
"asset_sale_clearing_account_id": ("int", None),
"asset_clearing_account_id": ("int", None), # purchasing's clearing account: a manual acquisition never offsets it
}
DAY_TO_DAY_SETTINGS = {"ai.ai_suggest_asset_class.mode": ("str", "suggest")}
ALL_SETTINGS = {**CONTROL_SETTINGS, **DAY_TO_DAY_SETTINGS}
def _now() -> datetime:
return datetime.now(timezone.utc)
def d(row) -> Optional[dict]:
return dict(row._mapping) if row is not None else None
class Repository:
def __init__(self, conn):
self.conn = conn
self._config: dict = {} # (table, keys) -> the dated rows this transaction read: a request reads each configuration key once
self._snapshots: dict = {} # (aggregate, root key) -> what its load read: the save hands it back so only changed rows are written
def load(self, agg, **root_key):
"""agg.load(), keeping the snapshot for this transaction (see spend())."""
got = agg.load(self.conn, root_key)
if got is not None:
self._snapshots[(id(agg), tuple(root_key.values()))] = got.snapshot
return got
def spend(self, agg, *root_key):
"""The snapshot a load of this root kept, once: after the save the rows are no longer what it read."""
return self._snapshots.pop((id(agg), tuple(root_key)), None)
# ── the business date ─────────────────────────────────────────────────────────────────────
def today(self, record: bool = False) -> date:
"""Business today (business_timezone), never below the last today this brick recorded. A command passes
record=True so the floor moves up with it."""
floor = self.conn.execute(select(T.fa_clock.c.last_today).where(T.fa_clock.c.clock_key == "today")).scalar()
day = business_time.today(self.conn, floor)
if record and (floor is None or day > floor):
upsert(self.conn, T.fa_clock, [{"clock_key": "today", "last_today": day}], ("clock_key",))
return day
# ── dated configuration ───────────────────────────────────────────────────────────────────
def rows(self, table, **keys) -> list:
memo = (table.name, tuple(sorted(keys.items())))
if memo not in self._config:
cond = [table.c[k] == v for k, v in keys.items()]
q = select(table).where(and_(*cond)) if cond else select(table)
self._config[memo] = [dict(r._mapping) for r in self.conn.execute(q.order_by(table.c.valid_from))]
return [dict(r) for r in self._config[memo]]
def config_written(self) -> None:
"""Configuration rows were written around put_dated (the Odoo load): drop what this transaction remembered."""
self._config.clear()
def row_on(self, table, on: date, **keys) -> Optional[dict]:
found = None
for r in self.rows(table, **keys):
if r["valid_from"] <= on:
found = r
return found
def put_dated(self, table, keys: dict, valid_from: date, values: dict, today: date, by: str) -> tuple:
"""Append one dated row. A row is never edited: a change is a later row. The first row of a key may start
today; every later one starts AFTER today. The same row again is a no-op. Returns (row, created)."""
existing = self.rows(table, **keys)
for r in existing:
if r["valid_from"] == valid_from:
same = all(_same(r.get(k), v) for k, v in values.items())
if same:
return r, False
raise RuleBroken("a dated row is never edited: give the change a later valid_from")
if existing:
if valid_from <= today:
raise RuleBroken(f"a change starts after today ({today.isoformat()}), not on {valid_from.isoformat()}: "
"pick a later date")
elif valid_from < today:
raise RuleBroken(f"the first row of this may start today ({today.isoformat()}) at the earliest")
row = {**keys, "valid_from": valid_from, **values, "set_by": by}
upsert(self.conn, table, [row], tuple(table.primary_key.columns.keys()))
self.config_written()
return row, True
# settings
def setting(self, key: str, on: Optional[date] = None):
on = on or self.today()
kind, default = ALL_SETTINGS[key]
row = self.row_on(T.fa_settings, on, setting_key=key)
if row is None or row["value"] is None:
return default
return _decode(kind, row["value"])
def settings_view(self, on: Optional[date] = None, controls: bool = True) -> dict:
on = on or self.today()
out = {k: self.setting(k, on) for k in (ALL_SETTINGS if controls else DAY_TO_DAY_SETTINGS)}
return {k: (str(v) if isinstance(v, Decimal) else v) for k, v in out.items()}
# areas, keys, classes
def areas(self, on: date) -> list:
codes = sorted({r["area_code"] for r in self.rows(T.fa_area)})
return [a for a in (self.row_on(T.fa_area, on, area_code=c) for c in codes) if a]
def area(self, code: str, on: date) -> Optional[dict]:
return self.row_on(T.fa_area, on, area_code=code)
def posting_areas(self, on: date, company_code: Optional[str] = None) -> list:
"""The areas that post. An area names ONE ledger, which belongs to ONE company: with `company_code` only that
company's areas (an area with no company recorded serves every company)."""
return [a for a in self.areas(on) if a["posting"] == "per_run" and a["ledger_id"] is not None
and (company_code is None or a.get("company_code") in (None, company_code))]
def key(self, code: str, on: date) -> Optional[dict]:
return self.row_on(T.fa_dep_key, on, key_code=code)
def rows_of_key(self, code: str) -> list:
return self.rows(T.fa_dep_key, key_code=code)
def klass(self, code: str, on: date) -> Optional[dict]:
return self.row_on(T.fa_class, on, class_code=code)
def classes(self, on: date) -> list:
codes = sorted({r["class_code"] for r in self.rows(T.fa_class)})
return [c for c in (self.klass(x, on) for x in codes) if c]
def class_areas(self, class_code: str, on: date) -> list:
codes = sorted({r["area_code"] for r in self.rows(T.fa_class_area, class_code=class_code)})
return [r for r in (self.row_on(T.fa_class_area, on, class_code=class_code, area_code=a) for a in codes) if r]
def account_map(self, class_code: str, area_code: str, on: date) -> Optional[dict]:
return self.row_on(T.fa_account_map, on, class_code=class_code, area_code=area_code)
def ca_rates(self, on: date) -> dict:
keys = sorted({r["rate_key"] for r in self.rows(T.fa_ca_rate)})
out = {}
for k in keys:
r = self.row_on(T.fa_ca_rate, on, rate_key=k)
if r:
out[k] = r["value"]
return out
# ── number series ─────────────────────────────────────────────────────────────────────────
def claim_number(self, series_id: str, year: int, fmt: str, request_id: str, seen_max: Optional[int] = None) -> str:
"""The next asset number of a class's series for a year: one more than the high-water mark and than anything
claimed, claimed by a plain INSERT keyed on the RENDERED number — a concurrent writer that took the same number
collides (23505) and retries on the next one. A gap is never filled, a number never reused."""
row = self.conn.execute(select(T.fa_number_series.c.high_water_mark).where(and_(
T.fa_number_series.c.series_id == series_id, T.fa_number_series.c.year == year))).scalar() or 0
top = self.conn.execute(select(func.max(T.fa_number_claim.c.seq)).where(and_(
T.fa_number_claim.c.series_id == series_id, T.fa_number_claim.c.year == year))).scalar() or 0
n = (seen_max if seen_max is not None else max(row, top)) + 1
number = render_number(fmt, year, series_id, n)
self.conn.execute(insert(T.fa_number_claim).values(asset_no=number, series_id=series_id, year=year, seq=n,
request_id=request_id, created_at=_now(), updated_at=_now()))
upsert(self.conn, T.fa_number_series, [{"series_id": series_id, "year": year, "format": fmt, "high_water_mark": n}],
("series_id", "year"))
return number
def claim_slot(self, kind: str, key: str, request_id: str = "") -> None:
"""Claim a number that is allocated per parent (a sub-asset number, a depreciation run number) by a plain INSERT
keyed on the namespaced rendered key: a concurrent writer that picked the same number collides (23505) and the
unit of work retries on the next free one."""
self.conn.execute(insert(T.fa_number_claim).values(asset_no=f"{kind}:{key}", series_id=kind, year=0, seq=0,
request_id=request_id, created_at=_now(), updated_at=_now()))
@staticmethod
def first_free(used) -> int:
"""The next number above the highest one held: a series never fills a gap below its high-water mark, and a held
number is never reused (rows are kept, only deactivated)."""
return max(used, default=0) + 1 # erp-lint: allow L6 high-water mark over held numbers; every pick is then claimed by a plain INSERT (claim_slot) so a collision retries
def claim_name(self, asset_no: str, series_id: str, request_id: str, class_code: Optional[str] = None,
fmt: Optional[str] = None) -> bool:
"""A number that came from outside (a migrated asset keeps its Odoo name) is claimed too, so a series can never
hand it out again. When the name is in the class's own format its year and sequence are recorded in the class's
series, so claim_number's high-water mark steps over it. False when the name is already claimed (checked first:
a failed INSERT would abort the transaction)."""
if self.conn.execute(select(T.fa_number_claim.c.asset_no).where(T.fa_number_claim.c.asset_no == asset_no)).first():
return False
year, seq = 0, 0
if class_code and fmt:
m = re.fullmatch(_number_pattern(fmt, class_code), asset_no)
if m:
year = int(m.groupdict().get("y") or 0)
seq, series_id = int(m.group("n")), class_code
self.conn.execute(insert(T.fa_number_claim).values(asset_no=asset_no, series_id=series_id, year=year, seq=seq,
request_id=request_id, created_at=_now(), updated_at=_now()))
return True
def number_gaps(self, series_id: str, year: int) -> list:
seqs = sorted(r[0] for r in self.conn.execute(select(T.fa_number_claim.c.seq).where(and_(
T.fa_number_claim.c.series_id == series_id, T.fa_number_claim.c.year == year))))
have = set(seqs)
return [n for n in range(1, (seqs[-1] if seqs else 0) + 1) if n not in have]
# ── assets ────────────────────────────────────────────────────────────────────────────────
def asset(self, asset_id: str) -> Optional["Loaded"]:
got = self.load(T.ASSET, asset_id=asset_id)
if got is None:
return None
root = got["fa_asset"][0]
return Loaded(root, sorted(got["fa_asset_master"], key=lambda r: r["valid_from"]),
sorted(got["fa_asset_area"], key=lambda r: (r["area_code"], r["valid_from"])), root["version"])
def assets(self, asset_ids) -> dict:
"""{asset_id: Loaded} for a set of assets: one query per table per 100 assets, in one round trip (a loop of asset() is three queries each)."""
out = {}
for (aid,), got in T.ASSET.load_many(self.conn, [{"asset_id": i} for i in dict.fromkeys(asset_ids)]).items():
if got is None:
continue
self._snapshots[(id(T.ASSET), (aid,))] = got.snapshot
root = got["fa_asset"][0]
out[aid] = Loaded(root, sorted(got["fa_asset_master"], key=lambda r: r["valid_from"]),
sorted(got["fa_asset_area"], key=lambda r: (r["area_code"], r["valid_from"])), root["version"])
return out
def save_asset(self, a: "Loaded") -> int:
return T.ASSET.save(self.conn, {"fa_asset": [a.root], "fa_asset_master": a.master, "fa_asset_area": a.areas},
expected_version=a.version, snapshot=self.spend(T.ASSET, a.root["asset_id"]))
def asset_by_request(self, request_id: str) -> Optional[str]:
return self.conn.execute(select(T.fa_asset_request.c.asset_id).where(
T.fa_asset_request.c.request_id == request_id)).scalar()
def asset_roots(self, company_code: Optional[str] = None, class_code: Optional[str] = None,
statuses: Optional[list] = None) -> list:
q = select(T.fa_asset)
if company_code:
q = q.where(T.fa_asset.c.company_code == company_code)
if class_code:
q = q.where(T.fa_asset.c.class_code == class_code)
if statuses:
q = q.where(T.fa_asset.c.status.in_(statuses))
return [dict(r._mapping) for r in self.conn.execute(q.order_by(T.fa_asset.c.asset_id))]
def masters_on(self, on: date) -> dict:
"""{asset_id: master row in force on `on`} for every asset."""
out: dict = {}
for r in self.conn.execute(select(T.fa_asset_master).order_by(T.fa_asset_master.c.valid_from)):
m = dict(r._mapping)
if m["valid_from"] <= on:
out[m["asset_id"]] = m
return out
def area_rows_on(self, on: date, asset_ids: Optional[list] = None) -> dict:
"""{(asset_id, area_code): area row in force on `on`}."""
q = select(T.fa_asset_area).order_by(T.fa_asset_area.c.valid_from)
if asset_ids is not None:
q = q.where(T.fa_asset_area.c.asset_id.in_(asset_ids))
out: dict = {}
for r in self.conn.execute(q):
m = dict(r._mapping)
if m["valid_from"] <= on:
out[(m["asset_id"], m["area_code"])] = m
return out
# ── transactions and values ───────────────────────────────────────────────────────────────
def transaction(self, transaction_id: str) -> Optional[dict]:
got = self.load(T.TRANSACTION, transaction_id=transaction_id)
if got is None:
return None
root = got["fa_transaction"][0]
return {"root": root, "areas": {r["area_code"]: r for r in got["fa_transaction_area"]}, "version": root["version"]}
def transactions(self, transaction_ids) -> dict:
"""{transaction_id: transaction()} for a set: one query per table per 100 ids, in one round trip (a loop of transaction() is two each)."""
out = {}
for (tid,), got in T.TRANSACTION.load_many(self.conn, [{"transaction_id": i} for i in dict.fromkeys(transaction_ids)]).items():
if got is None:
continue
self._snapshots[(id(T.TRANSACTION), (tid,))] = got.snapshot
root = got["fa_transaction"][0]
out[tid] = {"root": root, "areas": {r["area_code"]: r for r in got["fa_transaction_area"]}, "version": root["version"]}
return out
def save_transaction(self, root: dict, areas: dict, expected_version: int) -> int:
return T.TRANSACTION.save(self.conn, {"fa_transaction": [root], "fa_transaction_area": list(areas.values())},
expected_version=expected_version, snapshot=self.spend(T.TRANSACTION, root["transaction_id"]))
def group(self, group_id: str) -> list:
"""Every leg of a transaction group, head first."""
rows = [dict(r._mapping) for r in self.conn.execute(
select(T.fa_transaction).where(T.fa_transaction.c.group_id == group_id).order_by(T.fa_transaction.c.transaction_id))]
return sorted(rows, key=lambda r: (r["transaction_id"] != group_id, r["transaction_id"]))
def tx_rows(self, asset_id: Optional[str] = None, statuses: tuple = ("posted", "reversed"),
company_code: Optional[str] = None, kind: Optional[str] = None) -> list:
q = select(T.fa_transaction).where(T.fa_transaction.c.status.in_(statuses))
if asset_id:
q = q.where(T.fa_transaction.c.asset_id == asset_id)
if company_code:
q = q.where(T.fa_transaction.c.company_code == company_code)
if kind:
q = q.where(T.fa_transaction.c.kind == kind)
return [dict(r._mapping) for r in self.conn.execute(q.order_by(T.fa_transaction.c.value_date,
T.fa_transaction.c.transaction_id))]
def prefetch_plans(self, asset_ids) -> None:
"""Read what build_plan asks of EVERY asset in two statements: the posted joined rows (area_values_rows) and the usage readings. Until
clear_plans(), those two calls answer from here. For a loop over many assets that writes nothing in between (a depreciation run)."""
ids = sorted(set(asset_ids))
t, a = T.fa_transaction, T.fa_transaction_area
rows: dict = {i: [] for i in ids}
use: dict = {i: {} for i in ids}
if ids:
q = select(t, a.c.area_code, a.c.cost, a.c.accumulated_depreciation, a.c.depreciation_in_year, a.c.revaluation_reserve,
a.c.gain_loss, a.c.proceeds, a.c.catch_up, a.c.journal_id, a.c.ledger_id).join(
a, a.c.transaction_id == t.c.transaction_id).where(t.c.status.in_(("posted",)), t.c.asset_id.in_(ids))
for r in self.conn.execute(q.order_by(t.c.value_date, t.c.transaction_id)):
rows[r._mapping["asset_id"]].append(dict(r._mapping))
for r in self.conn.execute(select(T.fa_usage).where(T.fa_usage.c.asset_id.in_(ids)).order_by(T.fa_usage.c.period_id, T.fa_usage.c.seq)):
use[r.asset_id][r.period_id] = r.units
self._plan_rows, self._plan_usage = rows, use
def clear_plans(self) -> None:
self._plan_rows = self._plan_usage = None
_plan_rows: Optional[dict] = None
_plan_usage: Optional[dict] = None
def area_values_rows(self, asset_id: Optional[str] = None, area_code: Optional[str] = None,
company_code: Optional[str] = None, statuses: tuple = ("posted", "reversed")) -> list:
"""Joined rows (transaction + its area values) of the transactions that count toward values."""
if self._plan_rows is not None and asset_id in self._plan_rows and company_code is None and tuple(statuses) == ("posted",):
return [r for r in self._plan_rows[asset_id] if area_code is None or r["area_code"] == area_code]
t, a = T.fa_transaction, T.fa_transaction_area
q = select(t, a.c.area_code, a.c.cost, a.c.accumulated_depreciation, a.c.depreciation_in_year,
a.c.revaluation_reserve, a.c.gain_loss, a.c.proceeds, a.c.catch_up, a.c.journal_id,
a.c.ledger_id).join(a, a.c.transaction_id == t.c.transaction_id).where(t.c.status.in_(statuses))
if asset_id:
q = q.where(t.c.asset_id == asset_id)
if area_code:
q = q.where(a.c.area_code == area_code)
if company_code:
q = q.where(t.c.company_code == company_code)
return [dict(r._mapping) for r in self.conn.execute(q.order_by(t.c.value_date, t.c.transaction_id))]
def postings(self, asset_id: Optional[str] = None, area_code: Optional[str] = None,
company_code: Optional[str] = None) -> list:
q = select(T.fa_dep_posting)
if asset_id:
q = q.where(T.fa_dep_posting.c.asset_id == asset_id)
if area_code:
q = q.where(T.fa_dep_posting.c.area_code == area_code)
if company_code:
q = q.where(T.fa_dep_posting.c.company_code == company_code)
return [dict(r._mapping) for r in self.conn.execute(q.order_by(T.fa_dep_posting.c.period_id))]
def usage(self, asset_id: str) -> dict:
"""{period: units}: the LAST reading of each period is the effective one (a correction is a later row)."""
if self._plan_usage is not None and asset_id in self._plan_usage:
return dict(self._plan_usage[asset_id])
out: dict = {}
for r in self.conn.execute(select(T.fa_usage).where(T.fa_usage.c.asset_id == asset_id).order_by(
T.fa_usage.c.period_id, T.fa_usage.c.seq)):
out[r.period_id] = r.units
return out
# ── year status ───────────────────────────────────────────────────────────────────────────
def year_closed(self, company_code: str, fiscal_year: int) -> bool:
s = self.conn.execute(select(T.fa_year.c.status).where(and_(T.fa_year.c.company_code == company_code,
T.fa_year.c.fiscal_year == fiscal_year))).scalar()
return s == "closed"
# ── ai ────────────────────────────────────────────────────────────────────────────────────
def findings(self, subject_ref: str) -> list:
q = select(T.fa_ai_finding).where(T.fa_ai_finding.c.subject_ref == subject_ref).order_by(
T.fa_ai_finding.c.run_no, T.fa_ai_finding.c.finding_no)
return [dict(r._mapping) for r in self.conn.execute(q)]
# ── migration map ─────────────────────────────────────────────────────────────────────────
def map_get(self, db: str, model: str, odoo_id) -> Optional[str]:
return self.conn.execute(select(T.fa_migration_map.c.natural_key).where(and_(
T.fa_migration_map.c.odoo_db == db, T.fa_migration_map.c.odoo_model == model,
T.fa_migration_map.c.odoo_id == str(odoo_id)))).scalar()
def map_put(self, db: str, model: str, odoo_id, natural_key: str, run_id: str) -> None:
upsert(self.conn, T.fa_migration_map, [{"odoo_db": db, "odoo_model": model, "odoo_id": str(odoo_id),
"natural_key": natural_key, "run_id": run_id}],
("odoo_db", "odoo_model", "odoo_id"))
class Loaded:
"""One asset as loaded: its root, its master rows and its area rows over time, and the version it was loaded at."""
def __init__(self, root: dict, master: list, areas: list, version: int):
self.root, self.master, self.areas, self.version = root, master, areas, version
@property
def asset_id(self) -> str:
return self.root["asset_id"]
def master_on(self, on: date) -> Optional[dict]:
found = None
for m in self.master:
if m["valid_from"] <= on:
found = m
return found
def area_on(self, area_code: str, on: date) -> Optional[dict]:
found = None
for r in self.areas:
if r["area_code"] == area_code and r["valid_from"] <= on:
found = r
return found
def area_codes(self) -> list:
return sorted({r["area_code"] for r in self.areas})
def _number_pattern(fmt: str, class_code: str) -> str:
"""The regex that reads back what render_number wrote: groups y (year, if the format has one) and n (seq)."""
out = []
for part in re.split(r"(\{yyyy\}|\{class\}|\{seq(?::\d+)?\})", fmt):
if part == "{yyyy}":
out.append(r"(?P<y>\d{4})")
elif part == "{class}":
out.append(re.escape(class_code))
elif part.startswith("{seq"):
out.append(r"(?P<n>\d+)")
else:
out.append(re.escape(part))
return "".join(out)
def render_number(fmt: str, year: int, class_code: str, n: int) -> str:
"""'{yyyy}-{class}-{seq:4}' -> '2026-PLANT-0007'."""
import re
out = fmt.replace("{yyyy}", f"{year:04d}").replace("{class}", class_code)
return re.sub(r"\{seq(?::(\d+))?\}", lambda m: str(n).zfill(int(m.group(1) or 1)), out)
def _same(a, b) -> bool:
if isinstance(a, Decimal) or isinstance(b, Decimal):
return a is not None and b is not None and Decimal(str(a)) == Decimal(str(b))
return a == b
def _decode(kind: str, raw: str):
v = json.loads(raw)
if kind == "decimal":
return Decimal(str(v))
if kind == "bool":
return bool(v)
if kind == "int":
return None if v is None else int(v)
return v