196 lines
6.9 KiB
Python
196 lines
6.9 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
FK label resolution: resolve foreign-key IDs to human-readable labels.
|
|
|
|
Works with the fk_target annotations on Pydantic models (see fkRegistry.py)
|
|
to auto-build label resolvers for paginated record sets.
|
|
"""
|
|
|
|
import logging
|
|
from functools import partial
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Individual FK label resolvers (db, ids) -> {id: label}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def resolveMandateLabels(db, ids: List[str]) -> Dict[str, Optional[str]]:
|
|
"""Resolve mandate IDs to labels. Returns None (not the ID!) for
|
|
unresolvable entries so the caller can distinguish "resolved" from "missing".
|
|
"""
|
|
from modules.datamodels.datamodelUam import Mandate
|
|
uniqueIds = list(set(ids))
|
|
records = db.getRecordset(Mandate, recordFilter={"id": uniqueIds}) or []
|
|
found: Dict[str, dict] = {}
|
|
for rec in records:
|
|
mid = rec.get("id", "")
|
|
found[mid] = rec
|
|
result: Dict[str, Optional[str]] = {}
|
|
for mid in ids:
|
|
m = found.get(mid)
|
|
label = (m.get("label") or m.get("name")) if m else None
|
|
if not label:
|
|
logger.debug("resolveMandateLabels: no label for id=%s (found=%s)", mid, m is not None)
|
|
result[mid] = label or None
|
|
return result
|
|
|
|
|
|
def resolveInstanceLabels(db, ids: List[str]) -> Dict[str, Optional[str]]:
|
|
"""Resolve feature-instance IDs to labels. Returns None for unresolvable."""
|
|
from modules.datamodels.datamodelFeatures import FeatureInstance
|
|
result: Dict[str, Optional[str]] = {}
|
|
for iid in ids:
|
|
records = db.getRecordset(FeatureInstance, recordFilter={"id": iid})
|
|
if records:
|
|
label = records[0].get("label") or None
|
|
result[iid] = label
|
|
else:
|
|
logger.debug("resolveInstanceLabels: no label for id=%s", iid)
|
|
result[iid] = None
|
|
return result
|
|
|
|
|
|
def resolveUserLabels(db, ids: List[str]) -> Dict[str, Optional[str]]:
|
|
"""Resolve user IDs to display names. Returns None for unresolvable."""
|
|
from modules.datamodels.datamodelUam import UserInDB as _UserInDB
|
|
uniqueIds = list(set(ids))
|
|
users = db.getRecordset(
|
|
_UserInDB,
|
|
recordFilter={"id": uniqueIds},
|
|
)
|
|
result: Dict[str, Optional[str]] = {}
|
|
found: Dict[str, dict] = {}
|
|
for u in (users or []):
|
|
uid = u.get("id", "")
|
|
found[uid] = u
|
|
for uid in ids:
|
|
u = found.get(uid)
|
|
if u:
|
|
result[uid] = u.get("displayName") or u.get("username") or u.get("email") or None
|
|
else:
|
|
result[uid] = None
|
|
return result
|
|
|
|
|
|
def resolveRoleLabels(db, ids: List[str]) -> Dict[str, Optional[str]]:
|
|
"""Resolve Role.id to roleLabel. Returns None for unresolvable."""
|
|
if not ids:
|
|
return {}
|
|
from modules.datamodels.datamodelRbac import Role as _Role
|
|
recs = db.getRecordset(
|
|
_Role,
|
|
recordFilter={"id": list(set(ids))},
|
|
) or []
|
|
out: Dict[str, Optional[str]] = {i: None for i in ids}
|
|
for r in recs:
|
|
rid = r.get("id")
|
|
if rid:
|
|
out[rid] = r.get("roleLabel") or None
|
|
for rid in ids:
|
|
if out.get(rid) is None:
|
|
logger.debug("resolveRoleLabels: no label for id=%s", rid)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resolver registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_BUILTIN_FK_RESOLVERS: Dict[str, Callable] = {
|
|
"Mandate": resolveMandateLabels,
|
|
"FeatureInstance": resolveInstanceLabels,
|
|
"UserInDB": resolveUserLabels,
|
|
"Role": resolveRoleLabels,
|
|
}
|
|
|
|
|
|
def buildLabelResolversFromModel(
|
|
modelClass: type,
|
|
db=None,
|
|
) -> Dict[str, Callable[[List[str]], Dict[str, str]]]:
|
|
"""
|
|
Auto-build labelResolvers dict from ``json_schema_extra.fk_target`` on a Pydantic model.
|
|
Maps field names to resolver functions when the target table has a registered builtin
|
|
resolver and ``fk_target.labelField`` is set (non-None).
|
|
|
|
When ``db`` is provided, the returned resolvers are pre-bound with partial(resolver, db)
|
|
so they can be called as resolver(ids).
|
|
"""
|
|
resolvers: Dict[str, Callable[[List[str]], Dict[str, str]]] = {}
|
|
for name, fieldInfo in modelClass.model_fields.items():
|
|
extra = fieldInfo.json_schema_extra
|
|
if not extra or not isinstance(extra, dict):
|
|
continue
|
|
tgt = extra.get("fk_target")
|
|
if not isinstance(tgt, dict):
|
|
continue
|
|
if tgt.get("labelField") is None:
|
|
continue
|
|
fkModel = tgt.get("table")
|
|
if fkModel and fkModel in _BUILTIN_FK_RESOLVERS:
|
|
fn = _BUILTIN_FK_RESOLVERS[fkModel]
|
|
resolvers[name] = partial(fn, db) if db else fn
|
|
return resolvers
|
|
|
|
|
|
def enrichRowsWithFkLabels(
|
|
rows: List[Dict[str, Any]],
|
|
modelClass: type = None,
|
|
*,
|
|
db=None,
|
|
labelResolvers: Optional[Dict[str, Callable[[List[str]], Dict[str, Optional[str]]]]] = None,
|
|
extraResolvers: Optional[Dict[str, Callable[[List[str]], Dict[str, Optional[str]]]]] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Add ``{field}Label`` columns to each row for every FK field that has a
|
|
registered resolver.
|
|
|
|
``modelClass`` — if provided, resolvers are auto-built from ``fk_target``
|
|
annotations on the Pydantic model (via ``buildLabelResolversFromModel``).
|
|
Requires ``db`` to be passed.
|
|
|
|
``labelResolvers`` — explicit resolver map that overrides auto-built ones.
|
|
Each resolver has signature ``(ids: List[str]) -> Dict[str, Optional[str]]``.
|
|
|
|
``extraResolvers`` — merged on top of auto-built / explicit resolvers. Use
|
|
for ad-hoc fields that are not FK-annotated on the model (e.g.
|
|
``createdByUserId`` on billing transactions).
|
|
|
|
If a label cannot be resolved the ``{field}Label`` value is ``None``
|
|
(never the raw ID — that would reintroduce the silent-truncation bug).
|
|
"""
|
|
resolvers: Dict[str, Callable] = {}
|
|
|
|
if modelClass is not None and labelResolvers is None:
|
|
resolvers = buildLabelResolversFromModel(modelClass, db)
|
|
elif labelResolvers is not None:
|
|
resolvers = dict(labelResolvers)
|
|
|
|
if extraResolvers:
|
|
resolvers.update(extraResolvers)
|
|
|
|
if not resolvers or not rows:
|
|
return rows
|
|
|
|
for field, resolver in resolvers.items():
|
|
ids = list({str(r.get(field)) for r in rows if r.get(field)})
|
|
if not ids:
|
|
continue
|
|
try:
|
|
labelMap = resolver(ids)
|
|
except Exception as e:
|
|
logger.error("enrichRowsWithFkLabels: resolver for '%s' raised: %s", field, e)
|
|
labelMap = {}
|
|
|
|
labelKey = f"{field}Label"
|
|
for r in rows:
|
|
fkVal = r.get(field)
|
|
if fkVal:
|
|
r[labelKey] = labelMap.get(str(fkVal))
|
|
else:
|
|
r[labelKey] = None
|
|
|
|
return rows
|