YPMS/src/ypms/framework/crud.py

"""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())