写SQLAlchemy ORM的文章不少,但大多数停留在“给你一段代码、你复制跑通”的层面。今天我想换个角度,从实际项目里怎么用、为什么要这么用、踩过哪些坑讲起,把SQLAlchemy ORM这层皮真正揭开来看看。
这篇文章适合谁?刚接触Python后端、想从裸SQL/手写连接池升级到ORM的开发者,以及已经用了SQLAlchemy但总觉得哪里别扭、想搞清楚内部机制的同学。我会结合一套实际的项目代码,把模型定义、会话管理、查询构造、事务边界、懒加载陷阱、迁移方案全部串起来讲,每一步都说明为什么这么做,而不是机械地堆代码。
先说结论:SQLAlchemy ORM真正值钱的不是“把SQL写成Python”,而是它帮你把数据访问层收拢成一套可维护、可测试、可演进的工程结构。但前提是,你必须理解它在背后做了什么,否则ORM只会帮你写出一堆莫名其妙的慢查询和连接泄漏。
1. 为什么选ORM而不是写原生SQL
1.1 ORM到底解决什么问题
我先抛一个场景。假设你要开发一套内容管理系统,涉及用户、文章、标签、评论四张表。用原生SQL写,你得自己管连接、游标、参数占位、结果集映射、事务提交回滚,每张表增删改查至少十来行样板代码,四张表下来就是一大坨重复劳动。更麻烦的是,表结构一旦调整,所有相关SQL全部要跟着改,漏一处就是线上事故。
ORM的核心价值在于,它把“数据库行”和“Python对象”之间的双向映射替你做了。你在代码里操作User(name='张三')、session.add(user),SQLAlchemy会在合适的时机翻译成INSERT INTO user (name) VALUES (?)并执行。反过来,查询结果也不再是一个个tuple或者dict,而是带类型、带关系、带方法的实体对象。这层抽象让业务代码可以只关注“对象之间的协作”,而不必时刻惦记“表之间怎么关联”。
我见过有人觉得ORM是累赘,理由是自己SQL写得很溜。这种想法在只有两三个表的小工具里成立,但一旦业务复杂度上来,对象关系映射带来的维护性优势就是碾压级的。你想想,几十张表、几百个字段、几十种关联查询,如果全部手写SQL并手工做映射,代码量和出错概率都是灾难级别。
1.2 SQLAlchemy在Python生态里的独特位置
Python里的ORM不止SQLAlchemy一个,Django ORM用的人也多,但两者定位完全不同。Django ORM绑定Django框架,你没法单独拿出来给Flask、FastAPI或普通脚本用。SQLAlchemy是框架无关的,你可以把它嵌进任何Python项目,配合Flask-SQLAlchemy、FastAPI的官方文档示例等中间件使用,也可以裸用Core部分只做SQL构造,完全按需取用。
SQLAlchemy另一个独到之处是它的两层架构:Core和ORM。Core层负责SQL表达式的构造与执行,你写的select(User).where(User.name == '张三')实际上先在Core层生成一个SQL表达式树,再编译成具体的方言SQL。ORM层则在这之上叠加了对象映射、身份映射(Identity Map)、变更追踪(Change Tracking)等工作。这种分层意味着,就算某些极端场景你不想要ORM的“自动魔法”,也可以在同一套工具里退回到Core级别手写SQL,不必引入第二种数据库访问库。
还有一个实际原因让我长期锁定SQLAlchemy:它对数据库方言的支持非常广。PostgreSQL、MySQL、SQLite、Oracle、SQL Server都能跑,切换数据库时大部分代码不用动,只有连接串和个别类型定义要改。这在创业项目里极其实用,因为早期用SQLite开发、上线换PostgreSQL是再常见不过的路径。
2. 模型定义:从表结构到Python类的映射
2.1 声明式映射的基础写法
SQLAlchemy最常用的是声明式映射(Declarative Mapping),你用一个Python类描述一张表,类属性对应列,类继承DeclarativeBase产生的基类。我用最经典的Base写法示例:先定义基类,然后每个模型类继承它。
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Integer, DateTime, Text, func class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(64), nullable=False, index=True) email: Mapped[str] = mapped_column(String(128), nullable=False, unique=True) bio: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())注意这里的类型注解Mapped[int]和mapped_column(Integer, ...)不是重复劳动。Mapped[int]让IDE能识别出user.id是int,写代码时有类型提示;mapped_column里的Integer则是告诉SQLAlchemy建表时用哪种数据库类型。两个东西各司其职,一个管Python侧的类型检查,一个管数据库侧的DDL生成。
在实际项目里,我习惯把Base放在单独的db/base.py里,所有模型文件都只从那里导入Base。这样后续做迁移脚本、模型自动发现都方便,不会出现循环导入。我曾经见过有人把Base定义在某个模型文件里,其他模型去import,结果牵一发动全身,改一个文件报一串循环依赖错误。
2.2 字段类型、约束与索引选择的实战要点
字段类型的选择直接决定数据库层面的存储效率和查询性能,这块值得仔细说。String一定要带上长度,虽然PostgreSQL的varchar可以不写长度,但MySQL里varchar不写长度会直接建表报错,SQLite则默认按无长度处理。为了让代码在不同数据库间最大程度可移植,所有String都写长度是习惯。
nullable和server_default这两个参数经常被搞混。nullable=True只是说这一列可以不填值,插入时数据库会存NULL;server_default是指数据库层面提供一个默认值,比如server_default=func.now()会让数据库在插入时自动写入当前时间,而不是Python这边生成时间再传过去。这两者的区别在批量插入场景里尤其明显:如果你用Python侧生成默认值,每一条记录都要在应用层多一步计算;用server_default,数据库闷头一插就完事了。
索引是查询性能的命门,但不是所有列都该建索引。我见过有人把每列都加上index=True,结果写多读少的表被索引拖垮了写入性能。通常来说,WHERE子句里高频出现的过滤列、外键列、需要ORDER BY或GROUP BY的列才值得建索引。唯一约束unique=True本身就会创建唯一索引,不需要重复指定index=True。
2.3 迁移工具:没有Alembic的ORM是不完整的
直接用Base.metadata.create_all(engine)建表只适合开发环境,生产环境你几乎一定会面临表结构变更:加字段、改类型、加索引、拆表、合并表。这时候没有迁移工具等于裸奔。Alembic是SQLAlchemy官方推荐的迁移工具,它的工作方式是把“当前模型定义”和“数据库当前结构”做对比,生成迁移脚本。
我的一般流程是:模型改完,执行alembic revision --autogenerate -m "add user bio",Alembic会生成一个迁移文件,里面包含upgrade()和downgrade()两个函数。先仔细读一遍自动生成的脚本,确认没有误判(比如把类型长度变化错当成新增列),再执行alembic upgrade head。自动生成不等于自动正确,这一步人工review必须做。
迁移这块我踩过一个具体的坑:某次给一张大表加了非空字段,但没有给server_default,Alembic生成的迁移脚本是ALTER TABLE ... ADD COLUMN xxx NOT NULL。在开发环境表是空的,一下就过了;生产环境里这张表有几十万行,这个ALTER直接锁表跑了十几分钟,差点把库拖死。后来学乖了,给大表加非空列一定分三步:先加可空列、填默认值、再收紧为非空。这种教训写不进任何文档,只能靠真实生产环境给你上一课。
3. 关系映射:让对象之间像代码一样协作
3.1 一对多与多对一
“用户有好多篇文章”是标准的一对多。在ORM里,User模型加一个articles属性,Article模型加一个user外键和user关系,两边互相能访问。
from sqlalchemy.orm import relationship, Mapped, mapped_column from sqlalchemy import ForeignKey, Integer, String, Text class User(Base): # ...上面的字段 articles: Mapped[list["Article"]] = relationship(back_populates="user") class Article(Base): __tablename__ = "articles" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, index=True) title: Mapped[str] = mapped_column(String(200), nullable=False) content: Mapped[str] = mapped_column(Text, nullable=False) user: Mapped["User"] = relationship(back_populates="articles")这里有几个值得反复琢磨的细节。ForeignKey("users.id")里写的是“表名.列名”,不是“类名.列名”,注意区分。relationship则是纯Python层面的导航属性,它告诉ORM怎么在内存里把两个对象串起来。back_populates两边互相指定,避免ORM在内存里维护出两份不一致的关系副本。你也可以只在一侧写relationship并用backref让另一侧自动生成,但显式的back_populates更容易让新人看出关系全貌,我推荐显式写法。
外键列一定要建索引,这和解引用查询性能直接相关。两个表关联查询时,如果外键列没有索引,数据库要做全表扫描逐行匹配,数据量一大就卡成幻灯片。这个索引通常叫ix_articles_user_id,你要么在mapped_column里直接加index=True,要么在Alembic迁移脚本里手工建,总之不能漏。
3.2 多对多:必须理解关联表的含义
文章和标签是多对多关系,因为一篇文章可以有多个标签,一个标签也对应多篇文章。关系型数据库里表达多对多必须引入第三张“关联表”,ORM里就是两个relationship配合一个中间表模型。
from sqlalchemy import Table, Column, ForeignKey article_tags = Table( "article_tags", Base.metadata, Column("article_id", ForeignKey("articles.id"), primary_key=True), Column("tag_id", ForeignKey("tags.id"), primary_key=True), ) class Tag(Base): __tablename__ = "tags" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(32), nullable=False, unique=True) class Article(Base): # ...上面的字段 tags: Mapped[list["Tag"]] = relationship(secondary=article_tags, back_populates="articles")注意article_tags表没有单独的模型类,只定义了一个Table对象,这种写法适合“纯关联表”场景——如果关联表本身还需要额外字段(比如“用户收藏文章”的时间、备注),那就需要升级为完整的模型类,用Association Object模式。
多对多关系里最容易踩的坑是重复插入关联记录。你用article.tags.append(tag)之后,如果tag已经存在于数据库,ORM一般能识别,但有些边缘情况下会出现IntegrityError,因为关联表的主键是(article_id, tag_id)联合主键,重复插入会撞唯一约束。我习惯在append前先查询一次标签是否存在,或者直接借助唯一索引让数据库兜底报错再捕获。
3.3 级联行为的取舍:别让delete变成灾难
relationship里的cascade参数控制“父对象操作时子对象怎么联动”。默认情况下,删除一个User不会自动删除他的Article,因为外键约束会阻止你删除仍有子记录引用的父记录。很多初学者指望cascade="all, delete-orphan"能帮忙,并随手把Article的user关系也配成同样的cascade,结果删一个用户连带把几万篇文章全删了,追悔莫及。
我的建议是:绝大多数场景下,业务数据的删除应该是“软删除”(加一个is_deleted字段)或者“手动显式删子记录”,而不是把级联删配置在ORM里。只有当子记录的生命周期完全由父记录决定、没有任何独立存在的意义时(比如一篇草稿的图片附件),才适合用cascade="all, delete-orphan"。级联删除是那种“配置一时爽、出事火葬场”的功能,务必谨慎。
4. Session与事务:ORM的核心引擎
4.1 Session的生命周期:它不是连接池,也不是事务本身
Session是SQLAlchemy ORM里最容易被误解的对象。很多人以为Session就是数据库连接,或者把它当成事务,其实都不准确。Session是一个“工作单元”容器,它持有数据库连接(通过底层的Engine获取),维持一个身份映射,追踪所有对象的状态变化,并在合适的时机把变更同步回数据库。
我常用的Session管理方式是:
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine engine = create_engine("postgresql+psycopg2://user:pass@localhost/mydb", pool_size=10, max_overflow=20) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)sessionmaker是一个工厂,每次调用SessionLocal()都会生成一个全新的Session实例。autoflush=False很重要——它让你能控制什么时候把内存里的变更刷到数据库;autocommit=False则是说必须显式调用session.commit()才算真正提交事务。
最核心的使用原则是:一个Session对应一个业务操作单元,操作完就关闭。在Web应用里,常见模式是“请求进来创建一个Session、请求结束关闭Session”,配合依赖注入框架(比如FastAPI的Depends)非常顺。但在普通脚本里,很多人的错误是创建一个全局Session一直用,用完之后也不关,最后连接全部泄漏。我自己的脚本里永远用with或try/finally包裹Session,确保无论成功失败都会close()。
4.2 事务边界:Commit、Rollback与异常处理
Session的commit()提交事务,rollback()回滚事务。但你要搞清楚,事务的实际起点不是Session创建时,而是第一条SQL执行时(更准确说,是底层DBAPI连接第一次被使用时)。所以你可以先创建Session、往里面加几个对象,再执行查询,事务才真正开始。
事务边界的核心原则是“一个业务用例对应一个事务”。例如“创建用户并发送欢迎邮件”,创建用户和发送邮件如果放在同一事务里,邮件服务再慢也得等数据库事务完成,代价极高;如果分开,发邮件失败时用户已经创建了,又得补偿。这类问题没有标准答案,要按业务一致性要求来权衡。但无论怎么选,Session的提交/回滚必须在明确的位置,不能让异常穿透到不确定的地方。
下面是我写的通用事务处理模板:
from sqlalchemy.exc import SQLAlchemyError def create_user_with_profile(name: str, email: str) -> User: session = SessionLocal() try: user = User(name=name, email=email) session.add(user) session.flush() # 提前拿到user.id profile = Profile(user_id=user.id, avatar_url="/default.png") session.add(profile) session.commit() return user except SQLAlchemyError: session.rollback() raise finally: session.close()这里session.flush()是我刻意加的,目的是让INSERT语句先执行,使user.id被数据库生成并回填到对象上,这样后面构造Profile时能拿到外键值。flush和commit的区别:flush只是把SQL发给数据库,事务还没提交;commit提交事务,同时也会隐式flush。如果不调用flush直接等到commit,那也没问题,只是你需要通过session.refresh(user)或者预期user.id为None之前拿到它,就会很别扭。
还有一个细节:IntegrityError(比如唯一约束冲突)发生后,Session会进入“脏”状态,此时你必须rollback()才能继续使用。不rollback而继续执行其他查询,SQLAlchemy会抛PendingRollbackError。我踩过这个坑——某个批处理脚本里前一条记录撞了唯一约束,我忽略了,继续处理下一条,结果整个循环崩掉。正确做法是每批操作要么各自开Session,要么在异常后立刻rollback重新开始。
4.3 身份映射:为什么同一个对象是同一个引用
Session内部有一个身份映射(Identity Map),它保证在同一Session内,按主键查询同一个对象多次,你拿到的始终是同一个Python对象引用。这个特性带来一个很有意思的行为:
user1 = session.get(User, 1) user2 = session.get(User, 1) user1.name = "新名字" print(user2.name) # 输出: 新名字这不是bug,而是ORM刻意设计的。它避免了同一个Session里出现同一行数据的两个副本,从而防止更新时互相覆盖。但这种特性也容易让人困惑:如果Session生命周期很长,你对某个对象做的修改会一直留着,直到提交或回滚;如果另一个进程/线程改了数据库里的同一行,你的Session里缓存的对象还是旧的,你需要在适当时候用session.expire_all()或session.refresh(obj)来重新加载。
4.4 显式地控制加载策略:默认的懒加载正在等你踩坑
SQLAlchemy默认使用懒加载(lazy loading):访问user.articles时,如果articles还没加载,它会立刻执行一条SQL去查。这在单个对象访问时很方便,但在循环里会变成灾难。
for user in session.scalars(select(User).limit(100)): for article in user.articles: print(article.title)这段代码会产生100次甚至更多次查询——每次访问user.articles都要查一次数据库,这就是经典的N+1问题。解决办法是在初始查询时就用selectinload或joinedload显式预加载。
from sqlalchemy.orm import selectinload users = session.scalars(select(User).options(selectinload(User.articles)).limit(100)).all() for user in users: for article in user.articles: print(article.title)selectinload会先查用户列表,再一次性查出所有关联的articles,用IN查询关联回主表。相比joinedload(JOIN方式),selectinload避免了大结果集的笛卡尔积膨胀,尤其在分页场景下更安全。选择哪种,看关联集合的行数,小集合JOIN没问题,大集合优先selectinload。
5. 查询进阶:从入门到用得顺手
5.1 构建查询的三种风格:Query vs Select
SQLAlchemy 2.0开始,官方推荐用select()函数加session.scalars()的风格,替代老旧的session.query(User)风格。新风格更贴近SQL表达式树的构造逻辑,也更便于类型推断。
from sqlalchemy import select, and_, or_, desc stmt = select(User).where( and_(User.name.like("%张%"), User.email.is_not(None)) ).order_by(desc(User.created_at)).limit(20) users = session.scalars(stmt).all()关键区别在于session.scalars()返回的是ScalarResult,迭代它的元素是模型对象本身;而session.execute()返回的是Result,每行是Row对象,元素位置对应select的列。如果你select的是单个模型,scalars()最省事;如果select的是多列或聚合结果,用execute()拿Row。
新风格的另一个好处是组合性更强。你可以先构造stmt = select(User),后面在不同分支里动态追加.where()、.order_by()、.offset(),最终生成一个完整查询。这在复杂搜索条件场景下特别灵活,不用像老式Query对象那样做各种拼接。
5.2 过滤、排序、分页与聚合的常见写法
分页是个高频需求。传统方式是limit/offset,但数据量大了以后offset越来越大,数据库要跳过前面所有行才能拿到目标数据,性能下降飞快。此时应该用“键集分页”(keyset pagination),也就是基于排序字段做条件过滤:
# 传统limit/offset stmt = select(User).order_by(User.id).limit(20).offset(40) # 键集分页:上一页最后一条的id是40 stmt = select(User).where(User.id > 40).order_by(User.id).limit(20)键集分页的SQL执行计划稳定,不管翻到第几页都很快,但缺点是你不能直接跳页,只能“下一页/上一页”式浏览。多数Web场景是够用的。
聚合查询方面,SQLAlchemy提供了func.count、func.sum、func.avg等工具,配合group_by可以完成复杂的统计。我们统计“每个用户的文章总数”:
from sqlalchemy import func stmt = select(User.name, func.count(Article.id)).join(Article, Article.user_id == User.id).group_by(User.id, User.name) results = session.execute(stmt).all()这里要注意的是,SQL标准要求GROUP BY的选择列之外,select的普通列都必须在聚合函数里或group by里。PostgreSQL有宽松模式,MySQL的ONLY_FULL_GROUP_BY模式下就会报错。所以我在写聚合时总是把非聚合列全部放进group_by,避免方言差异。
5.3 原生SQL与ORM的混合:什么时候该放弃ORM
ORM不是万能的,复杂报表查询、多表大JOIN、窗口函数、递归查询等场景,硬套ORM会很痛苦。SQLAlchemy没有强迫你全程使用ORM——session.execute(text("..."))可以直接执行原生SQL,结果也可以通过Result对象取用。
from sqlalchemy import text rows = session.execute(text("SELECT date_trunc('month', created_at) AS month, count(*) AS cnt FROM users GROUP BY 1")).all() for row in rows: print(row.month, row.cnt)我的经验是:核心业务CRUD用ORM,复杂报表查询用原生SQL,中间地带用SQLAlchemy Core的表达式构造(比如select函数不带ORM实体)。这三者可以在同一个Session里共存,这也是SQLAlchemy真正强大之处——它不是搏命一刀切的框架,而是一个分层工具集。
5.4 懒加载的坑与解决方案
除了N+1问题,还有一个容易踩的坑是“Session关闭后访问未加载关系”。SQLAlchemy的默认行为是:Session关闭后,访问未加载的lazy属性会抛DetachedInstanceError。这经常出现在Web应用里——你在视图函数里查了对象,把它传给模板或序列化器,等模板渲染时Session已经关了,再访问user.articles就炸了。
解决方案无非两种:要么在Session还开着的时候主动预加载(上面说的selectinload),要么用expire_on_commit=False让对象在commit后保持“可用状态”。但我强调一句,expire_on_commit=False只是缓解问题,不是根治。如果你把一个对象传到Session生命周期之外还频繁访问它的懒加载属性,迟早还是要踩坑。本质上前置加载、明确访问路径,才是正路。
6. 常见问题与排查技巧实录
6.1 连接泄漏与连接池配置
连接泄漏是我在帮助别人排查SQLAlchemy问题时遇到最多的问题。它的经典症状是:应用跑一段时间后越来越慢,数据库连接数持续上升,直到耗尽。原因几乎都是Session没有正确关闭。
排查这个问题的思路分两步。第一步,检查代码路径,确保每个SessionLocal()都有对应的close();最佳方案是用上下文管理器或者FastAPI/Flask的依赖注入容器统一管理生命周期,而不是在各个函数里手动创建关闭。第二步,开启连接池的回收机制:
engine = create_engine( "postgresql+psycopg2://user:pass@localhost/mydb", pool_size=10, max_overflow=20, pool_pre_ping=True, pool_recycle=1800, )pool_pre_ping=True会在每次从连接池拿连接之前先发送一个轻量的ping,确保这个连接没有被数据库服务端断开;pool_recycle=1800表示连接最多存活30分钟就被强制回收重建,避免数据库侧的wait_timeout把空闲连接掐断。这两项配置在长期运行的Web服务里几乎是必开的。
6.2 N+1查询的定位与修复
N+1查询的定位方法很简单:开启SQLAlchemy的日志或者用第三方工具记录实际执行的SQL条数。你肉眼扫一眼日志,如果一次业务请求打了几百条SQL,基本可以断定是N+1。
修复N+1的方向就两个:一是用selectinload/joinedload预加载,二是在查询时直接“反着查”——先查你真正要输出的主体表,再用子查询批量带出关联数据。具体用哪个,要看你的数据访问模式。如果总是访问一对多关系的“多”那侧,selectinload几乎总是安全选择;如果经常需要按关联表的字段过滤,那么先JOIN再distinct也可能更合适。
我特别提醒一点:不要对relationship设置lazy="select"之外的“全局预加载”选项,除非你确定所有访问路径都需要预加载。我见过有人在relationship里写lazy="joined",结果每查一个用户都强制JOIN出他的所有文章,哪怕只是做个用户名列表,性能惨不忍睹。预加载应该是查询级别的、按需开启的,而不是模型级别的一刀切。
6.3 事务超时与死锁的处理
死锁在ORM代码里常常是因为两个事务以不同顺序申请同一组资源。比如事务A先更新用户再更新文章,事务B先更新文章再更新用户,某次并发下就可能互相等待。SQLAlchemy不会帮你解决死锁,它只是在检测到数据库层面的死锁后抛出OperationalError。
处理策略有几个方向:所有涉及到多行更新的业务,统一固定按某种顺序(比如按主键从小到大)操作;必要的时候用SELECT FOR UPDATE锁定行,但注意只锁必要的行;另外,缩短事务时间也能显著降低死锁概率——不要在事务里做网络请求、文件读写等耗时操作。
6.4 常见错误速查表
| 错误类型 | 典型报错 | 原因与解法 |
|---|---|---|
| 会话状态异常 | PendingRollbackError | 上一个异常后未rollback,捕获异常后必须显式rollback再继续 |
| 对象失效 | DetachedInstanceError | Session关闭后访问懒加载属性,改用selectinload预加载或调整生命周期 |
| 唯一约束冲突 | IntegrityError | 数据重复插入,捕获后rollback并做去重或换个业务策略 |
| 类型不匹配 | StatementError | 查询参数类型与列类型不一致,检查Mapped注解与mapped_column定义 |
| 连接耗尽 | TimeoutError/QueuePool limit | Session未关闭,排查连接泄漏并配置pool_pre_ping和pool_recycle |
| 锁超时 | LockNotAvailable/deadlock detected | 事务持锁时间过长或死锁,缩短事务、固定更新顺序 |
这张表我建议你收藏,写代码的时候碰到对应报错可以快速定位,不用每次从头搜。
6.5 性能排查的实战思路
面对一个慢的ORM查询,我一般的排查路径是:先看最终转换出来的SQL长什么样,其次看有没有多余的查询,最后再看数据库执行计划。拿到查询的编译后SQL很简单:
stmt = select(Article).where(Article.user_id == 1) print(stmt.compile(compile_kwargs={"literal_binds": True}))这个打印出来的SQL可以直接扔进psql里执行,然后EXPLAIN ANALYZE看执行计划。大部分慢查询的解决方案无非三种:加索引、改查询结构、控制预加载范围。工具和手段就那么几个,复杂的其实是对业务数据分布的理解——哪张表数据量大、哪个字段区分度高、哪些查询是真的高频。
写在最后
这篇文章的内容基本来自我在各种项目里反复折腾SQLAlchemy的经验沉淀。说实话,ORM这个东西,写起来容易,写好很难。SQLAlchemy尤其如此,它的灵活性和功能深度是一把双刃剑:用得好,数据访问层变得干净整齐;用得不好,各种隐式行为会让代码变得难以预测。
我个人最大的感受是:不要把SQLAlchemy当成一种帮你“逃避SQL”的工具,而要把它当成一种“把SQL化的思维融入Python工程”的工具。当你理解了Session的作用边界、关系加载的时机、事务的生命周期,你会发现它不是在代替你思考数据库,而是在帮你把底层细节封装成更可控的抽象,让你有精力去思考真正复杂的业务问题。
如果你刚开始用SQLAlchemy,建议你先跑通一套最小的CRUD,然后按本文的顺序逐步往里加关系、加事务、加预加载、加迁移。别一次性引入所有特性,每加一层都搞清楚它背后的原理。这样踩过的坑少,写出来的代码也会扎实很多。