Soft delete mixin class for SQLAlchemy with event hooks

README.md

SQLAlchemy mixin example with bulk event assignment

This is a full example that shows how mixin classes can be implemented for use with SQLALchemy in a way that also automatically apply event listeners to appropriate subclasses of a given mixin class. The use case started with this question on StackOverflow which asked about how the before_delete mapper event can be used, which lead down the path of exploration that lead to this more generic skeleton on how this might be achieved in a more generic manner.

The following is an example run of the associated Python code:

INFO:sqlalchemy.engine.Engine:SELECT entity.id AS entity_id, entity.field_a AS entity_field_a, entity.field_b AS entity_field_b, entity.created_at AS entity_created_at, entity.deleted_at AS entity_deleted_at 
FROM entity
INFO:sqlalchemy.engine.Engine:[generated in 0.00009s] ()
INFO:__main__:<Entity id=1> is deleted
INFO:sqlalchemy.engine.Engine:UPDATE entity SET deleted_at=? WHERE entity.id = ?
INFO:sqlalchemy.engine.Engine:[generated in 0.00011s] ('2023-04-01 00:20:14.149779', 1)
INFO:sqlalchemy.engine.Engine:COMMIT
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:SELECT entity.id AS entity_id, entity.field_a AS entity_field_a, entity.field_b AS entity_field_b, entity.created_at AS entity_created_at, entity.deleted_at AS entity_deleted_at 
FROM entity
INFO:sqlalchemy.engine.Engine:[cached since 0.002176s ago] ()
INFO:__main__:<Entity id=1> is undeleted
INFO:sqlalchemy.engine.Engine:UPDATE entity SET deleted_at=? WHERE entity.id = ?
INFO:sqlalchemy.engine.Engine:[cached since 0.001268s ago] (None, 1)
INFO:sqlalchemy.engine.Engine:UPDATE entity SET field_a=? WHERE entity.id = ?
INFO:sqlalchemy.engine.Engine:[generated in 0.00009s] ('id:3 updated field_a', 3)
INFO:sqlalchemy.engine.Engine:COMMIT
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:SELECT document.id AS document_id, document.title AS document_title, document.deleted_at AS document_deleted_at 
FROM document
 LIMIT ? OFFSET ?
INFO:sqlalchemy.engine.Engine:[generated in 0.00010s] (1, 0)
INFO:__main__:<Document id=1> is deleted
INFO:sqlalchemy.engine.Engine:UPDATE document SET deleted_at=? WHERE document.id = ?
INFO:sqlalchemy.engine.Engine:[generated in 0.00011s] ('2023-04-01 00:20:14.154136', 1)
INFO:sqlalchemy.engine.Engine:SELECT user.id AS user_id, user.name AS user_name, user.deleted_at AS user_deleted_at 
FROM user
 LIMIT ? OFFSET ?
INFO:sqlalchemy.engine.Engine:[generated in 0.00013s] (1, 0)
INFO:__main__:<User id=1> is deleted
INFO:sqlalchemy.engine.Engine:UPDATE user SET deleted_at=? WHERE user.id = ?
INFO:sqlalchemy.engine.Engine:[generated in 0.00010s] ('2023-04-01 00:20:14.155961', 1)
INFO:sqlalchemy.engine.Engine:COMMIT
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:SELECT document.id AS document_id, document.title AS document_title, document.deleted_at AS document_deleted_at 
FROM document
 LIMIT ? OFFSET ?
INFO:sqlalchemy.engine.Engine:[cached since 0.003432s ago] (1, 0)
INFO:sqlalchemy.engine.Engine:COMMIT

sa_softdel_mixin_event.py

from datetime import datetime
from logging import getLogger
from sqlalchemy import ForeignKey
from sqlalchemy import create_engine, event, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session
from sqlalchemy.orm import InstrumentedAttribute, Relationship
from sqlalchemy.orm import mapped_column, relationship

logger = getLogger(__name__)


class Base(DeclarativeBase):
    def __repr__(self):
        return '<%s id=%r>' % (type(self).__name__, self.id)


