"""Generic CRUD service. One instance per entity model. Handles list/get/create/update/soft-delete/restore + version snapshot + audit emission. Bulk variants live in `ypms.framework.bulk`. Vendored from OWC `framework/crud.py` (snapshot 2026-04-20). Changes from upstream: - `owc` → `ypms` import renames. - **Pagination bugfix**: `next_cursor` is now the id of the last **returned** row (not the peek-ahead row). Upstream returned the (limit+1)th row's id, which combined with `where id > cursor` on the next call silently skipped that row. Caught by `tests/framework/test_crud.py::test_list_pagination_forwards_cursor`. Flag to OWC when re-vendoring. - **Snapshot redaction** (milestone 0.5): models may declare `__audit_redact__ = ("col", ...)`; `_to_dict` replaces those columns' values with `"[REDACTED]"` (nulls stay null, so "was it set" is still visible). Keeps app-layer ciphertext (e.g. `registrants.ssn_encrypted`) out of `audit_log` and `version_history`. Flag to OWC — Claimant SSN has the same exposure there. """ from __future__ import annotations import builtins from datetime import UTC, date, datetime from typing import Any, Generic, TypeVar from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.exc import NoResultFound from sqlalchemy.orm import Session from ypms.audit import service as audit from ypms.framework.pagination import Page from ypms.framework.version_history import VersionHistory from ypms.users.models import User ModelT = TypeVar("ModelT") CreateT = TypeVar("CreateT", bound=BaseModel) UpdateT = TypeVar("UpdateT", bound=BaseModel) REDACTED = "[REDACTED]" def _to_dict(obj: Any) -> dict[str, Any]: if obj is None: return {} redact = getattr(obj, "__audit_redact__", ()) out: dict[str, Any] = {} for c in obj.__table__.columns: v = getattr(obj, c.name, None) if c.name in redact and v is not None: v = REDACTED elif isinstance(v, (datetime, date)): v = v.isoformat() out[c.name] = v return out class CRUDService(Generic[ModelT, CreateT, UpdateT]): # noqa: UP046 — TypeVars reused by bulk.py def __init__(self, model: type[ModelT], entity: str): self.model = model self.entity = entity # ------------- read ------------- def get(self, db: Session, id: int, *, include_deleted: bool = False) -> ModelT: obj = db.get(self.model, id) if obj is None: raise NoResultFound(f"{self.entity} {id} not found") if not include_deleted and getattr(obj, "deleted_at", None) is not None: raise NoResultFound(f"{self.entity} {id} not found") return obj def list( self, db: Session, *, cursor: int | None = None, limit: int = 50, include_deleted: bool = False, newest_first: bool = False, ) -> Page[ModelT]: if newest_first: stmt = select(self.model).order_by(self.model.id.desc()).limit(limit + 1) # type: ignore[attr-defined] if cursor is not None: stmt = stmt.where(self.model.id < cursor) # type: ignore[attr-defined] else: stmt = select(self.model).order_by(self.model.id.asc()).limit(limit + 1) # type: ignore[attr-defined] if cursor is not None: stmt = stmt.where(self.model.id > cursor) # type: ignore[attr-defined] if not include_deleted and hasattr(self.model, "deleted_at"): stmt = stmt.where(self.model.deleted_at.is_(None)) # type: ignore[attr-defined] rows = db.execute(stmt).scalars().all() # Fetched limit+1 to peek ahead; if we got more than `limit`, there is # another page. next_cursor is the id of the LAST RETURNED row so the # next call's `where id > cursor` starts strictly after it (no skip). items = list(rows[:limit]) next_cursor = items[-1].id if len(rows) > limit else None # type: ignore[attr-defined] return Page(items=items, next_cursor=next_cursor) # ------------- write ------------- def create(self, db: Session, payload: CreateT, *, actor: User | None = None) -> ModelT: data = payload.model_dump(exclude_unset=True) obj = self.model(**data) # type: ignore[call-arg] db.add(obj) db.flush() self._snapshot(db, obj, actor) audit.record( db, actor=actor, action="create", entity=self.entity, entity_id=obj.id, after=_to_dict(obj), ) return obj def update(self, db: Session, id: int, payload: UpdateT, *, actor: User | None = None) -> ModelT: obj = self.get(db, id) before = _to_dict(obj) for k, v in payload.model_dump(exclude_unset=True).items(): setattr(obj, k, v) if hasattr(obj, "version"): obj.version = (obj.version or 1) + 1 db.flush() self._snapshot(db, obj, actor) audit.record( db, actor=actor, action="update", entity=self.entity, entity_id=obj.id, before=before, after=_to_dict(obj), ) return obj def soft_delete(self, db: Session, id: int, *, actor: User | None = None) -> ModelT: obj = self.get(db, id) if not hasattr(obj, "deleted_at"): raise TypeError(f"{self.entity} does not support soft delete") before = _to_dict(obj) obj.deleted_at = datetime.now(UTC) db.flush() audit.record( db, actor=actor, action="delete", entity=self.entity, entity_id=obj.id, before=before, after=_to_dict(obj), ) return obj def restore(self, db: Session, id: int, *, actor: User | None = None) -> ModelT: obj = self.get(db, id, include_deleted=True) before = _to_dict(obj) obj.deleted_at = None db.flush() audit.record( db, actor=actor, action="restore", entity=self.entity, entity_id=obj.id, before=before, after=_to_dict(obj), ) return obj # ------------- internals ------------- def _snapshot(self, db: Session, obj: ModelT, actor: User | None) -> None: if not hasattr(obj, "version"): return db.add( VersionHistory( entity=self.entity, entity_id=obj.id, # type: ignore[attr-defined] version=obj.version, # type: ignore[attr-defined] actor_id=actor.id if actor else None, snapshot=_to_dict(obj), ) ) db.flush() def history(self, db: Session, id: int) -> builtins.list[VersionHistory]: stmt = ( select(VersionHistory) .where(VersionHistory.entity == self.entity, VersionHistory.entity_id == id) .order_by(VersionHistory.version.asc()) ) return list(db.execute(stmt).scalars().all())