如何在Python Flask中通过SQLAlchemy执行复杂的原生SQL查询？-Python教程-PHP中文网

flask-sqlalchemy 3.x+ 中必须用 db.session.execute() 执行原生 sql，返回 result 对象，需调用 .mappings().all() 等方法取值；参数须用命名绑定（如 :user_id）并以字典传入；dml 操作后须手动 commit()；字段名不匹配时推荐 .mappings() 获取字典式结果。

如何在python flask中通过sqlalchemy执行复杂的原生sql查询？

直接用 `db.session.execute()` 执行原生 SQL

Flask-SQLAlchemy 3.x+ 默认禁用 db.engine.execute()，必须走 db.session.execute()。它返回一个 Result 对象（不是 list），需显式取值。

查询后必须调用 .scalars()（取单列）或 .mappings()（取字典）或 .all()（取元组）才能拿到数据
参数必须用命名绑定，不能用 %s 或 ? 占位符：text("SELECT * FROM users WHERE id = :user_id")
传参用字典： db.session.execute(stmt, {"user_id": 123})
如果只是 DML（INSERT/UPDATE/DELETE），记得手动 db.session.commit()

处理返回字段名不匹配 ORM 模型的情况

原生 SQL 返回的列名可能和模型字段名不一致（比如用了 AS 别名、聚合函数、跨表 JOIN），这时不能直接 .scalars() 或 .all() 映射到模型类。

用 .mappings().all() 得到每行一个 Mapping 对象（类似字典），字段名就是 SQL 中的列名或别名
如果想转成模型实例，得手动构造：User(id=row["user_id"], name=row["full_name"])
别用 db.session.query(User).from_statement(...) —— 它要求 SQL 字段严格对齐模型定义，否则报 sqlalchemy.exc.InvalidRequestError: Entity '<class>' has no property 'xxx'</class>

带参数的复杂查询示例：窗口函数 + 多表 JOIN

比如查每个用户的最新订单，并带用户信息和订单统计——这类逻辑在纯 ORM 中写起来绕，原生 SQL 更直白：

Python 3.14.2

Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本，属于3.14系列的第二个维护更新。该版本包含了18项修复，重点解决了多进程、数据类及正则表达式等模块的回归问题，并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式（移除GIL）正式获得官方支持，是Python发展的重要里程碑。

下载

<pre class="brush:php;toolbar:false;">from sqlalchemy import text
<p>stmt = text("""
SELECT
u.id AS user_id,
u.name,
o.id AS order_id,
o.created_at,
COUNT(*) OVER (PARTITION BY u.id) AS total_orders
FROM users u
INNER JOIN (
SELECT user_id, id, created_at,
ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at DESC) AS rn
FROM orders
) o ON u.id = o.user_id AND o.rn = 1
""")</p><p>result = db.session.execute(stmt).mappings().all()</p><h1>result 是 list[Mapping]，可直接用 result[0]["user_id"]、result[0]["total_orders"]</h1>

事务与连接生命周期容易踩的坑

原生 SQL 不受 ORM 自动 session 管理的“保护”，稍不注意就会出问题：

长耗时查询没加 timeout？数据库连接可能卡死 —— 建议在 <code>text() 后链式加 .execution_options(timeout=30)
执行了 UPDATE 但忘了 db.session.commit()？下次请求可能复用同一个 session，看到未提交的脏数据
在异步视图（如 @app.route(..., methods=["POST"]) 配合 async/await）里混用 db.session.execute()？SQLAlchemy 2.0+ 的 session 默认非 async，会阻塞事件循环

复杂查询涉及多步、状态依赖或高一致性要求时，别只盯着 SQL 写得对不对，得盯住 session 生命周期和事务边界。