sqlalchemy 如何实现逻辑删除并在查询中自动过滤
技术百科
舞夢輝影
发布时间:2026-01-26
浏览: 次 逻辑删除是通过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)),否则一查就漏数据。
# app
# 子类
# 为什么
# NULL
# session
# cad
# sql
# Boolean
相关栏目:
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
AI推广<?muma echo $count; ?>
】
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
SEO优化<?muma echo $count; ?>
】
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
技术百科<?muma echo $count; ?>
】
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
谷歌推广<?muma echo $count; ?>
】
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
百度推广<?muma echo $count; ?>
】
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
网络营销<?muma echo $count; ?>
】
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
案例网站<?muma echo $count; ?>
】
<?muma
$count = M('archives')->where(['typeid'=>$field['id']])->count();
?>
【
精选文章<?muma echo $count; ?>
】
相关推荐
- PHP 中如何在函数内持久化修改引用变量的指向
- php下载安装后memory_limit怎么设置_
- Mac如何使用听写功能_Mac语音输入打字【效率技
- Win11怎么关闭自动更新 Win11永久关闭系统
- Windows11如何设置专注助手_Windows
- php怎么下载安装并配置环境变量_命令行调用PHP
- c++怎么使用std::filesystem遍历文
- 如何快速验证Golang安装是否成功_运行go v
- Go语言中slice追加操作的底层共享机制解析
- 如何在 VS Code 中正确配置并使用 NumP
- php485支持哪些操作系统_php485跨系统支
- Win11玩游戏全屏闪退怎么办_Win11全屏优化
- 一文教你快速开通网站LOGO图
- 如何使用Golang实现容器健康检查_监控和自动重
- php怎么操作Redis_Redis扩展连接与基本
- Windows10蓝屏代码DPC_WATCHDOG
- 如何使用正则表达式批量替换重复的“-”模式为固定字
- Win10怎样卸载iTunes_Win10卸载iT
- 如何在 Laravel 中通过嵌套关联关系进行 o
- php怎么下载安装后测试是否成功_简单脚本验证方法
- Windows10如何更改鼠标图标_Win10鼠标
- Python高性能计算项目教程_NumPyCyth
- Windows10无法连接到Internet_Wi
- 短链接还原php提示内存不足_调整PHP内存限制设
- c++怎么设置线程优先级与cpu亲和性_c++ 多
- 如何在 Go 中可靠地测试含 time.Time
- Mac如何查看电池健康百分比_Mac系统信息电源检
- php转mp4怎么保留字幕_php处理带字幕视频转
- Win10如何更改开机密码_Windows10登录
- Windows电脑如何截屏?(四种快捷方法)
- php本地部署后数据库连接报错_1045acces
- Win11怎么设置默认浏览器Chrome_Wind
- C++如何使用std::transform批量处理
- Windows10怎么卸载预装软件_Windows
- 如何使用Golang benchmark测量函数延
- Linux如何申请SSL免费证书_Linux下Ce
- 如何在包含多值的列中精准搜索指定演员?
- Linux怎么实现内网穿透_Linux安装Frp客
- Win11怎么设置开机自动连接宽带_Windows
- Windows服务无法启动错误1067是什么_进程
- php485函数怎么捕获异常_php485错误处理
- Go语言中CookieJar的持久化机制解析:内存
- 如何使用Golang反射将map转换为struct
- Flask 表单数据通过 SMTP 发送邮件的完整
- windows如何禁用驱动程序强制签名_windo
- Windows 11无法安全删除U盘提示设备正在使
- Windows10怎样设置家长控制_Windows
- Windows10怎么查看硬件信息_Windows
- Windows10电脑怎么设置虚拟内存_Win10
- Python抽象类与接口设计_规范说明【指导】

QQ客服