class SoftDeleteMixin:
    deleted_at: Mapped[datetime] = mapped_column(nullable=True)
    _deleted_at_updated: bool = False

    def delete(self):
        if self.deleted_at is None:
            # only update deleted date if not already deleted
            self.deleted_at = datetime.now()

    def undelete(self):
        self.deleted_at = None


def deleted_at_set_listener(target, value, old_value, initiator):
    if isinstance(target, SoftDeleteMixin):
        target._deleted_at_updated = (value != old_value)
        for name, attr in vars(type(target)).items():
            if (isinstance(attr, InstrumentedAttribute) and
                    isinstance(attr.property, Relationship) and
                    attr.property.uselist):
                for item in getattr(target, name):
                    if isinstance(item, SoftDeleteMixin):
                        if value:
                            item.delete()
                        else:
                            item.undelete()


def before_update(mapper, connection, target):
    if isinstance(target, SoftDeleteMixin) and target._deleted_at_updated:
        if target.deleted_at:
            logger.info('%r is deleted', target)
        else:
            logger.info('%r is undeleted', target)


@event.listens_for(Base, "before_mapper_configured", propagate=True)
def on_new_class(mapper, cls_):
    if issubclass(cls_, SoftDeleteMixin):
        event.listen(cls_, "before_update", before_update)
        event.listen(cls_.deleted_at, "set", deleted_at_set_listener)


class Entity(Base, SoftDeleteMixin):
    __tablename__ = 'entity'
    id: Mapped[int] = mapped_column(primary_key=True)
    field_a: Mapped[str]
    field_b: Mapped[str]
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)


class User(Base, SoftDeleteMixin):
    __tablename__ = 'user'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    documents: Mapped[list["Document"]] = relationship(viewonly=True)
    comments: Mapped[list["Comment"]] = relationship(viewonly=True)


class Document(Base, SoftDeleteMixin):
    __tablename__ = 'document'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'))
    author: Mapped[User] = relationship()
    comments: Mapped[list["Comment"]] = relationship(back_populates='document')


class Comment(Base, SoftDeleteMixin):
    __tablename__ = 'comment'
    id: Mapped[int] = mapped_column(primary_key=True)
    content: Mapped[str]
    document_id: Mapped[int] = mapped_column(ForeignKey('document.id'))
    document: Mapped[Document] = relationship()
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'))
    author: Mapped[User] = relationship()


def setup_db():
    engine = create_engine('sqlite://')
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        john = User(name='John')
        dave = User(name='Dave')
        doc1 = Document(title='A serious document', author=john)
        com11 = Comment(content="John's first remark", document=doc1, author=john)
        com12 = Comment(content="Dave's first remark", document=doc1, author=dave)
        doc2 = Document(title='Second document', author=dave)
        com21 = Comment(content="John's second remark", document=doc2, author=john)
        com22 = Comment(content="Dave's second remark", document=doc2, author=dave)
        session.add(john)
        session.add(dave)
        session.add(doc1)
        session.add(doc2)
        session.add(com11)
        session.add(com12)
        session.add(com21)
        session.add(com22)
        session.commit()

    return engine


def run(engine):
    with Session(engine) as session:
        logger.info('*** deleting the first document (auto deletes its comments)')
        document = session.scalars(select(Document)).first().delete()
        session.commit()

    with Session(engine) as session:
        logger.info('*** deleting everything dave did')
        dave = session.scalars(
            select(User)
                .filter_by(name='Dave')
        ).first()
        dave.delete()
        session.commit()

    with Session(engine) as session:
        logger.info('*** undeleting everything dave did')
        # done _without_ consideration for deletes triggered by deletion 
        # of first document
        dave = session.scalars(
            select(User)
                .filter_by(name='Dave')
        ).first()
        dave.undelete()
        session.commit()


def main():
    engine = setup_db()
    run(engine)


if __name__ == '__main__':
    from logging import basicConfig, INFO
    basicConfig()
    getLogger().setLevel(INFO)
    # getLogger('sqlalchemy.engine').setLevel(INFO)
    main()