SQLAlchemy Cheatsheet 13 — Soft Delete, Audit, and Timestamps

class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=datetime.utcnow,
    )

class User(TimestampMixin, Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)

class SoftDeleteMixin:
    deleted_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), default=None, index=True,
    )
    
    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

class User(SoftDeleteMixin, Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)

async def soft_delete(session, user: User):
    user.deleted_at = datetime.utcnow()
    await session.commit()

from sqlalchemy.orm import with_loader_criteria
from sqlalchemy import event, Session

@event.listens_for(Session, "do_orm_execute")
def soft_delete_filter(execute_state):
    if not execute_state.is_select:
        return
    if execute_state.execution_options.get("include_deleted"):
        return
    execute_state.statement = execute_state.statement.options(
        with_loader_criteria(
            SoftDeleteMixin,
            lambda cls: cls.deleted_at == None,
            include_aliases=True,
        )
    )

result = await session.execute(
    select(User).execution_options(include_deleted=True)
)

class TrackedMixin(TimestampMixin, SoftDeleteMixin):
    pass

class User(TrackedMixin, Base):
    __tablename__ = "users"

class AuditLog(Base):
    __tablename__ = "audit_log"
    id: Mapped[int] = mapped_column(primary_key=True)
    actor_id: Mapped[int | None]
    action: Mapped[str]                       # "user.create", "user.update", ...
    target_type: Mapped[str]
    target_id: Mapped[str]
    before: Mapped[dict | None] = mapped_column(JSONB if PG else JSON)
    after: Mapped[dict | None] = mapped_column(JSONB if PG else JSON)
    occurred_at: Mapped[datetime] = mapped_column(server_default=func.now())

@event.listens_for(Session, "before_flush")
def collect_audits(session, flush_context, instances):
    audits = []
    for obj in session.new:
        audits.append(AuditLog(
            actor_id=session.info.get("actor_id"),
            action=f"{type(obj).__name__.lower()}.create",
            target_type=type(obj).__name__,
            target_id=str(getattr(obj, "id", "")),
            after=_serialize(obj),
        ))
    for obj in session.dirty:
        attrs = inspect(obj).attrs
        before = {k: a.history.deleted[0] for k, a in attrs.items() if a.history.has_changes()}
        after = {k: getattr(obj, k) for k in before}
        audits.append(AuditLog(
            actor_id=session.info.get("actor_id"),
            action=f"{type(obj).__name__.lower()}.update",
            target_type=type(obj).__name__,
            target_id=str(obj.id),
            before=before, after=after,
        ))
    for obj in session.deleted:
        audits.append(AuditLog(...))
    
    for a in audits:
        session.add(a)

async def get_db_with_actor(
    user: User = Depends(current_user),
) -> AsyncSession:
    async with AsyncSessionLocal() as session:
        session.info["actor_id"] = user.id
        yield session

def _serialize(obj):
    # Drop sensitive fields; only primitive-cast values
    SENSITIVE = {"hashed_password", "password"}
    out = {}
    for col in inspect(obj.__class__).columns:
        if col.name in SENSITIVE: continue
        v = getattr(obj, col.name)
        out[col.name] = v.isoformat() if isinstance(v, datetime) else v
    return out

class UserHistory(Base):
    __tablename__ = "user_history"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int]
    version: Mapped[int]
    snapshot: Mapped[dict] = mapped_column(JSONB)
    occurred_at: Mapped[datetime] = mapped_column(server_default=func.now())

CREATE OR REPLACE FUNCTION audit_users() RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO audit_log (action, target_type, target_id, before, after)
    VALUES (TG_OP, 'User', NEW.id::text, row_to_json(OLD), row_to_json(NEW));
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER user_audit_trigger
AFTER INSERT OR UPDATE OR DELETE ON users
FOR EACH ROW EXECUTE FUNCTION audit_users();

DELETE FROM users WHERE deleted_at < now() - interval '30 days';