当前位置： 首页 > news >正文

Flask ORM 查询详解：Model.query vs db.session.query vs db.session.execute

news 2025/8/15 5:28:29

Flask ORM 查询详解：Model.query vs db.session.query vs db.session.execute

在 Flask 中使用 SQLAlchemy ORM 进行数据库查询时，主要有三种方式：Model.query、db.session.query 和 db.session.execute。下面我将详细解释每种方法的使用方式、结果获取和结果类型。

1. Model.query (推荐用于简单查询)

这是最简洁的查询方式，直接通过模型类进行查询。

基本用法

from models import User  # 假设有 User 模型# 查询所有用户
all_users = User.query.all()# 根据ID查询单个用户
user = User.query.get(1)# 带过滤条件的查询
admin_users = User.query.filter_by(role='admin').all()

结果获取方法

方法返回类型描述示例
all()list[Model]返回所有结果的列表users = User.query.all()
first()Model 或 None返回第一个结果user = User.query.first()
get(ident)Model 或 None根据主键查询user = User.query.get(1)
one()Model必须有且只有一个结果，否则抛异常user = User.query.filter_by(email='admin@ex.com').one()
one_or_none()Model 或 None有结果返回结果，无结果返回Noneuser = User.query.filter_by(email='notexist@ex.com').one_or_none()
count()int返回结果数量count = User.query.count()
paginate()Pagination 对象分页查询page = User.query.paginate(page=1, per_page=20)

结果处理示例

# 获取所有用户列表
users = User.query.all()
print(type(users))  # <class 'list'>
print(type(users[0]))  # <class 'models.User'># 遍历用户
for user in users:print(user.username, user.email)# 分页查询
page = User.query.paginate(page=1, per_page=10)
print(page.items)  # 当前页的用户列表
print(page.total)  # 总记录数

2. db.session.query (推荐用于复杂查询)

这是更灵活的查询方式，特别适合需要查询多个表或特定字段的场景。

基本用法

from models import User, Post
from sqlalchemy.orm import aliased# 查询所有用户
all_users = db.session.query(User).all()# 查询特定字段
user_emails = db.session.query(User.email).all()# 多表查询
results = db.session.query(User, Post).join(Post, User.id == Post.user_id).all()# 使用别名
ua = aliased(User, name='ua')
admin_users = db.session.query(ua).filter(ua.role == 'admin').all()

结果获取方法

方法返回类型描述
all()list[Row] 或 list[tuple]返回所有结果
first()Row 或 tuple 或 None返回第一个结果
one()Row 或 tuple必须有且只有一个结果
one_or_none()Row 或 tuple 或 None最多一个结果
scalar()标量值返回第一行第一列的值
count()int返回结果数量

结果类型处理

# 查询整个模型对象
users = db.session.query(User).all()
# 类型: list[User]
for user in users:print(user.username)# 查询特定字段 - 返回元组
emails = db.session.query(User.email).all()
# 类型: list[Row] (行为类似元组)
for row in emails:print(row[0])  # 索引访问print(row.email)  # 属性访问# 查询多个字段
user_data = db.session.query(User.id, User.username).all()
# 类型: list[Row]
for row in user_data:print(f"ID: {row.id}, Username: {row.username}")# 多表查询 - 返回元组
user_posts = db.session.query(User, Post).join(Post).all()
for user, post in user_posts:print(f"{user.username} wrote: {post.title}")# 使用标量值
email_count = db.session.query(func.count(User.email)).scalar()
print(f"Total emails: {email_count}")

3. db.session.execute (用于原生SQL查询)

当需要执行原生SQL时使用此方法，返回结果为 ResultProxy 对象。

基本用法

# 执行原生SQL查询
result = db.session.execute("SELECT * FROM users WHERE role = :role", {'role': 'admin'})# 使用text构造SQL
from sqlalchemy import text
sql = text("SELECT * FROM users WHERE created_at > :start_date")
result = db.session.execute(sql, {'start_date': '2023-01-01'})

结果获取方法

方法返回类型描述
fetchall()list[RowProxy]返回所有结果
fetchone()RowProxy 或 None返回一行结果
fetchmany(size)list[RowProxy]返回指定数量的结果
scalar()标量值返回第一行第一列的值
keys()list[str]返回列名列表

结果类型处理

# 执行查询
result = db.session.execute("SELECT id, username, email FROM users")# 获取列名
print(result.keys())  # ['id', 'username', 'email']# 获取所有行
rows = result.fetchall()
# 类型: list[RowProxy]
for row in rows:# 索引访问print(row[0], row[1], row[2])# 列名访问print(row['id'], row['username'], row['email'])# 转换为字典row_dict = dict(row)print(row_dict)# 获取单行
row = result.fetchone()
if row:print(row.username)# 遍历结果集（适用于大量数据）
result = db.session.execute("SELECT * FROM large_table")
for row in result:process_row(row)  # 逐行处理，减少内存占用

