如何在SQLAlchemy软删除Mixin中触发before/after delete映射器事件？

需求背景 #

我正在实现一个soft delete混入类（Mixin），调用Object.delete()时不会真实删除数据，而是将deleted_at字段设置为当前时间。尽管并未实际删除数据，仍希望应用表现得如同执行了删除操作，并触发before/after delete事件，以此让钩子处理删除后需执行的一系列操作。

现有软删除Mixin代码：

from sqlalchemy import Column, DateTime, func
from sqlalchemy.ext.declarative import Base

class SoftDeleteMixin:
    deleted_at = Column(DateTime, nullable=True)

    def delete(self):
        self.deleted_at = func.now()
        # 希望在此处触发before/after delete映射器事件，但不知道怎么做

    def undelete(self):
        self.deleted_at = None

继承该Mixin的类：

class User(Base, SoftDeleteMixin):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    # 其他字段定义

期望触发的事件监听：

from sqlalchemy import event

@event.listens_for(User, "before_delete")
def before_delete(mapper, connection, target):
    # 执行删除前的操作
    pass

@event.listens_for(User, "after_delete")
def after_delete(mapper, connection, target):
    # 执行删除后的操作
    pass

解决方案 #

SQLAlchemy的before_delete/after_delete事件默认仅在执行真实DELETE SQL语句时触发，直接修改deleted_at字段不会自动触发这些事件。我们可以手动调用SQLAlchemy的事件分发器来触发指定事件，修改SoftDeleteMixin的delete方法即可实现需求：

from sqlalchemy import Column, DateTime, func, inspect
from sqlalchemy.orm import object_session

class SoftDeleteMixin:
    deleted_at = Column(DateTime, nullable=True)

    def delete(self):
        # 获取当前对象的会话与数据库连接
        session = object_session(self)
        connection = session.connection() if session else None
        # 获取当前对象对应的映射器
        mapper = inspect(self).mapper

        # 手动触发before_delete事件
        mapper.dispatch.before_delete.send(mapper, connection=connection, target=self)

        # 执行软删除逻辑
        self.deleted_at = func.now()
        # 若对象在会话中，标记为已修改以确保后续提交更新
        if session:
            session.add(self)

        # 手动触发after_delete事件
        mapper.dispatch.after_delete.send(mapper, connection=connection, target=self)

    def undelete(self):
        self.deleted_at = None
        session = object_session(self)
        if session:
            session.add(self)

说明 #

• 通过inspect(self).mapper获取对象对应的映射器，利用映射器的dispatch属性访问事件分发器
• 使用send()方法触发事件时，需传递事件定义要求的参数（mapper、connection、target）
• 若对象处于SQLAlchemy会话管理中，调用session.add(self)可确保修改被会话追踪，最终提交时同步到数据库
• 该方案仅针对实例级别的delete()方法，若需支持批量删除（如User.query.filter(...).delete()），需额外拦截真实删除操作并替换为软删除逻辑

内容的提问来源于stack exchange，提问作者Michael Seltenreich