用ORM用了两三年真正让我下决心把原生SQL这块彻底吃透的是去年接手的一套统计报表系统。报表里的留存率、分时趋势、多表聚合用ORM表达出来又臭又长生成的SQL执行计划还经常走偏DBA每次看到慢查询日志都来找我。后来我把SQLAlchemy里所有执行原生SQL的方式研究了一遍发现很多细节官方文档写了但没写透实际踩坑不少。这篇就把这些内容整理出来什么时候该用原生SQL、Engine和连接池怎么配、text()有哪些坑、结果集怎么高效处理以及怎么在FastAPI里组织事务和会话。基本都是可以直接抄作业的方案。1. 什么时候必须写原生SQLORM真的扛不住吗1.1 ORM的舒适区与盲区SQLAlchemy的ORM在常规CRUD、模型映射、关系加载上确实很爽。新手用它写增删改查代码量比纯SQL少一半而且数据库从MySQL换成PostgreSQL时大部分模型代码不用动。这也是大多数团队选择ORM的根本原因可维护性高、类型约束清晰、迁移方便。但ORM也有明显的盲区。复杂报表、多级子查询、窗口函数、递归CTE、数据库专有函数比如MySQL的DATE_FORMAT、JSON_TABLE、查询提示hint、以及批量更新场景ORM写起来要么API特别别扭要么生成的SQL形态不是DBA期望的样子。我见过最典型的案例一个按月分组统计的报表用ORM的func.date_format加上group_by执行耗时2秒多换成手写SQL后同样的结果0.1秒就出来了。不是ORM写错了而是它生成的谓词顺序、函数嵌套方式和现有索引不匹配优化器吃了亏。1.2 原生SQL的典型应用场景原生SQL适合的场景我个人总结下来大概这么几类统计报表尤其是跨表、跨库、带窗口函数的分组统计复杂的更新和删除比如update join、delete using这种调用存储过程、批量初始化数据需要精细控制SQL形态让执行计划符合预期从旧系统迁移来的存量SQL暂时没有精力翻译成ORM但别走极端我见过有人把所有查询全部硬编码成原生SQL动态条件一多拼接逻辑变得很难维护。实际上SQLAlchemy本身的查询构造器也可以帮你做条件组装然后用text()片段嵌入。所以我的习惯是ORM负责维护型CRUD原生SQL负责分析型查询和复杂更新两者混用。2. 建好Engine才能在SQLAlchemy里玩转原生SQL2.1 装依赖和创建引擎执行原生SQL的第一步是创建Engine。如果用SQLAlchemy 2.x需要先装好对应的数据库驱动pip install sqlalchemy pymysql # MySQL pip install sqlalchemy psycopg2-binary # PostgreSQL数据库URL的格式是固定的MySQL一般写成这样from sqlalchemy import create_engine engine create_engine( mysqlpymysql://root:your_password127.0.0.1:3306/example_db?charsetutf8mb4, echoFalse, )注意charset要放在URL参数里尤其是MySQL 8.0以下环境不指定字符集容易出现中文乱码。PostgreSQL则不需要在URL里写charset数据库编码一般由服务端控制。create_engine这一步实际上是惰性的调用它不会真的去连接数据库真正建立连接是在第一次execute或connect()时。这个特性意味着创建Engine的开销很小可以在应用启动时全局创建一次。2.2 连接池参数一定要配线上被断连的教训很多教程只教最小化配置没有讲连接池结果项目一上线就出问题。最典型的现象是系统运行一段时间后第一个请求非常慢或者直接报MySQL server has gone away。原因通常是数据库端的wait_timeout把空闲连接关了而客户端连接池不知道继续用已经失效的连接。解决办法就是配置连接池参数engine create_engine( mysqlpymysql://root:password127.0.0.1:3306/example_db?charsetutf8mb4, pool_size10, # 池中保持的连接数 max_overflow20, # 高峰期最多额外创建的连接数 pool_recycle3600, # 连接最多复用1小时主动换新 pool_pre_pingTrue, # 每次借出连接前先发一个轻量查询探测 )pool_pre_pingTrue特别推荐它会在每次从池里取出连接时自动发SELECT 1如果连接已经失效会自动丢弃并重新创建。这个机制的成本极低却能解决绝大多数断连问题。pool_recycle则适合数据库端超时时间比较短的场景主动让连接在超时前回收重建。在FastAPI项目里Engine通常在模块加载时创建一次作为全局对象使用。千万不要在每个请求里调用create_engine否则连接池形同虚设还白白增加开销。3. 四种执行原生SQL的方式我推荐这样选3.1 engine.connect()一次性的快速查询最基础的用法是engine.connect()它从连接池里拿出一个连接执行完SQL后把连接还回池子with engine.connect() as conn: result conn.execute( text(SELECT id, name, age FROM users WHERE age :min_age), {min_age: 18} ) rows result.fetchall() print(rows)这里有个容易踩的坑with engine.connect()只负责把连接归还连接池不会自动提交事务。如果执行的是INSERT、UPDATE、DELETE必须显式conn.commit()否则数据不会真正写入数据库。单独用connect()执行查询没问题但一旦涉及写操作我建议用下一种方式。3.2 engine.begin()把事务交给上下文管理器engine.begin()是专门为写操作设计的它会自动开启事务并在代码块正常结束时自动提交异常时自动回滚with engine.begin() as conn: conn.execute( text(UPDATE users SET age :age WHERE id :id), [{age: 20, id: 1}, {age: 21, id: 2}] )这是我执行原生写SQL最推荐的入口。它把事务边界和代码块绑定在一起不需要记着什么时候commit也避免了忘记提交或异常时忘记回滚的老大难问题。如果代码块中途抛异常整个事务会回滚到执行前状态数据安全性有保障。3.3 Session.execute()ORM项目里最顺手如果项目里已经在用ORM的Session那直接在session.execute()里执行原生SQL也完全可行from sqlalchemy.orm import Session with Session(engine) as session: rows session.execute( text(SELECT id, name FROM users WHERE status :status), {status: active} ).mappings().all()Session本身是围绕Connection的工作单元默认在一个事务上下文中操作。它最大的价值在于同一个Session里你可以让ORM模型操作和原生SQL操作共享同一个事务。比如先通过ORM插入一条用户记录再用原生SQL更新关联表的统计字段只要用的是同一个Session这两步就是一套原子操作要么全成功要么全回滚。3.4 直接拿DBAPI游标执行什么时候才用SQLAlchemy底层是DBAPI理论上你还可以拿到最底层的裸游标执行with engine.connect() as conn: raw_cursor conn.connection.cursor() raw_cursor.execute(SELECT COUNT(*) FROM users) print(raw_cursor.fetchone())这种做法非常少见它绕过了SQLAlchemy的参数绑定和类型转换容易出现SQL注入和数据类型问题。我只在两种情况下用过一是需要调用数据库驱动特有方法时二是做数据库备份或特殊管理操作时。日常开发里能不用就不用别给自己找麻烦。3.5 四种方式怎么选方式事务行为适用场景engine.connect()不自动提交需手动commit只读查询、一次性查询engine.begin()自动提交或回滚原生SQL写操作、批量DMLSession.execute()跟随Session事务ORM与原生SQL混用DBAPI裸游标完全手动控制驱动特有操作极少场景我日常的原则很简单只查询用engine.connect()写操作或需要事务保护只能用engine.begin()项目里已有Session且要混合ORM操作时用Session.execute()。按这个原则选基本不会错。4. text()是核心参数绑定的正确姿势4.1 为什么不要拼字符串执行原生SQL时SQL语句本身要用text()包一层。这不仅仅是格式要求更是安全底线。如果你这样写# 不推荐的写法存在SQL注入风险 sql fSELECT * FROM users WHERE name {name}一旦name里含有单引号SQL结构就被破坏了。name传; DROP TABLE users; --后果不堪设想。而text()结合参数绑定可以让SQL语句和参数分开传递给数据库驱动数据库端会严格区分“SQL结构”和“参数值”从机制上杜绝注入。4.2 命名参数和bindparamstext()内部使用冒号开头的命名参数from sqlalchemy import text stmt text(SELECT * FROM users WHERE name :name AND age :min_age) result conn.execute(stmt, {name: 张三, min_age: 18})传参时用字典键名要和SQL里的参数名一致。参数值可以是字符串、数字、None等常见类型默认行为是None时生成NULL。有些场景还需要给参数指定更具体的类型或者做额外控制这时候用bindparams()from sqlalchemy import bindparam stmt text(SELECT * FROM products WHERE id :id).bindparams( bindparam(id, type_Integer) ) result conn.execute(stmt, {id: 1001})开发中遇到MySQL类型转换报错或者参数类型推断不正确时就该想到用bindparam显式指定类型。4.3 批量参数执行与IN查询批量更新和批量插入用参数列表一次execute就可以完成with engine.begin() as conn: conn.execute( text(INSERT INTO audit_log (user_id, action) VALUES (:user_id, :action)), [ {user_id: 1, action: login}, {user_id: 2, action: logout}, {user_id: 3, action: login}, ] )这里传入的列表会被SQLAlchemy转成DBAPI的executemany批量执行比循环单条execute快十倍不止。如果是几万行的大批量写入建议分批提交每批几百到一千条比较稳妥避免事务日志暴涨。IN查询算是动态SQL里比较麻烦的。直接在SQL里写IN (:ids)会报参数无法渲染的错误正确做法是配合bindparam(..., expandingTrue)stmt text(SELECT * FROM products WHERE id IN :ids).bindparams( bindparam(ids, expandingTrue) ) result conn.execute(stmt, {ids: [1, 2, 3, 4, 5]})expandingTrue会把列表参数展开成IN (1, 2, 3, 4, 5)同时依然保持参数绑定不会有SQL注入问题。4.4 冒号的坑类型转换、URL和特殊字符使用text()最容易栽跟头的是冒号。因为text()把冒号当作参数前缀所以当SQL里出现其他意义的冒号时就会被误判。PostgreSQL的类型转换语法是::比如WHERE create_time :ts::timestamp。这条SQL实际上是有问题的因为::里的第一个冒号会被解析器认为是绑定参数开始导致参数解析错乱。解决办法是加空格改写或者用CAST函数-- 方案一加空格让解析器区分 WHERE create_time :ts ::timestamp -- 方案二更推荐不写数据库专有语法 WHERE create_time CAST(:ts AS timestamp)还有一种情况是SQL字符串常量里的URL例如http://abc.com。字符串中的冒号也可能干扰text()的解析。遇到这种问题不要硬拼SQL把URL当成参数传进去stmt text(SELECT * FROM sites WHERE url :url) result conn.execute(stmt, {url: http://abc.com})5. 结果集处理从Row到字典、模型对象5.1 fetchall、fetchone、fetchmany怎么配合执行查询后返回的是Result对象它不是简单的列表而是一个支持迭代的结果集。三种取数据方式对应不同场景with engine.connect() as conn: result conn.execute(text(SELECT id, name FROM users)) # 一次性取全部适合数据量小的场景 all_rows result.fetchall() # 一条一条取适合交互式处理 row result.fetchone() # 分块取适合大结果集避免内存爆炸 rows result.fetchmany(500)大数据量查询时不要无脑fetchall几十万行数据一次性加载到内存服务不崩也会严重拖垮性能。我一般配合fetchmany做分块处理with engine.connect() as conn: result conn.execute(text(SELECT id, payload FROM big_table)) while True: chunk result.fetchmany(1000) if not chunk: break for row in chunk: process(row)注意一点如果你用了with engine.connect()务必要在代码块内把结果集消费完或者提前关掉结果集。如果只取了几条就退出SQLAlchemy会在连接归还时自动清理剩余数据这个行为一般没问题但如果开启了服务端游标就需要主动关闭游标否则连接会被一直占用。5.2 mappings()拿字典性能敏感怎么办Row对象本身支持三种访问方式按下标、按列名、按属性。例如row[0]、row[name]、row.name。但在序列化成JSON、传给模板层时字典往往是更方便的形式。SQLAlchemy 2.0提供了mappings()方法with engine.connect() as conn: result conn.execute(text(SELECT id, name, age FROM users)) rows result.mappings().all() # 返回 [{id: 1, name: 张三, age: 18}, ...]调用mappings()后每一行变成RowMapping对象支持row[name]这种字典式访问在做接口返回时尤其好用。如果你的接口对性能很敏感比如每秒上千次查询直接使用元组访问row[0]会更快一点因为字典构造有额外开销。但通常这个差异是可以接受的优先保证代码可读性。5.3 把原生SQL结果转换成模型对象很多时候业务代码希望原生SQL查询结果能直接转换成Pydantic模型、数据类或ORM模型。在FastAPI接口里我一般是先变成字典再交给Pydantic做校验和序列化from pydantic import BaseModel class UserOut(BaseModel): id: int name: str age: int rows result.mappings().all() users [UserOut(**row) for row in rows]这里有个细节Row对象不能直接用**row解包需要先转成RowMapping用mappings()或者用dict(row._mapping)。如果你是在SQLAlchemy里做转换row result.fetchone() data dict(row._mapping) user UserOut(**data)对于Decimal和datetime类型直接塞进Pydantic模型一般没问题Pydantic会自动转换类型。但如果你使用标准库的json.dumpsDecimal会报错必须先转成float或str。这个细节在写接口时经常会遇到建议统一在序列化层处理不要在业务代码里到处加float()。6. 在FastAPI里用原生SQL组织事务6.1 生命周期与依赖注入FastAPISQLAlchemy组合开发时最常见的组织方式是把Session的创建和销毁放进依赖注入容器里from fastapi import Depends, FastAPI from sqlalchemy import create_engine, text from sqlalchemy.orm import Session app FastAPI() engine create_engine(mysqlpymysql://root:password127.0.0.1:3306/example_db?charsetutf8mb4) def get_session(): session Session(engine) try: yield session finally: session.close() app.get(/users/{min_age}) def list_users(min_age: int, session: Session Depends(get_session)): rows session.execute( text(SELECT id, name, age FROM users WHERE age :min_age), {min_age: min_age} ).mappings().all() return {items: rows}get_session里用yield把Session交给路由函数请求结束后在finally里关闭。这里确保每个请求都使用新的Session实例避免线程安全问题。一个容易被忽略的点是FastAPI中使用同步的SQLAlchemy时路由函数最好声明为普通def不要声明为async def。原因很简单普通def函数FastAPI会把它丢到线程池执行不会阻塞事件循环而async def函数如果直接调用同步的数据库操作会卡住事件循环影响整个服务的并发能力。如果非要用async def就得配合asyncio.to_thread或者选择异步驱动加create_async_engine那是另一套方案了。6.2 事务边界和回滚FastAPI的依赖注入里如果直接在get_session中yield session事务并不会自动提交。我建议在依赖里把提交和回滚逻辑集中处理好而不是散落在各个路由def get_session(): session Session(engine) try: yield session session.commit() except Exception: session.rollback() raise finally: session.close()这样每个请求只要不抛异常事务自动提交一旦接口处理中抛出异常事务自动回滚。路由函数本身不需要关心commit和rollback代码干净很多。当然如果业务需要在一个接口内做多个独立事务就不能用这个全局方案了需要手动控制边界。6.3 并发场景的资源控制FastAPI并发高时每个请求都会从连接池拿连接如果连接池太小会排队等待。这里有两个方向一是调大连接池参数比如pool_size20, max_overflow40二是减少每个请求占用连接的时间比如不要在事务里做耗时的外部HTTP调用、文件读写、计算等操作事务的范围越小越好。另一个隐藏问题是如果业务代码里不小心在使用Session的过程中又去调用了另一个新的engine.connect()就会同时占用两个连接高并发下容易出现连接池耗尽。建议一个请求内只维护一个数据库连接入口或者显式使用同一个Session避免嵌套申请连接。7. 原生SQL实战避坑清单常见问题与排查7.1 常见异常与处理方案异常信息可能原因处理方案MySQL server has gone away连接被数据库超时关闭设置pool_pre_pingTrue和pool_recycleCompileError: bind parameter name without a renderable value字典参数缺失或键名拼写不一致检查SQL冒号后的参数名与字典键名完全一致TypeError: Row object does not support item assignment试图修改Row对象Row是不可变对象需要转换成list或dictOperationalError: (1146, Table xxx doesnt exist)表名拼写错误或数据库选错确认Engine URL中的库名和表名sqlalchemy.exc.StatementError ... text() with unbound parameterstext()里写了:param但没传对应参数给execute传完整参数字典7.2 调试技巧echo和日志开发阶段排查SQL问题最简单的办法是打开echo功能engine create_engine(..., echoTrue)这样控制台会打印每条SQL语句及绑定参数。在生产环境不建议打开日志量太大。更精细的方式是配置日志级别import logging logging.getLogger(sqlalchemy.engine).setLevel(logging.INFO)还有一个比较好用的技巧把带绑定参数的text()编译成可直接执行的SQL字符串便于复制到数据库客户端排查from sqlalchemy.dialects import mysql stmt text(SELECT * FROM users WHERE age :min_age) compiled stmt.compile( dialectmysql.dialect(), compile_kwargs{literal_binds: True} ) print(compiled.string)这样输出的SQL会直接把min_age替换成实际值粘贴到Navicat或命令行就能跑。但注意literal_binds不会处理所有类型特殊类型可能转换失败一般调试够用。7.3 字符集、时区和Decimal序列化字符集问题多发于MySQL。建库、建表、连接三个层面统一使用utf8mb4才能正确存储Emoji和生僻字。连接层可以在URL里加?charsetutf8mb4如果还出现乱码检查一下数据库表本身的字符集SHOW CREATE TABLE users时区问题主要在datetime类型上。SQLAlchemy读出来的datetime默认不带时区信息如果数据库存的是UTC而业务需要展示东八区时间建议在查询时直接转换SELECT CONVERT_TZ(create_time, 00:00, 08:00) AS create_time FROM users这样数据库层就完成转换应用层不需要再处理时区偏移。Decimal序列化的问题前面提过。FastAPI使用Pydantic v2时Decimal类型可以直接响应但如果你在响应模型里定义为float会自动做转换。如果用标准库JSON需要手动处理json.dumps({amount: amount}, defaultlambda x: float(x) if isinstance(x, Decimal) else str(x))7.4 慢SQL排查的一个小顺序遇到原生SQL响应变慢我排查的顺序是先开数据库慢查询日志把实际执行的SQL捞出来然后EXPLAIN看执行计划重点看是否走了索引、扫描行数多少再用profile或连续执行观察稳定性。如果SQL本身没优化空间就考虑业务层面加缓存或者改造分页逻辑。很多慢SQL不是SQLAlchemy的问题而是SQL本身缺索引或写法没走最优路径。给一个很小的建议把项目中使用的原生SQL集中管理比如单独建一个sqls.py文件或者用SQL模板目录。这样排查问题时能快速定位也方便DBA统一review。我自己的习惯是每次写完一条复杂原生SQL会顺手在注释里写清这条SQL的使用场景和预期数据量对后续维护帮助很大。