如果你写 Python 已经一阵子了大概率会撞上这堵墙项目逻辑越来越复杂数据表越来越多手写 SQL 越来越像家务活。我当年也一样列表页要拼条件、插入记录要拼括号、联表查询要反复确认字段名好不容易跑通改需求时又得回去翻十几条 SQL 字符串。后来切到 SQLAlchemy 配合 MySQL用 ORM 的方式重新组织数据层整个项目的维护成本直接降了一个台阶。这篇东西不是什么高深理论就是我从 0 到 1 把 Python SQLAlchemy MySQL 这套组合落地的一手记录。内容包括环境怎么搭、模型怎么建、CRUD 怎么写、查询怎么避免性能坑最后还会分享我改造老项目原生 SQL 时踩过的真实问题。适合刚接触 ORM 的 Python 开发者也适合那些已经会用 SQL 但还没下定决心切 ORM 的朋友参考。1. 先从为什么用 ORM说起一段手写 SQL 的体验1.1 手写 SQL 半年后我决定换条路在切换到 SQLAlchemy 之前我维护过一个内部管理系统数据层全部是 pymysql 加字符串拼 SQL。听起来并不可怕真正跑起来才难受。最典型的场景是列表筛选。用户在前端勾几个筛选条件后端就要动态拼WHERE。我的代码长这样sql SELECT id, name, age, email FROM users WHERE 11 if name: sql AND name LIKE %s params.append(f%{name}%) if age_min: sql AND age %s params.append(age_min) ...写一次无所谓写多了你会发现同样一段逻辑散落在各个业务模块里改一个字段名要全局搜索替换而且容易漏。更麻烦的是联表场景JOIN一多查询结果里重名字段、类型转换、Null 处理都会变成隐性地雷。再加上连接管理、事务提交、异常回滚这些样板代码真正写业务的时间可能连一半都不到。ORM 解决的正是这一类问题把表结构映射成 Python 类把行记录映射成对象把常见的增删改查封装成方法把条件拼接改成方法链或查询表达式。它不是说 SQL 不重要而是让你不必在 80% 的常规操作上重复劳动把精力留给真正复杂的查询和性能优化。1.2 ORM 到底解决了什么以及它不解决什么很多人对 ORM 的印象是自动生成 SQL这个说法没错但不完整。SQLAlchemy 的定位更准确它是一个 SQL 工具包加对象关系映射器。前半句意味着你随时可以写原生 SQL后半句意味着常规操作可以完全面向对象。用 ORM 的好处我体感最明显的三件事字段定义集中管理。数据表的列在模型类里一眼看全改表结构时改一处所有用到模型的地方同步生效不用到处找字符串。关系查询变得直观。user.posts这种写法让我不用每次手写JOIN代码读起来接近自然语言。数据库差异被隔离。SQLAlchemy 的方言机制让同一套模型可以跑在 MySQL、SQLite、PostgreSQL 上本地测试用 SQLite生产用 MySQL几乎不用改业务代码。但 ORM 不是万能药。它不适合重度报表类查询那种几十行 JOIN 加子查询加窗口函数的 SQL直接写原生语句反而更清晰。SQLAlchemy 也支持这种组合拳你可以在 ORM 里用text()写原生 SQL结果照样映射成对象。所以我们不用纠结用 ORM 还是写 SQL而应该是能用 ORM 表达的常规操作用 ORM复杂查询和批量操作用 SQL两者相辅相成。2. 环境搭建Python、MySQL、SQLAlchemy 的版本坑位图2.1 版本选择与安装路线这套组合里最容易出问题的不是 SQLAlchemy而是 MySQL 的驱动和认证插件。我先说结论再解释为什么。推荐组合Python 3.10 以上MySQL 8.0SQLAlchemy 2.0 系列驱动用 PyMySQL另外装一个cryptography库。安装命令很简单pip install sqlalchemy pymysql cryptography很多人装完 PyMySQL 后连接 MySQL 8.0 报错提示Authentication plugin caching_sha2_password cannot be loaded。这是因为 MySQL 8.0 默认的认证插件是caching_sha2_password老版本的 PyMySQL 或者某些中间件不支持。解决方案有两个方向一是升级 PyMySQL 并安装cryptography库让驱动支持新的认证方式二是在 MySQL 里把用户的认证插件改回mysql_native_password但这不是长久之计新项目建议直接走第一种。如果你是在 Windows 本机装 MySQL记住安装时一路 Next 容易踩坑选 Server only 还是 Developer Default 看你的需求但字符集一定要选utf8mb4或者装完后在 my.ini 里配置。这点后文会单独说。2.2 验证安装与连接串配置装好之后先不用急着写模型用最小代码验证连通性from sqlalchemy import create_engine, text engine create_engine( mysqlpymysql://root:yourpassword127.0.0.1:3306/test_db?charsetutf8mb4 ) with engine.connect() as conn: result conn.execute(text(SELECT 1)) print(result.scalar())这里解释一下连接串的结构dialectdriver://username:passwordhost:port/database?参数拆开来看就是mysqlpymysql告诉 SQLAlchemy 使用 MySQL 方言并且通过 PyMySQL 驱动访问。root:yourpassword用户名和密码。127.0.0.1:3306地址和端口。test_db目标数据库需要先手动创建。charsetutf8mb4字符集参数这个非常关键。utf8mb4是 MySQL 上真正意义上的完整 UTF-8编码支持 emoji 和生僻字。MySQL 默认的utf8实际上最多只存 3 字节遇到 4 字节字符会报错或乱码。所以连接串里写utf8mb4建表时也统一用utf8mb4这是我从乱码坑里爬出来的经验。2.3 最小可运行 Demoengine 与 connectionSQLAlchemy 里有两个容易混淆的概念Engine 和 Connection。Engine 是全局唯一的数据库引擎对象负责维护连接池、方言解析和 SQL 编译。整个应用生命周期里一般只创建一次放在模块顶部复用。Connection 是实际连接到数据库的会话资源用完要释放。我推荐的日常模式是这样的from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine create_engine( mysqlpymysql://root:yourpassword127.0.0.1:3306/test_db?charsetutf8mb4, pool_pre_pingTrue, pool_recycle3600, echoFalse, )三个参数是实战经验pool_pre_pingTrue连接池里的连接在每次使用前先发送一次探测相当于 SELECT 1如果发现连接已经断开就自动剔除并新建。MySQL 默认wait_timeout是 8 小时连接闲置超过这个时间会被服务端关闭没有这个参数就会报MySQL server has gone away。pool_recycle3600强制让连接最多存活 3600 秒就回收重建进一步避免拿到失效连接。echoFalse开启会打印所有 SQL 日志开发调试时设为 True 很爽生产环境务必关掉。3. 第一个模型从数据表到 Python 类的映射3.1 声明式基类与字段映射SQLAlchemy 2.0 的声明式写法已经很成熟我直接用它做示例。核心是先定义一个继承自DeclarativeBase的基类然后所有模型类继承它。from sqlalchemy import String, Integer from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class User(Base): __tablename__ users id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(64), nullableFalse, uniqueTrue) email: Mapped[str] mapped_column(String(128), nullableFalse, default) age: Mapped[int] mapped_column(Integer, nullableFalse, default0)这里有几个在 MySQL 场景下需要特别注意的细节Integer映射到 MySQL 是INT。如果数据量很大主键建议显式用BigInteger。String(64)映射为VARCHAR(64)给长度是为了索引效率和存储优化不要无脑给 255尤其在utf8mb4下一个字符最多占 4 字节长 VARCHAR 会让磁盘和内存消耗变大。nullable和default的含义要区分清楚nullableFalse是数据库层面的非空约束default0是 Python 层面的默认值它不会直接改变数据库表结构里的DEFAULT。如果你想让 MySQL 表结构本身也带默认值要写成server_defaulttext(0)。这是新手经常被坑的点模型里给了default客户端不传字段时确实会用默认值但如果直接用原生 SQL 插入数据默认值就不一定生效。3.2 建表与改表create_all 的现实操作建表最简单的方式是直接在主代码里Base.metadata.create_all(engine)它只会创建数据库中不存在的表不会修改已存在的表。所以开发阶段改模型字段后跑create_all不会自动加列你需要先删除旧表再重建或者用迁移工具。我在这件事上的建议非常明确任何要从开发走向上线的项目尽早引入 Alembic 做迁移管理不要依赖drop_all和create_all那一套。Alembic 是 SQLAlchemy 官方的迁移工具它能生成版本化的迁移脚本支持在已有表上安全地加列、改类型、加索引。等你生产库有数据之后再想 删库重建 就晚了。如果只是想快速验证模型可以这样做Base.metadata.drop_all(engine) Base.metadata.create_all(engine)仅限本地开发我每次跑都会再三确认连的是不是本地库这种命令误连生产库的教训在网上一抓一大把。3.3 一对多与多对多关系声明数据表之间的关联是 ORM 的精髓。先看最常用的一对多一个用户有多篇文章。class Post(Base): __tablename__ posts id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200), nullableFalse) user_id: Mapped[int] mapped_column(ForeignKey(users.id), nullableFalse) author: Mapped[User] relationship(back_populatesposts)同时在 User 类里补上posts: Mapped[list[Post]] relationship(back_populatesauthor)ForeignKey(users.id)是数据库层面的外键约束relationship是 ORM 层面的导航属性。两者配合才能做到user.posts这种对象式访问。多对多关系需要一张中间关联表。假设文章和标签是多对多from sqlalchemy import Table, Column post_tags Table( post_tags, Base.metadata, Column(post_id, ForeignKey(posts.id), primary_keyTrue), Column(tag_id, ForeignKey(tags.id), primary_keyTrue), ) class Tag(Base): __tablename__ tags id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), nullableFalse, uniqueTrue)然后在 Post 和 Tag 里分别加# Post 里 tags: Mapped[list[Tag]] relationship(secondarypost_tags, back_populatesposts) # Tag 里 posts: Mapped[list[Post]] relationship(secondarypost_tags, back_populatestags)核心在于secondarypost_tags它告诉 SQLAlchemy 这张中间表是纯关联表不需要单独的模型类。多对多查询时ORM 会自动生成跨两张表的 JOIN。关系这个功能我之前觉得可有可无直到接手一个接口返回多层嵌套 JSON 的项目时才发现没有 ORM 关系的话每层都要手写联表查询字段一多很容易漏。而这种声明式关系配上序列化辅助工具代码量和出错率都明显下降。4. Session 与 CRUD真正决定每天编码体验的环节4.1 session 的创建与生命周期模型定义了表结构但是真正跟数据库打交道要靠 Session。Session 可以理解为一个工作单元它跟踪你在本次操作中加载和修改的所有对象直到commit()才把变更一次性提交到数据库。我推荐用sessionmaker创建一个工厂函数然后在每个请求或业务处理里用上下文管理器的方式获取 Sessionfrom sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine, expire_on_commitFalse) def get_db(): with SessionLocal() as session: yield session注意expire_on_commitFalse这个参数它直接影响你在commit()之后还能不能继续访问对象的属性。默认情况下commit()之后会话会把所有对象的属性标记为过期下次访问属性时会重新发 SQL 查询。这会导致一个很常见的困扰我在commit()之后想读取user.id结果触发了一次额外的数据库查询如果 Session 已经关闭就直接报DetachedInstanceError。把它设为 Falsecommit()之后对象保持原样省心很多。Session 的生命周期一定要短用完就关。很多人写脚本时创建了一个全局 Session跑完不关程序一直不退出连接池很快被占满。用with SessionLocal() as session的语法退出代码块时 Session 会自动关闭这也是我最推荐的方式。4.2 增删改查的标准姿势插入数据with SessionLocal() as session: user User(nametom, emailtomexample.com, age25) session.add(user) session.commit() print(user.id) # 提交后自增主键已写回对象批量插入用add_allwith SessionLocal() as session: session.add_all([ User(namea, emailaexample.com), User(nameb, emailbexample.com), ]) session.commit()查询数据with SessionLocal() as session: # 查询单个对象按条件 user session.scalars(select(User).where(User.name tom)).first() # 查询所有用户按年龄倒序 users session.scalars(select(User).order_by(User.age.desc())).all()SQLAlchemy 2.0 的查询风格统一成select()不再推荐旧的session.query()写法。虽然旧的还能用但新项目建议直接学新的免得看文档时精神分裂。更新数据有两种方式。第一种是拿到对象后直接改属性Session 会帮你跟踪with SessionLocal() as session: user session.scalars(select(User).where(User.id 1)).first() if user: user.age 30 session.commit()第二种是批量更新用update()方法避免把所有数据加载到内存from sqlalchemy import update with SessionLocal() as session: session.execute( update(User).where(User.age 18).values(statusminor) ) session.commit()删除数据同理from sqlalchemy import delete with SessionLocal() as session: session.execute(delete(User).where(User.id 999)) session.commit()二维表操作如update和delete不经过 ORM 对象缓存性能好但也不会触发 ORM 层的级联操作使用时要自己判断是否需要同时清理关联数据。4.3 事务提交与回滚什么时候该 commit什么时候该 rollback事务是数据库一致性的根基。在 SQLAlchemy 中Session 默认开启事务commit()是事务结束点rollback()则放弃本次所有变更。我踩过一个印象很深的坑批量处理任务时循环里每条数据都commit()一次结果中间某条失败前面已提交的数据无法回滚数据处于改了一半的状态。正确做法是尽量把一批操作放进同一个事务最后统一提交with SessionLocal() as session: try: for data in task_list: session.add(User(**data)) session.commit() except Exception: session.rollback() raise但事务又不宜过大。我见过有人把几万条记录的导入放进一个事务里跑了几分钟还没结束MySQL 锁冲突、undo log 膨胀、内存飙升全来了。合理的做法是分批提交每批 500 到 1000 条即使失败也只需要回滚当前批次。这个经验在数据清洗、迁移、定时任务里特别重要。5. 查询层面最容易踩的坑N1、分页、连接池5.1 懒加载引发的 N1 问题以及两种解法ORM 最方便的是关系导航最坑的也是关系导航。默认情况下访问user.posts时 SQLAlchemy 会立刻发一条查询去拿该用户的所有文章。如果循环里先查出 100 个用户再逐个访问user.posts就会变成 1 条主查询加 100 条附属查询这就是经典的 N1 问题。# 反例会产生 1 N 条 SQL with SessionLocal() as session: users session.scalars(select(User)).all() for user in users: print(user.posts) # 每次访问都发一条新 SQL解法一是使用selectinload或joinedload主动预加载关系from sqlalchemy.orm import selectinload with SessionLocal() as session: users session.scalars( select(User).options(selectinload(User.posts)) ).all() for user in users: print(user.posts) # 不会产生额外 SQL我自己更常用selectinload它的原理是先加载用户列表再生成一条WHERE post.user_id IN (...)的查询把关联数据一次性取回再按内存中的外键把对象关联好。joinedload则是在原 SQL 上直接 JOIN如果主表数据量大且每行关联数据也多结果集会成倍膨胀。大多数业务场景selectinload更稳。连接没有预加载时如果你知道自己接下来要访问关系但不想改原查询也可以用session.refresh(user)配合指定属性但日常开发还是建议在查询时就把加载策略写清楚不要依赖运行时补救。5.2 分页、过滤与排序的正确姿势分页最基础的方式是limit加offsetusers session.scalars( select(User).order_by(User.id).limit(20).offset(40) ).all()在数据量小的后台管理中够用。但数据量超过几十万行时OFFSET会导致 MySQL 扫描并丢弃前 N 行页码越深性能越差。这时候我会改用键集分页利用上次查询的最后一条记录位置继续往后翻last_id 40 users session.scalars( select(User).where(User.id last_id).order_by(User.id).limit(20) ).all()这种方式的查询条件能直接走主键索引无论翻到第几页耗时都差不多。缺点是无法随意跳页适合瀑布流或加载更多场景。动态过滤条件的经验是用where()方法逐步追加条件比手拼字符串安全得多字段类型和转义都由 ORM 处理。排序方面order_by接受列对象支持多字段组合例如先按状态分组再按时间排序select(Order).where(Order.status paid).order_by(Order.created_at.desc(), Order.id.desc())注意 MySQL 对 NULL 值的排序是升序在最前、降序在最后如果你需要自定义 NULL 位置要用nullslast()或nullsfirst()在 SQLAlchemy 里也有对应的函数。这个边界条件很容易在测试时漏掉。5.3 连接池、并发与会话清理前面提到create_engine自带连接池。默认池大小是 5最大溢出是 10也就是说并发超过 15 个连接请求时后来的请求会等待。小应用一般够用但接口并发上来了以后会遇到连接等待超时。按我的经验可以从两个方向调一是加大连接池二是排查是否有连接泄漏。连接泄漏的典型表现是运行一天后数据库连接数持续上涨最终报Too many connections。常见原因就是 Session 没关闭或者线程里创建了 Session 没有释放。调参参考engine create_engine( url, pool_size10, max_overflow20, pool_pre_pingTrue, pool_recycle3600, )线上 MySQL 的max_connections默认是 151连接池总大小要留有余量别把数据库连接池顶满。除了调参数我更想强调代码习惯Session 一定要遵循使用即关闭的原则在 Web 应用里最好通过依赖注入或中间件确保每个请求结束时关闭连接。并发写入场景还有个容易忽视的问题MySQL 默认隔离级别是可重复读但在高并发更新同一行时可能出现更新丢失。SQLAlchemy 层面没有自动处理乐观锁你需要自己加版本号字段或者用FOR UPDATE做悲观锁。这个话题展开能写一整篇这里只提醒一点使用 ORM 不代表并发问题自动消失事务边界和锁策略依然必须自己把握。6. 从 pymysql 裸 SQL 迁移到 SQLAlchemy一次真实的老项目改造记录6.1 迁移第一步不要推倒重来改造老项目最大的风险在于想一口吃成胖子。我当时的习惯是先搭好模型映射但不急着替换业务逻辑。流程是这样的先从现有表结构反推模型。如果表已经存在我不会用create_all而是直接根据表字段写模型类。写完以后先写一段对照脚本分别用原 SQL 和 ORM 跑同一条查询比对结果是否一致。我当时连的是一张订单表有 30 多个字段还有三张关联表。反推模型时特别注意了字段类型MySQL 的DECIMAL对应 SQLAlchemy 的Numeric或DecimalDATETIME对应DateTimeTINYINT(1)默认对应SmallInteger而不是布尔这些映射关系不一致会导致数据读写出现类型偏差。对照校验阶段我用得最多的调试手段是把echoTrue打开让 SQLAlchemy 把所有生成的 SQL 打印出来然后跟原生 SQL 的执行计划对比。这一步能发现很多隐性差异例如 ORM 生成的查询里条件顺序不同、连接顺序不同可能导致索引选择不一致。6.2 迁移过程中踩过的坑与解决思路改造中最常遇到的问题是DetachedInstanceError。原本单独使用session.query时拿到对象一切正常但对象一旦离开 Session 作用域再访问关系属性就会抛错。解决方案有几种设置expire_on_commitFalse减少对象过期。需要长时间持有对象时用session.expunge(obj)把对象从会话中分离出来但分离后关系属性依然需要手动加载。在返回给前端前把需要的关系属性全部加载完成再关闭 Session。第二个常见坑是懒加载在with Session退出后失效。很多 Web 框架的序列化发生在请求处理完之后Session 已经关闭访问user.posts就直接报错。所以接口里要么用selectinload提前加载要么让 Session 在序列化完成前保持开启。我更推荐前者因为 Session 的生命周期应该尽量短。第三种是 MySQL 字符集不一致。老表用的是latin1新连接串用的utf8mb4查询结果里中文变问号。最彻底的办法是把表结构和列都改成utf8mb4改之前先备份。如果暂时不能动表至少在连接串里保持原有字符集别在连接层和表结构层搞混。6.3 迁移完成之后值得长期坚持的几个习惯迁移完成不等于一劳永逸。我会在代码里坚持两个原则所有数据库操作尽量走模型和 Session复杂报表类查询单独用text()写原生 SQL 并注明原因。这样既保持数据层统一又不会在极端查询上硬凹 ORM 写法。另外建议在项目里尽早引入 Alembic。每次模型变化都生成一条迁移脚本代码评审时可以看见表结构的变化历史回滚也更加容易。没有迁移脚本的项目模型和数据库表一旦对不上排查问题的成本会很高。关于接口性能我会在关键查询路径上定期检查 SQL 日志看看有没有非预期的多查、慢查。MySQL 的慢查询日志也可以配合使用我在迁移后的一段时间里每周扫一次慢日志专门清理新出现的 N1 和未走索引的查询。这套检查机制比临时发现问题再修复要稳得多。最后说一点我的实际体会这套 Python SQLAlchemy MySQL 的组合我用了快三年最深的感受是ORM 真正减少的不是 SQL 学习成本而是日常增删改查和关系处理的重复劳动。你依然需要懂表结构设计、索引、事务、隔离级别但这些知识的应用场景变得更加集中和清晰。对一个从手写 SQL 过来的开发者来说初期最需要克服的是不放心——总觉得要让 ORM 打印出 SQL 看一眼才踏实。这个习惯保持下去是好事因为理解 ORM 生成什么 SQL正是你写出高性能 ORM 代码的基础。
如果你写 Python 已经一阵子了大概率会撞上这堵墙项目逻辑越来越复杂数据表越来越多手写 SQL 越来越像家务活。我当年也一样列表页要拼条件、插入记录要拼括号、联表查询要反复确认字段名好不容易跑通改需求时又得回去翻十几条 SQL 字符串。后来切到 SQLAlchemy 配合 MySQL用 ORM 的方式重新组织数据层整个项目的维护成本直接降了一个台阶。这篇东西不是什么高深理论就是我从 0 到 1 把 Python SQLAlchemy MySQL 这套组合落地的一手记录。内容包括环境怎么搭、模型怎么建、CRUD 怎么写、查询怎么避免性能坑最后还会分享我改造老项目原生 SQL 时踩过的真实问题。适合刚接触 ORM 的 Python 开发者也适合那些已经会用 SQL 但还没下定决心切 ORM 的朋友参考。1. 先从为什么用 ORM说起一段手写 SQL 的体验1.1 手写 SQL 半年后我决定换条路在切换到 SQLAlchemy 之前我维护过一个内部管理系统数据层全部是 pymysql 加字符串拼 SQL。听起来并不可怕真正跑起来才难受。最典型的场景是列表筛选。用户在前端勾几个筛选条件后端就要动态拼WHERE。我的代码长这样sql SELECT id, name, age, email FROM users WHERE 11 if name: sql AND name LIKE %s params.append(f%{name}%) if age_min: sql AND age %s params.append(age_min) ...写一次无所谓写多了你会发现同样一段逻辑散落在各个业务模块里改一个字段名要全局搜索替换而且容易漏。更麻烦的是联表场景JOIN一多查询结果里重名字段、类型转换、Null 处理都会变成隐性地雷。再加上连接管理、事务提交、异常回滚这些样板代码真正写业务的时间可能连一半都不到。ORM 解决的正是这一类问题把表结构映射成 Python 类把行记录映射成对象把常见的增删改查封装成方法把条件拼接改成方法链或查询表达式。它不是说 SQL 不重要而是让你不必在 80% 的常规操作上重复劳动把精力留给真正复杂的查询和性能优化。1.2 ORM 到底解决了什么以及它不解决什么很多人对 ORM 的印象是自动生成 SQL这个说法没错但不完整。SQLAlchemy 的定位更准确它是一个 SQL 工具包加对象关系映射器。前半句意味着你随时可以写原生 SQL后半句意味着常规操作可以完全面向对象。用 ORM 的好处我体感最明显的三件事字段定义集中管理。数据表的列在模型类里一眼看全改表结构时改一处所有用到模型的地方同步生效不用到处找字符串。关系查询变得直观。user.posts这种写法让我不用每次手写JOIN代码读起来接近自然语言。数据库差异被隔离。SQLAlchemy 的方言机制让同一套模型可以跑在 MySQL、SQLite、PostgreSQL 上本地测试用 SQLite生产用 MySQL几乎不用改业务代码。但 ORM 不是万能药。它不适合重度报表类查询那种几十行 JOIN 加子查询加窗口函数的 SQL直接写原生语句反而更清晰。SQLAlchemy 也支持这种组合拳你可以在 ORM 里用text()写原生 SQL结果照样映射成对象。所以我们不用纠结用 ORM 还是写 SQL而应该是能用 ORM 表达的常规操作用 ORM复杂查询和批量操作用 SQL两者相辅相成。2. 环境搭建Python、MySQL、SQLAlchemy 的版本坑位图2.1 版本选择与安装路线这套组合里最容易出问题的不是 SQLAlchemy而是 MySQL 的驱动和认证插件。我先说结论再解释为什么。推荐组合Python 3.10 以上MySQL 8.0SQLAlchemy 2.0 系列驱动用 PyMySQL另外装一个cryptography库。安装命令很简单pip install sqlalchemy pymysql cryptography很多人装完 PyMySQL 后连接 MySQL 8.0 报错提示Authentication plugin caching_sha2_password cannot be loaded。这是因为 MySQL 8.0 默认的认证插件是caching_sha2_password老版本的 PyMySQL 或者某些中间件不支持。解决方案有两个方向一是升级 PyMySQL 并安装cryptography库让驱动支持新的认证方式二是在 MySQL 里把用户的认证插件改回mysql_native_password但这不是长久之计新项目建议直接走第一种。如果你是在 Windows 本机装 MySQL记住安装时一路 Next 容易踩坑选 Server only 还是 Developer Default 看你的需求但字符集一定要选utf8mb4或者装完后在 my.ini 里配置。这点后文会单独说。2.2 验证安装与连接串配置装好之后先不用急着写模型用最小代码验证连通性from sqlalchemy import create_engine, text engine create_engine( mysqlpymysql://root:yourpassword127.0.0.1:3306/test_db?charsetutf8mb4 ) with engine.connect() as conn: result conn.execute(text(SELECT 1)) print(result.scalar())这里解释一下连接串的结构dialectdriver://username:passwordhost:port/database?参数拆开来看就是mysqlpymysql告诉 SQLAlchemy 使用 MySQL 方言并且通过 PyMySQL 驱动访问。root:yourpassword用户名和密码。127.0.0.1:3306地址和端口。test_db目标数据库需要先手动创建。charsetutf8mb4字符集参数这个非常关键。utf8mb4是 MySQL 上真正意义上的完整 UTF-8编码支持 emoji 和生僻字。MySQL 默认的utf8实际上最多只存 3 字节遇到 4 字节字符会报错或乱码。所以连接串里写utf8mb4建表时也统一用utf8mb4这是我从乱码坑里爬出来的经验。2.3 最小可运行 Demoengine 与 connectionSQLAlchemy 里有两个容易混淆的概念Engine 和 Connection。Engine 是全局唯一的数据库引擎对象负责维护连接池、方言解析和 SQL 编译。整个应用生命周期里一般只创建一次放在模块顶部复用。Connection 是实际连接到数据库的会话资源用完要释放。我推荐的日常模式是这样的from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine create_engine( mysqlpymysql://root:yourpassword127.0.0.1:3306/test_db?charsetutf8mb4, pool_pre_pingTrue, pool_recycle3600, echoFalse, )三个参数是实战经验pool_pre_pingTrue连接池里的连接在每次使用前先发送一次探测相当于 SELECT 1如果发现连接已经断开就自动剔除并新建。MySQL 默认wait_timeout是 8 小时连接闲置超过这个时间会被服务端关闭没有这个参数就会报MySQL server has gone away。pool_recycle3600强制让连接最多存活 3600 秒就回收重建进一步避免拿到失效连接。echoFalse开启会打印所有 SQL 日志开发调试时设为 True 很爽生产环境务必关掉。3. 第一个模型从数据表到 Python 类的映射3.1 声明式基类与字段映射SQLAlchemy 2.0 的声明式写法已经很成熟我直接用它做示例。核心是先定义一个继承自DeclarativeBase的基类然后所有模型类继承它。from sqlalchemy import String, Integer from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class User(Base): __tablename__ users id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(64), nullableFalse, uniqueTrue) email: Mapped[str] mapped_column(String(128), nullableFalse, default) age: Mapped[int] mapped_column(Integer, nullableFalse, default0)这里有几个在 MySQL 场景下需要特别注意的细节Integer映射到 MySQL 是INT。如果数据量很大主键建议显式用BigInteger。String(64)映射为VARCHAR(64)给长度是为了索引效率和存储优化不要无脑给 255尤其在utf8mb4下一个字符最多占 4 字节长 VARCHAR 会让磁盘和内存消耗变大。nullable和default的含义要区分清楚nullableFalse是数据库层面的非空约束default0是 Python 层面的默认值它不会直接改变数据库表结构里的DEFAULT。如果你想让 MySQL 表结构本身也带默认值要写成server_defaulttext(0)。这是新手经常被坑的点模型里给了default客户端不传字段时确实会用默认值但如果直接用原生 SQL 插入数据默认值就不一定生效。3.2 建表与改表create_all 的现实操作建表最简单的方式是直接在主代码里Base.metadata.create_all(engine)它只会创建数据库中不存在的表不会修改已存在的表。所以开发阶段改模型字段后跑create_all不会自动加列你需要先删除旧表再重建或者用迁移工具。我在这件事上的建议非常明确任何要从开发走向上线的项目尽早引入 Alembic 做迁移管理不要依赖drop_all和create_all那一套。Alembic 是 SQLAlchemy 官方的迁移工具它能生成版本化的迁移脚本支持在已有表上安全地加列、改类型、加索引。等你生产库有数据之后再想 删库重建 就晚了。如果只是想快速验证模型可以这样做Base.metadata.drop_all(engine) Base.metadata.create_all(engine)仅限本地开发我每次跑都会再三确认连的是不是本地库这种命令误连生产库的教训在网上一抓一大把。3.3 一对多与多对多关系声明数据表之间的关联是 ORM 的精髓。先看最常用的一对多一个用户有多篇文章。class Post(Base): __tablename__ posts id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200), nullableFalse) user_id: Mapped[int] mapped_column(ForeignKey(users.id), nullableFalse) author: Mapped[User] relationship(back_populatesposts)同时在 User 类里补上posts: Mapped[list[Post]] relationship(back_populatesauthor)ForeignKey(users.id)是数据库层面的外键约束relationship是 ORM 层面的导航属性。两者配合才能做到user.posts这种对象式访问。多对多关系需要一张中间关联表。假设文章和标签是多对多from sqlalchemy import Table, Column post_tags Table( post_tags, Base.metadata, Column(post_id, ForeignKey(posts.id), primary_keyTrue), Column(tag_id, ForeignKey(tags.id), primary_keyTrue), ) class Tag(Base): __tablename__ tags id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), nullableFalse, uniqueTrue)然后在 Post 和 Tag 里分别加# Post 里 tags: Mapped[list[Tag]] relationship(secondarypost_tags, back_populatesposts) # Tag 里 posts: Mapped[list[Post]] relationship(secondarypost_tags, back_populatestags)核心在于secondarypost_tags它告诉 SQLAlchemy 这张中间表是纯关联表不需要单独的模型类。多对多查询时ORM 会自动生成跨两张表的 JOIN。关系这个功能我之前觉得可有可无直到接手一个接口返回多层嵌套 JSON 的项目时才发现没有 ORM 关系的话每层都要手写联表查询字段一多很容易漏。而这种声明式关系配上序列化辅助工具代码量和出错率都明显下降。4. Session 与 CRUD真正决定每天编码体验的环节4.1 session 的创建与生命周期模型定义了表结构但是真正跟数据库打交道要靠 Session。Session 可以理解为一个工作单元它跟踪你在本次操作中加载和修改的所有对象直到commit()才把变更一次性提交到数据库。我推荐用sessionmaker创建一个工厂函数然后在每个请求或业务处理里用上下文管理器的方式获取 Sessionfrom sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine, expire_on_commitFalse) def get_db(): with SessionLocal() as session: yield session注意expire_on_commitFalse这个参数它直接影响你在commit()之后还能不能继续访问对象的属性。默认情况下commit()之后会话会把所有对象的属性标记为过期下次访问属性时会重新发 SQL 查询。这会导致一个很常见的困扰我在commit()之后想读取user.id结果触发了一次额外的数据库查询如果 Session 已经关闭就直接报DetachedInstanceError。把它设为 Falsecommit()之后对象保持原样省心很多。Session 的生命周期一定要短用完就关。很多人写脚本时创建了一个全局 Session跑完不关程序一直不退出连接池很快被占满。用with SessionLocal() as session的语法退出代码块时 Session 会自动关闭这也是我最推荐的方式。4.2 增删改查的标准姿势插入数据with SessionLocal() as session: user User(nametom, emailtomexample.com, age25) session.add(user) session.commit() print(user.id) # 提交后自增主键已写回对象批量插入用add_allwith SessionLocal() as session: session.add_all([ User(namea, emailaexample.com), User(nameb, emailbexample.com), ]) session.commit()查询数据with SessionLocal() as session: # 查询单个对象按条件 user session.scalars(select(User).where(User.name tom)).first() # 查询所有用户按年龄倒序 users session.scalars(select(User).order_by(User.age.desc())).all()SQLAlchemy 2.0 的查询风格统一成select()不再推荐旧的session.query()写法。虽然旧的还能用但新项目建议直接学新的免得看文档时精神分裂。更新数据有两种方式。第一种是拿到对象后直接改属性Session 会帮你跟踪with SessionLocal() as session: user session.scalars(select(User).where(User.id 1)).first() if user: user.age 30 session.commit()第二种是批量更新用update()方法避免把所有数据加载到内存from sqlalchemy import update with SessionLocal() as session: session.execute( update(User).where(User.age 18).values(statusminor) ) session.commit()删除数据同理from sqlalchemy import delete with SessionLocal() as session: session.execute(delete(User).where(User.id 999)) session.commit()二维表操作如update和delete不经过 ORM 对象缓存性能好但也不会触发 ORM 层的级联操作使用时要自己判断是否需要同时清理关联数据。4.3 事务提交与回滚什么时候该 commit什么时候该 rollback事务是数据库一致性的根基。在 SQLAlchemy 中Session 默认开启事务commit()是事务结束点rollback()则放弃本次所有变更。我踩过一个印象很深的坑批量处理任务时循环里每条数据都commit()一次结果中间某条失败前面已提交的数据无法回滚数据处于改了一半的状态。正确做法是尽量把一批操作放进同一个事务最后统一提交with SessionLocal() as session: try: for data in task_list: session.add(User(**data)) session.commit() except Exception: session.rollback() raise但事务又不宜过大。我见过有人把几万条记录的导入放进一个事务里跑了几分钟还没结束MySQL 锁冲突、undo log 膨胀、内存飙升全来了。合理的做法是分批提交每批 500 到 1000 条即使失败也只需要回滚当前批次。这个经验在数据清洗、迁移、定时任务里特别重要。5. 查询层面最容易踩的坑N1、分页、连接池5.1 懒加载引发的 N1 问题以及两种解法ORM 最方便的是关系导航最坑的也是关系导航。默认情况下访问user.posts时 SQLAlchemy 会立刻发一条查询去拿该用户的所有文章。如果循环里先查出 100 个用户再逐个访问user.posts就会变成 1 条主查询加 100 条附属查询这就是经典的 N1 问题。# 反例会产生 1 N 条 SQL with SessionLocal() as session: users session.scalars(select(User)).all() for user in users: print(user.posts) # 每次访问都发一条新 SQL解法一是使用selectinload或joinedload主动预加载关系from sqlalchemy.orm import selectinload with SessionLocal() as session: users session.scalars( select(User).options(selectinload(User.posts)) ).all() for user in users: print(user.posts) # 不会产生额外 SQL我自己更常用selectinload它的原理是先加载用户列表再生成一条WHERE post.user_id IN (...)的查询把关联数据一次性取回再按内存中的外键把对象关联好。joinedload则是在原 SQL 上直接 JOIN如果主表数据量大且每行关联数据也多结果集会成倍膨胀。大多数业务场景selectinload更稳。连接没有预加载时如果你知道自己接下来要访问关系但不想改原查询也可以用session.refresh(user)配合指定属性但日常开发还是建议在查询时就把加载策略写清楚不要依赖运行时补救。5.2 分页、过滤与排序的正确姿势分页最基础的方式是limit加offsetusers session.scalars( select(User).order_by(User.id).limit(20).offset(40) ).all()在数据量小的后台管理中够用。但数据量超过几十万行时OFFSET会导致 MySQL 扫描并丢弃前 N 行页码越深性能越差。这时候我会改用键集分页利用上次查询的最后一条记录位置继续往后翻last_id 40 users session.scalars( select(User).where(User.id last_id).order_by(User.id).limit(20) ).all()这种方式的查询条件能直接走主键索引无论翻到第几页耗时都差不多。缺点是无法随意跳页适合瀑布流或加载更多场景。动态过滤条件的经验是用where()方法逐步追加条件比手拼字符串安全得多字段类型和转义都由 ORM 处理。排序方面order_by接受列对象支持多字段组合例如先按状态分组再按时间排序select(Order).where(Order.status paid).order_by(Order.created_at.desc(), Order.id.desc())注意 MySQL 对 NULL 值的排序是升序在最前、降序在最后如果你需要自定义 NULL 位置要用nullslast()或nullsfirst()在 SQLAlchemy 里也有对应的函数。这个边界条件很容易在测试时漏掉。5.3 连接池、并发与会话清理前面提到create_engine自带连接池。默认池大小是 5最大溢出是 10也就是说并发超过 15 个连接请求时后来的请求会等待。小应用一般够用但接口并发上来了以后会遇到连接等待超时。按我的经验可以从两个方向调一是加大连接池二是排查是否有连接泄漏。连接泄漏的典型表现是运行一天后数据库连接数持续上涨最终报Too many connections。常见原因就是 Session 没关闭或者线程里创建了 Session 没有释放。调参参考engine create_engine( url, pool_size10, max_overflow20, pool_pre_pingTrue, pool_recycle3600, )线上 MySQL 的max_connections默认是 151连接池总大小要留有余量别把数据库连接池顶满。除了调参数我更想强调代码习惯Session 一定要遵循使用即关闭的原则在 Web 应用里最好通过依赖注入或中间件确保每个请求结束时关闭连接。并发写入场景还有个容易忽视的问题MySQL 默认隔离级别是可重复读但在高并发更新同一行时可能出现更新丢失。SQLAlchemy 层面没有自动处理乐观锁你需要自己加版本号字段或者用FOR UPDATE做悲观锁。这个话题展开能写一整篇这里只提醒一点使用 ORM 不代表并发问题自动消失事务边界和锁策略依然必须自己把握。6. 从 pymysql 裸 SQL 迁移到 SQLAlchemy一次真实的老项目改造记录6.1 迁移第一步不要推倒重来改造老项目最大的风险在于想一口吃成胖子。我当时的习惯是先搭好模型映射但不急着替换业务逻辑。流程是这样的先从现有表结构反推模型。如果表已经存在我不会用create_all而是直接根据表字段写模型类。写完以后先写一段对照脚本分别用原 SQL 和 ORM 跑同一条查询比对结果是否一致。我当时连的是一张订单表有 30 多个字段还有三张关联表。反推模型时特别注意了字段类型MySQL 的DECIMAL对应 SQLAlchemy 的Numeric或DecimalDATETIME对应DateTimeTINYINT(1)默认对应SmallInteger而不是布尔这些映射关系不一致会导致数据读写出现类型偏差。对照校验阶段我用得最多的调试手段是把echoTrue打开让 SQLAlchemy 把所有生成的 SQL 打印出来然后跟原生 SQL 的执行计划对比。这一步能发现很多隐性差异例如 ORM 生成的查询里条件顺序不同、连接顺序不同可能导致索引选择不一致。6.2 迁移过程中踩过的坑与解决思路改造中最常遇到的问题是DetachedInstanceError。原本单独使用session.query时拿到对象一切正常但对象一旦离开 Session 作用域再访问关系属性就会抛错。解决方案有几种设置expire_on_commitFalse减少对象过期。需要长时间持有对象时用session.expunge(obj)把对象从会话中分离出来但分离后关系属性依然需要手动加载。在返回给前端前把需要的关系属性全部加载完成再关闭 Session。第二个常见坑是懒加载在with Session退出后失效。很多 Web 框架的序列化发生在请求处理完之后Session 已经关闭访问user.posts就直接报错。所以接口里要么用selectinload提前加载要么让 Session 在序列化完成前保持开启。我更推荐前者因为 Session 的生命周期应该尽量短。第三种是 MySQL 字符集不一致。老表用的是latin1新连接串用的utf8mb4查询结果里中文变问号。最彻底的办法是把表结构和列都改成utf8mb4改之前先备份。如果暂时不能动表至少在连接串里保持原有字符集别在连接层和表结构层搞混。6.3 迁移完成之后值得长期坚持的几个习惯迁移完成不等于一劳永逸。我会在代码里坚持两个原则所有数据库操作尽量走模型和 Session复杂报表类查询单独用text()写原生 SQL 并注明原因。这样既保持数据层统一又不会在极端查询上硬凹 ORM 写法。另外建议在项目里尽早引入 Alembic。每次模型变化都生成一条迁移脚本代码评审时可以看见表结构的变化历史回滚也更加容易。没有迁移脚本的项目模型和数据库表一旦对不上排查问题的成本会很高。关于接口性能我会在关键查询路径上定期检查 SQL 日志看看有没有非预期的多查、慢查。MySQL 的慢查询日志也可以配合使用我在迁移后的一段时间里每周扫一次慢日志专门清理新出现的 N1 和未走索引的查询。这套检查机制比临时发现问题再修复要稳得多。最后说一点我的实际体会这套 Python SQLAlchemy MySQL 的组合我用了快三年最深的感受是ORM 真正减少的不是 SQL 学习成本而是日常增删改查和关系处理的重复劳动。你依然需要懂表结构设计、索引、事务、隔离级别但这些知识的应用场景变得更加集中和清晰。对一个从手写 SQL 过来的开发者来说初期最需要克服的是不放心——总觉得要让 ORM 打印出 SQL 看一眼才踏实。这个习惯保持下去是好事因为理解 ORM 生成什么 SQL正是你写出高性能 ORM 代码的基础。
