本文深入探讨了在 SQLAlchemy 中实现健壮、安全且灵活的逻辑删除机制，不仅解析了为何简单依赖手动添加 `WHERE deleted_at IS NULL` 极易引发数据泄露和一致性问题，更系统性地提供了覆盖全局自动过滤（通过自定义 Query 子类）、无感软删转换（借助 `before_flush` 事件拦截 `session.delete()`）、关联关系语义保障（显式配置 `primaryjoin`）三大核心难点的生产级解决方案，兼顾默认安全性与按需绕过能力，让逻辑删除真正成为可信赖、可维护、可审计的数据生命周期管理基石。

什么是逻辑删除，为什么不能只靠 WHERE deleted_at IS NULL

逻辑删除不是真删数据，而是用字段（比如 deleted_at 或 is_deleted）标记某条记录“已删除”。但手动在每个查询里加 WHERE deleted_at IS NULL 极易遗漏，尤其 ORM 场景下关联查询、query.filter()、join() 都可能绕过这个条件。

SQLAlchemy 本身不内置逻辑删除支持，得靠组合机制实现“自动过滤”——核心是让所有查询默认带上未删除条件，且不影响硬删除、恢复、以及显式查已删除数据的场景。

用 __mapper_args__ + default / server_default 初始化删除字段

先确保模型有可被查询和更新的删除标识字段。推荐用 deleted_at（datetime 类型），比布尔值更利于审计和软恢复。

• deleted_at 设为 nullable=True，插入时默认为 None（即未删除）
• 用 default=None 和 server_default=text("NULL") 双保险，避免 ORM 层或直连 SQL 插入时意外写入非空值
• 不要用 is_deleted = Column(Boolean, default=False)，它无法区分“从未删除”和“删过又恢复”的状态

from sqlalchemy import Column, Integer, String, DateTime, text
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class User(Base):
tablename = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50))
deleted_at = Column(DateTime, nullable=True, default=None, server_default=text("NULL"))

用 Query 子类 + session.query_class 实现全局自动过滤

这是最稳定、兼容性最好的方式：自定义一个继承 Query 的类，在 all()、first()、filter() 等方法触发前自动补上未删除条件；同时保留绕过能力（比如查回收站时显式指定 deleted_at.isnot(None)）。

• 重写 __iter__() 和 all() 不够，必须覆盖 _execute_and_instances() 或更底层的 __clause_element__() —— 实际推荐直接改写 filter() 和 filter_by()，再配合 enable_assertions=False 避免重复添加
• 通过 session.query_class = SoftDeleteQuery 注入到 session，所有 session.query(Model) 都生效
• 关键点：只对“未显式涉及 deleted_at 字段”的查询自动加条件；一旦用户写了 .filter(User.deleted_at.isnot(None))，就不干预

from sqlalchemy.orm import Query
class SoftDeleteQuery(Query):
def new(cls, *args, **kwargs):
检查是否已手动过滤 deleted_at，避免重复添加
    if args and hasattr(args[0], '__table__'):
        has_deleted_filter = any(
            isinstance(criterion, (BinaryExpression, BooleanClauseList))
            and 'deleted_at' in str(criterion)
            for criterion in getattr(args[0], '_whereclause', []).compile().params.keys()
        )
        # 实际判断需解析 where clause，此处简化示意；生产建议用 _whereclause.traverse()
    return super().__new__(cls)

def filter(self, *criterion):
    # 如果没显式过滤 deleted_at，自动加上
    from sqlalchemy.sql import expression
    from sqlalchemy import or_
    no_deleted_filter = all(
        not (hasattr(c, 'left') and hasattr(c.left, 'name') and c.left.name == 'deleted_at')
        for c in criterion
    )
    if no_deleted_filter:
        criterion = criterion + (self.mapper.class_.deleted_at.is_(None),)
    return super().filter(*criterion)

然后创建 session 时绑定：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(query_cls=SoftDeleteQuery)
session = Session()

用 event.listen() 拦截 delete/update，避免手写 update(... set deleted_at=...)

业务代码不该暴露 deleted_at 赋值细节。监听 before_flush 事件，把 session.delete(obj) 转成更新操作，同时保持外键约束、级联行为可控。

• 监听 before_flush，遍历 session.deleted，对每个对象设置 deleted_at = datetime.utcnow()，再从 deleted 中移除
• 必须检查对象是否真有 deleted_at 属性，避免对无逻辑删除字段的模型报错
• 如果需要级联软删（如删 user 同时软删其 posts），得额外处理 session.new 和 session.dirty 中关联对象，不能只靠数据库 ON DELETE CASCADE
• 硬删除（绕过逻辑删除）可用特殊标记，例如给对象加临时属性 __hard_delete__ = True，在事件中跳过

from sqlalchemy import event
from datetime import datetime
@event.listens_for(Session, 'before_flush')
def before_flush(session, flush_context, instances):
for obj in list(session.deleted):
if hasattr(obj, 'deleted_at'):
obj.deleted_at = datetime.utcnow()
session.deleted.discard(obj)  # 移出 deleted 集合，防止真删

逻辑删除真正的复杂点不在“怎么加条件”，而在于关联查询时的语义一致性：比如 User.posts 关系默认应只返回未删除的 Post，但 SQLAlchemy 的 relationship() 不会自动继承主模型的软删规则。这需要在 relationship() 中显式用 primaryjoin 补上 and_(Post.user_id == User.id, Post.deleted_at.is_(None))，否则一查就漏数据。

本篇关于《SQLAlchemy逻辑删除与自动过滤实现》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！