Snippets

from sqlalchemy import event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import with_loader_criteria

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# A mixin you can add to your models
class SoftDeletableMixin(object):
    deleted_at: Mapped[datetime | None] = mapped_column(
        nullable=True, default=None, sort_order=999
    )

    def soft_delete(self):
        self.deleted_at = datetime.utcnow()

# This listener automatically excludes all rows that have deleted_at not NULL
@event.listens_for(SessionLocal, "do_orm_execute")
def _do_orm_execute(orm_execute_state):
    if (
        orm_execute_state.is_select
        and not orm_execute_state.is_column_load
        and not orm_execute_state.is_relationship_load
    ):
        orm_execute_state.statement = orm_execute_state.statement.options(
            with_loader_criteria(
                SoftDeletableMixin,
                lambda cls: cls.deleted_at.is_(None),
                include_aliases=True,
            )
        )

# This listener instead of deleting, sets deleted_at to current time 
# if object has SoftDeletableMixin
@event.listens_for(SessionLocal, "before_flush")
def before_flush(session, flush_context, instances):
    for instance in session.deleted:
        if isinstance(instance, SoftDeletableMixin) and instance.deleted_at is None:
            instance.soft_delete()
            session.add(instance)
    session.deleted.clear()