逻辑删除是通过deleted_at等字段标记数据“已删除”而非物理删除，需自动过滤未删除记录以避免遗漏；SQLAlchemy需结合Query子类、事件监听和显式关系条件实现全局、安全、可绕过的软删机制。
什么是逻辑删除，为什么不能只靠 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 datetimeBase = 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 Queryclass 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 sessionmakerSession = 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))，否则一查就漏数据。
