flask-sqlalchemy 3.0+ 已弃用 basequery,须改用继承 sqlalchemy.orm.query 的自定义查询类,并通过 mapper_args 或初始化参数指定;逻辑删除需重写 query_class 的 delete() 方法实现 update 替代 delete,并在 init 中自动添加 is_deleted==false 过滤,同时必须重写 count() 方法以确保分页等统计准确。

为什么直接继承 BaseQuery 不再管用了
Flask-SQLAlchemy 3.0+ 已弃用 BaseQuery,改用 Query 类或更推荐的查询构造方式。如果你在新项目里还写 class MyQuery(BaseQuery),启动时会报 AttributeError: type object 'SQLAlchemy' has no attribute 'BaseQuery' —— 这不是你代码写错了,是框架本身删了这个入口。
真正该做的,是用 query_class 参数显式指定自定义查询类,且这个类必须继承自 sqlalchemy.orm.Query(不是已移除的 flask_sqlalchemy.BaseQuery)。
- 旧写法(失效):
class User(db.Model, QueryMixin): query_class = MyQuery - 新写法(有效):
class User(db.Model): __mapper_args__ = {"query_class": MyQuery}或在db = SQLAlchemy(query_class=MyQuery)初始化时传入 - 注意:
__mapper_args__是 SQLAlchemy 原生机制,Flask-SQLAlchemy 完全兼容
怎么让 delete() 变成 UPDATE 而非 DELETE
逻辑删除的核心是把 DELETE FROM user WHERE id=1 换成 UPDATE user SET is_deleted=True WHERE id=1。不能靠重写 __delete__,得拦截 ORM 的删除动作。
最稳妥的方式是覆写模型的 __mapper__.delete 方法,或者更常用、更清晰的做法:在自定义 query_class 中重载 delete() 方法,并配合一个标记字段(如 is_deleted)。
- 模型需定义字段:
is_deleted = db.Column(db.Boolean, default=False, nullable=False) - 自定义查询类中重写
delete(self, synchronize_session=False) - 里面调用
self.update({User.is_deleted: True}, synchronize_session=synchronize_session) - 务必保留
synchronize_session参数,否则 session 缓存不同步,后续查不到刚“删”掉的数据
如何默认过滤掉 is_deleted=True 的记录
自动过滤不是靠全局钩子,而是靠修改查询起点 —— 即让每次 User.query 默认带上 filter(is_deleted==False)。这要通过重写自定义查询类的 __iter__ 和 all() 等方法来实现?不,太重。正确做法是覆盖 __clause_element__() 或更简单:在自定义查询类的 __init__ 里自动加 filter。
- 在自定义查询类
__init__(self, *args, **kwargs)中追加:super().__init__(*args, **kwargs)→self = self.filter(self._entity_zero().is_deleted == False) - 但注意:不是所有查询都来自模型类(比如 join 查询),所以更安全的是只对单实体查询生效,可用
self._entities判断是否主实体是当前模型 - 更轻量方案:用
default_filter+apply_default_filters配合Query子类,但 Flask-SQLAlchemy 不直接暴露该机制,所以仍推荐 init 里条件过滤
软删除后 count() 和分页怎么不出错
count() 会绕过你自定义查询类的 __init__ filter,直接走底层 SQL COUNT(*),导致统计包含已逻辑删除的记录。这是最常被忽略的坑。
解决办法只有一个:永远不要用 query.count(),改用 query.with_entities(func.count()).scalar(),这样能确保 filter 生效;或者更干脆,在自定义查询类里重写 count(self) 方法:
def count(self):
# 先 clone 当前 query,去掉 limit/offset,再加 count
count_query = self.statement.with_only_columns([func.count()]).order_by(None)
return self.session.execute(count_query).scalar()
- 分页(如
paginate())依赖count(),所以必须重写它,否则页码和总数对不上 - 如果用 Flask-SQLAlchemy 自带的
paginate(),记得传error_out=False,避免因 count 错误触发异常 - 测试时重点验证:
User.query.filter(User.name.contains("x")).count()是否真的排除了is_deleted=True的行
逻辑删除看似只是加个字段、改个 delete 行为,但真正落地时,count、paginate、join、subquery 各种上下文都会绕过你的默认 filter,得一个个补漏。别指望一次封装就万事大吉。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











