逻辑删除是通过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 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))，否则一查就漏数据。