三种方法对比

特性Model.querydb.session.querydb.session.execute
易用性⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐
灵活性⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐⭐
返回类型模型对象Row对象/元组RowProxy对象
ORM特性支持完整完整有限
原生SQL支持不支持部分支持完整支持
性能高高最高
适用场景简单CRUD操作复杂查询、连接查询高性能需求、复杂SQL

最佳实践建议

  1. 优先使用 Model.query

    • 适合大多数简单查询场景
    • 代码简洁易读
    • 直接返回模型对象

  2. 复杂查询使用 db.session.query

    • 需要多表连接时
    • 需要选择特定字段时
    • 需要使用SQL函数时
    from sqlalchemy import func# 每月新增用户统计
monthly_stats = db.session.query(func.date_trunc('month', User.created_at).label('month'),func.count(User.id).label('count')
).group_by('month').all()

  3. 只在必要时使用 db.session.execute

    • 执行复杂原生SQL
    • 需要最高性能时
    • 执行数据库特定功能
    # 使用数据库特定函数（PostgreSQL示例）
result = db.session.execute("""SELECT id, username, json_build_object('id', id, 'name', username) AS user_json FROM users
""")

  4. 结果转换技巧

    # 将Row对象转换为字典
def row_to_dict(row):return {column: getattr(row, column) for column in row._fields}# 将查询结果直接转为字典列表
users = db.session.query(User.id, User.username).all()
user_dicts = [dict(zip(u._fields, u)) for u in users]# 使用ScalarResult获取单一列的值列表
emails = db.session.query(User.email).scalars().all()

  5. 性能优化

    # 使用yield_per处理大数据集
large_query = User.query.yield_per(100)
for user in large_query:process_user(user)# 只选择需要的字段
# 而不是 select * 
db.session.query(User.id, User.name)

事务管理

所有查询都应该在事务上下文中执行：

@app.route('/users')
def get_users():try:# 查询操作users = User.query.all()# 修改操作new_user = User(name="John")db.session.add(new_user)db.session.commit()return jsonify([u.serialize() for u in users])except Exception as e:db.session.rollback()return jsonify({"error": str(e)}), 500

总结

  1. 简单查询：优先使用 Model.query，直接返回模型对象列表
  2. 复杂查询：使用 db.session.query，灵活处理多表关联和字段选择
  3. 原生SQL：使用 db.session.execute，处理特殊需求
  4. 结果处理：
    • 模型对象：直接访问属性
    • Row/RowProxy 对象：索引或列名访问
    • 标量值：使用 scalar() 或 scalars()
  5. 性能关键：使用 yield_per 处理大数据，只选择必要字段

根据具体需求选择最合适的查询方式，并在开发过程中注意结果类型的处理，可以大大提高 Flask 应用的数据库操作效率和代码可维护性。

http://www.dtcms.com/a/330386.html

相关文章：

  • Java Socket 基础教程
  • 新手入门 Makefile：FPGA 项目实战教程（一）
  • pnpm常用命令；为什么使用pnpm？
  • 古中医学习笔记专题文章导航
  • 【STM32入门教程】stm32简介
  • 基于 NVIDIA 生态的 Dynamo 风格分布式 LLM 推理架构
  • Kotlin Data Classes 快速上手
  • SwiftUI 页面弹窗操作
  • Windows批处理脚本自动合并当前目录下由You-get下载的未合并的音视频文件
  • Polyak-Ruppert 平均
  • UCLAMP3311T.TCT TVS二极管阵列 Semtech升特半导体 集成电路IC
  • tp5集成elasticsearch笔记
  • 20. 了解过尾递归优化吗
  • ASCII与Unicode：编码世界的奥秘
  • TLS 终止在真实业务中的防护价值
  • 36 C++ STL模板库5-string
  • Python网络爬虫（二） - 解析静态网页
  • IPTV系统：开启视听与管理的全新篇章
  • CMake 如何查找 Python2和Python3
  • 利用 Python 爬虫按图搜索 1688 商品（拍立淘）实战指南
  • 17. 如何判断一个对象是不是数组
  • 肖臻《区块链技术与应用》第十一讲：比特币核心概念重温：一文读懂私钥、交易、挖矿与网络现状
  • Redis7学习——Redis的十大类型String、List、Hash、Set、Zset
  • 解决：Gazebo连接模型数据库失败
  • linux 内核 - 内存管理概念
  • Apifox精准定义复杂API参数结构（oneOf/anyOf/allOf）
  • aave v3 存款与借款利息的计算方式
  • 码上爬第七题【协程+参数加密+响应解密+格式化检测】
  • C#面试题及详细答案120道（11-20）-- 面向对象编程（OOP）
  • LeetCode Day5 -- 二叉树