sqlalchemy 如何实现“软删除”字段并自动过滤查询

软删除应使用 deleted_at 字段而非 is_deleted。建 datetime 类型可空字段，删除时设为当前时间，查询默认过滤 deleted_at IS NULL，需封装安全查询与 soft_delete 方法，并注意索引、恢复冲突及跨服务一致性。

软删除字段怎么建：用 deleted_at 比 is_deleted 更可靠

直接在模型里加一个可为空的 deleted_at 字段，类型为 DateTime（或数据库对应的时序类型），默认值为 None。不推荐用布尔字段 is_deleted，因为无法记录删除时间、难以做恢复审计、且在多条件过滤时容易漏掉 IS NULL 判断。

• deleted_at = column(DateTime, Nullable=True)
• 删除时不删行，而是设为 datetime.utcnow()
• 查询时默认只取 deleted_at IS NULL 的记录

如何自动过滤查询：用 default_Filter + Query 子类（SQLAlchemy 1.x）或 apply_criteria（2.x）

SQLAlchemy 1.x 可通过自定义 Query 类，在 iter 或 all() 前自动加过滤；SQLAlchemy 2.x 推荐用 select() + where() 组合，配合 apply_criteria 或封装查询函数。

• 1.x 方案：继承 Query，重写 iter，对所有模型检查是否存在 deleted_at 字段，有则追加 deletedat.is(None)
• 2.x 更自然：把“非删除”逻辑下沉到查询构造层，比如封装 safe_select(Model) 函数，内部自动 .where(Model.deletedat.is(None))
• 注意：手动写的 select().where(...) 不会自动触发过滤，必须显式调用封装逻辑

删除操作别手写 session.delete()：用自定义方法统一处理

session.delete(obj) 是真删，不能用于软删除。必须提供模型级方法，比如 obj.soft_delete()，它只更新 deleted_at 并提交。

• 在模型中定义方法：def soft_delete(self): self.deleted_at = datetime.utcnow()
• 配合 session.flush() 或 session.commit() 生效
• 如果用了 Event.listen(mapper, 'before_update')，注意判断是否真的在删（比如检查 deleted_at 从 None 变成非空），否则可能误触发
• 多表关联删除时，外键行不会自动软删，得手动遍历或用 cascade + 自定义事件处理

硬删、恢复、索引这些事容易被忽略

软删除不是万能的，几个实际踩坑点：

• 真要物理删除时，得显式 .filter(Model.deleted_at.isnot(None)).delete(synchronize_session=False)，否则会被默认过滤拦住
• 恢复操作就是 obj.deleted_at = None，但要注意唯一约束冲突（比如邮箱已被“删掉”的用户占着）
• deleted_at 字段务必加索引，否则带 WHERE deleted_at IS NULL 的大表查询会全表扫
• 使用 bulk_insert_mappings 或 execute(insert().values(...)) 时，不会触发模型方法或事件，deleted_at 默认是 None，但得确认业务是否允许批量插入即“已删除”状态

真正麻烦的是跨服务或异步任务里忘了查 deleted_at —— 一次漏判，数据就错位了。