写 Python 的人早晚会碰数据库。一开始你可能用 pymysql 裸写 SQL写几条 insert 还好等到表多了、字段改了、查询逻辑复杂了你就发现自己陷入一堆字符串拼接里改一个字段名得全局搜索查一个带条件的列表要拼半天的 where。这时候就该上 ORM 了而 SQLAlchemy 就是 Python 生态里最成熟、最完整的那一个也是我把爬虫数据、后台管理系统、分析项目存库时的默认选择。它到底解决了什么问题简单说把你脑子里对业务对象的理解类、属性和数据库里的表、字段之间搭一座桥。你定义一个User类它对应users表类属性对应字段。写代码时你操作的是对象SQLAlchemy 帮你翻译成 SQL再执行。你不用再记着每个字段的拼写不用担心手写的 SQL 字符串里混入意外的单引号导致语法错误更重要的是参数自动转义从根上规避了注入风险。谁适合读这篇刚入门 Python、想正经管数据的初学者爬虫写了不少但数据存在 JSON 文件里、想迁到 MySQL 的进阶玩家正在从裸 SQL 转向 ORM 的开发者。SQLAlchemy 的官方文档其实很全但组织方式比较工程化新手翻起来容易迷失。下面这篇是我按自己实际项目经验捋出来的版本该深入的深入该跳过的跳过照着操作基本能落地。1. 项目概述与核心思路拆解1.1 SQLAlchemy 的两层架构Core 和 ORMSQLAlchemy 跟其他 ORM 最大的不同是它从一开始就没打算只做对象映射。它分两层底层叫 Core是 SQL 表达式语言不依赖对象映射那一套上层叫 ORM在 Core 之上构建对象模型。这个设计带来的直接好处是学习曲线可以缓着走——你完全可以只用 Core 写参数化查询获得比裸 SQL 拼接更安全的体验等熟悉了再上 ORM 的对象玩法。我见过不少人一上来就懵在我该学 Core 还是 ORM这个问题上。其实不用纠结大多数业务场景直接用 ORM 就够了。Core 的价值在少数场景会体现出来比如大批量插入、复杂报表查询、需要精细控制 SQL 的时候。两者共用同一套连接和事务机制也就是说你可以在一个项目里混着用ORM 管常规增删改查Core 管批量写入和统计查询互不冲突。从为什么的角度多说一句ORM 不是用来消灭 SQL 的是让你把注意力放在业务模型上。SQL 依然要懂因为你会遇到需要手写原生 SQL 的场景但日常的增删改查、联表查询用 ORM 表达起来更清晰、更安全。调试时开echoTrue让 SQLAlchemy 把实际执行的 SQL 打印出来多看看你很快就能建立对象操作对应什么 SQL的映射直觉。1.2 为什么选 SQLAlchemy 而不是其他 ORMPython 生态里的 ORM 不止它一个Django 自带 ORMpeewee 也轻量好用。但我常年用 SQLAlchemy理由很实在第一数据库适配广。SQLite、MySQL、PostgreSQL、Oracle、SQL Server 都支持。最难得的是切换数据库时大部分业务代码不用改只改连接字符串。这对做项目的人来说太实用了本地开发用 SQLite 零配置跑起来部署到服务器换 PostgreSQL代码基本不动。我有个数据分析项目就是这样本地 SQLite 验证逻辑上线后切到 MySQL 就改了一行 URL。第二生态地位稳。Flask-SQLAlchemy、Alembic数据库迁移工具都是围绕它构建的资料多搜索引擎一找一大把遇到问题好查。相比之下 peewee 轻但功能少Django ORM 好使但绑死 Django 框架。如果你不想被某个 Web 框架绑架SQLAlchemy 几乎是唯一的主流选择。第三性能可控。ORM 并不意味着牺牲性能SQLAlchemy 给了你大量后门可以写原生 SQL、可以用 Core 批量操作、可以精细控制加载策略预加载、懒加载、只取需要的列。我自己实测过同样的批量插入场景用对方式后性能差距能达到 5 倍以上。这一点在第 5 节会展开讲。2. 环境准备与基础配置2.1 安装与版本选择先用 pip 装基础包如果连的是 MySQL还得装驱动。示例pip install sqlalchemy pip install pymysqlSQLAlchemy 2.0 系列是当前主力版本。2.0 相比 1.4 有比较大的 API 调整最明显的是Session.query()这种老写法虽然还兼容官方推荐的是select()函数式写法。这篇以 2.0 的推荐写法为主。装完验证一下版本python -c import sqlalchemy; print(sqlalchemy.__version__)如果你的环境是刚配好的Python 版本建议 3.9 以上SQLAlchemy 2.0 对 3.7 也支持但很多新特性用不上。Windows 用户如果在安装时遇到Microsoft Visual C Build Tools的报错多半是某些依赖需要本地编译换用较新的 pymysql 版本通常能解决或者直接改用纯 Python 的驱动。SQLite 不需要额外驱动这也是我推荐新手用它起步的原因。2.2 建立数据库连接连接数据库的入口是create_engine这步搞定之后所有操作都围绕 engine 进行。SQLite 和 MySQL 的连接示例from sqlalchemy import create_engine # SQLite适合本地练习、小项目 engine create_engine(sqlite:///blog.db, echoFalse) # MySQL需要安装 pymysql engine create_engine(mysqlpymysql://user:passwordlocalhost:3306/blog?charsetutf8mb4)engine 负责管理数据库连接池本身不执行业务逻辑你把它理解成一个总闸口就行。echoTrue会把 SQLAlchemy 生成的实际 SQL 打印到控制台调试阶段强烈建议开你能看到 ORM 到底执行了什么语句对排查问题帮助很大生产环境记得关掉。连接 URL 里包含用户名、密码、主机、端口、库名。密码里有特殊字符时注意转义问题比如要写成%40否则解析连接串时会截断这算是新手的常见坑。另外 MySQL 连接串里我通常都加charsetutf8mb4否则遇到 emoji 或生僻字容易乱码或报错。2.3 定义声明式模型从类到表2.0 声明式模型用Mapped和mapped_column定义字段类型更明确。看一下最简单的用户表from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Integer class Base(DeclarativeBase): pass class User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), nullableFalse) email: Mapped[str] mapped_column(String(100), uniqueTrue) age: Mapped[int] mapped_column(Integer, default0) def __repr__(self): return fUser(id{self.id}, name{self.name}, email{self.email})Base是所有模型的父类__tablename__指定表名。字段类型尽量用 SQLAlchemy 提供的String、Integer、DateTime等因为要翻译成不同数据库的方言不能直接用 Python 原生类型。nullableFalse对应数据库的 NOT NULLuniqueTrue对应唯一索引default0是 Python 侧的默认值这些约束都会在建表时生效。建表语句执行如下Base.metadata.create_all(engine)这句会把所有继承Base的模型对应表建出来。注意create_all只创建不存在的表对已存在的表不会做任何修改所以它只适合初次建库。生产环境表结构变更是迁移的活得用 Alembic 这种迁移工具生成变更脚本、记录版本、可回滚。我第一次做项目时手动 ALTER TABLE改几个字段就乱成一锅粥后来老老实实用迁移工具省心太多了。3. 核心操作详解增删改查3.1 会话机制所有操作的起点ORM 操作对象不是直接通过 engine而是通过 Session。Session 是工作单元的概念跟踪你操作的对象变更最后统一提交。你可以把它理解为一张工作台所有要操作的对象都摆在这张台上摆好了、确认没问题了一起推到数据库。from sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine) with SessionLocal() as session: user User(name张三, emailzhangsanexample.com, age30) session.add(user) session.commit()commit()提交事务事务提交后数据才真正落库。flush()可以把数据先发送到数据库但还没提交比如为了立刻取到自增 id 就需要 flush。实际项目里用with块基本就对了会话在块结束时自动关闭省去忘记 close 的隐患对新手极其友好。3.2 新增数据一次性加多条用add_allwith SessionLocal() as session: users [ User(name李四, emaillisiexample.com, age25), User(name王五, emailwangwuexample.com, age35), ] session.add_all(users) session.commit()值得提醒的是如果你批量插入几十万条数据一条条add再commit会很慢。SQLAlchemy 的 ORM 层有对象跟踪、状态维护这些开销大批量场景建议直接用 Core 的insert()或者session.bulk_insert_mappings()速度差别非常明显。批量场景我放到第 5 节专门讲这里先记住结论小批量和业务处理用 ORM大批量上 Core。3.3 查询数据最常用的部分2.0 推荐用select()来构造查询执行后用scalars()取出模型对象列表from sqlalchemy import select with SessionLocal() as session: # 查全部 users session.scalars(select(User)).all() # 条件过滤 users session.scalars(select(User).where(User.age 30)).all() # 排序 限制数量 users session.scalars( select(User).where(User.age 18).order_by(User.id.desc()).limit(10) ).all() # 按主键取单条 user session.get(User, 1)session.get(User, 1)是按主键取对象最简单也最常用没有匹配结果时返回None。where支持多个条件多个条件之间是 AND 关系。如果要 OR 条件用or_函数包起来from sqlalchemy import or_ users session.scalars( select(User).where(or_(User.age 20, User.age 60)) ).all()一个容易混淆的地方session.execute(select(User))返回的是Row对象列表要用.scalars()才能拿到User对象列表。如果只取几个列可以直接写session.execute(select(User.name, User.email))返回元组列表适合做列表展示不用把整个对象捞出来。统计场景用select(func.count()).select_from(User)配合group_by可以做分组统计。3.4 更新与删除更新在 ORM 里很直观查出来再改属性commit 即可with SessionLocal() as session: user session.get(User, 1) if user: user.age 31 session.commit()不查询直接批量更新用update()语句适合把满足条件的记录统一改个状态这种场景from sqlalchemy import update with SessionLocal() as session: session.execute(update(User).where(User.id 1).values(age32)) session.commit()删除同理user session.get(User, 1) session.delete(user) session.commit()说一个容易踩的坑更新对象后如果忘了 commit容易在不同代码块里看到改了但没生效的假象排查半天发现是事务没提交。Session 默认是自动提交关掉的所以一定要显式 commit。另一个坑是删除有外键关联的数据时如果关联表里还有引用会报IntegrityError这时候要先处理关联数据或者设计级联删除别硬删。4. 关系映射与进阶实战4.1 一对多与多对多关系表之间关系是 ORM 的核心价值也是它比手写 join 舒服的地方。一个用户有多篇文章典型的一对多。定义方式from sqlalchemy import ForeignKey from sqlalchemy.orm import relationship class Article(Base): __tablename__ articles id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200)) user_id: Mapped[int] mapped_column(ForeignKey(users.id)) user: Mapped[User] relationship(back_populatesarticles) # 在 User 类里补充 articles: Mapped[list[Article]] relationship(back_populatesuser)ForeignKey建立物理外键关联relationship建立 ORM 层面的对象关系。有了这个查询时可以直接访问user.articles拿文章列表或者article.user拿作者信息不用手写 join。这里注意back_populates要两边都写名字对得上才联得起来写漏一边关系就是单向的某些操作会失灵。多对多关系比如用户和文章的点赞关系需要一张中间关联表用Table定义两个模型分别声明relationship(secondary关联表)。实操中把关联表的唯一约束建好避免重复数据。关系定义本身不难难的是理解加载策略。4.2 常用查询技巧速查我把日常最常用的查询方法整理成表方便索引需求写法取所有记录select(User).all()条件过滤.where(User.age 18)多条件 AND.where(User.age 18, User.city 北京)或条件.where(or_(条件1, 条件2))排序.order_by(User.id.desc())分页.offset(20).limit(10)按主键取session.get(User, 1)统计总数select(func.count()).select_from(User)去重.distinct()分页这个值得展开.offset(20).limit(10)翻译成 SQL 是LIMIT 10 OFFSET 20数据量大时 offset 越深越慢因为数据库要扫描并跳过前面所有行。几万条内无所谓百万级以上可以考虑用主键游标分页where id 上次最后一条id。这个优化我在一个后台管理模块里实测过翻到第 100 页时查询时间从 1.2 秒降到 50 毫秒左右差距相当可观。4.3 事务与会话生命周期管理事务是数据库操作的基本保障。Session 的commit()提交事务rollback()回滚。一个典型事务处理场景with SessionLocal() as session: try: session.add(user) session.add(article) session.commit() except Exception: session.rollback() raise只要中间的某个操作失败rollback()会把之前所有未提交的改动撤销避免半截数据落库。做转账、订单这类涉及多张表的业务这个结构是底线。反过来如果你不 try 也不 rollback一个失败的操作可能让 session 处于半死不活的状态后续操作全部异常排查起来血压拉满。实际项目里建议给会话管理加一层统一封装提供一个获取 session 的入口。Flask 里用flask-sqlalchemy的db.sessionFastAPI 里用依赖注入拿 session。核心原则是每个请求对应一个 session请求结束关闭不要跨线程共享 session。跨线程共享 session 会出现各种诡异问题比如数据没提交就看不见、状态错乱新手很容易在这上面栽跟头。5. 常见问题与避坑指南5.1 懒加载与 N1 查询陷阱前面提过懒加载。再补充一个现象开启 ORM 的懒加载后一旦 session 关闭再访问未加载的属性会报DetachedInstanceError: Instance is not bound to a Session。这是新手踩得最多的坑之一。查出来的对象在with块里好好的一出来就报错原因就是 session 关闭后对象变成了游离态。解决思路有三种要么在 session 内提前用selectinload把关联数据加载好要么用session.expunge_all()把对象变成普通对象后保存必要字段要么干脆取需要的字段组建普通的 Python 对象或字典返回。我看很多教程推荐第三种虽然麻烦点但数据边界最清晰。别指望 session 关闭后还能像热数据一样随意访问这个认知是每个 ORM 使用者必须建立的。N1 查询的问题也在这里。访问user.articles时 SQLAlchemy 会即时发一条新 SQL 去查文章开发时方便但如果你循环打印 100 个用户的文章列表就会执行额外 100 条 SQL。解决办法是查询时用selectinload预加载from sqlalchemy.orm import selectinload users session.scalars( select(User).options(selectinload(User.articles)) ).all()这样一条主查询加一条批量查询就能拿到所有数据。建议在查询方法里显式声明需要的关联而不是全都依赖懒加载。5.2 大批量数据插入的性能优化ORM 插入几十万条数据会很慢因为每条都要经过对象创建、状态跟踪、Flush 等环节。我实测过50 万条数据用add_all大约需要 20-30 秒而用 Core 的insert批量模式只需要 3-5 秒差距非常明显。批量场景建议直接上 Corefrom sqlalchemy import insert data [ {name: fuser_{i}, email: fuser{i}example.com, age: 20} for i in range(10000) ] with engine.begin() as conn: conn.execute(insert(User), data)engine.begin()自动开启事务并在结束时 commit传入字典列表做批量写入效率比 ORM 逐条高得多。如果数据量真的特别大建议分批处理每批 1000-5000 条再加个进度打印方便观察任务跑到了哪。这里有个细节批量插入字典里的字段必须跟表完全对应传多了会报错传少了用默认值所以构造数据时要把字段对齐。5.3 连接池与超时问题engine 自带连接池默认pool_size5。高并发小项目经常遇到TimeoutError: QueuePool limit of size 5 overflow 10 reached说明连接被占满。处理方式加大连接池、启用pool_pre_ping每次取连接前先探测可用性。MySQL 默认的wait_timeout可能是 8 小时连接闲置过久会被服务端断开下次复用时直接报错。pool_pre_pingTrue能自动识别失效连接并重建这个参数我几乎是必开的。完整配置engine create_engine( mysqlpymysql://user:passlocalhost:3306/blog?charsetutf8mb4, pool_size10, max_overflow20, pool_pre_pingTrue, )另外如果你遇到Lost connection during query或者server has gone away除了连接池问题还要检查是不是单条 SQL 执行时间超过了数据库的超时阈值。做大数据量查询时该分批就分批别一条 SQL 跑几分钟。5.4 常见报错速查表报错信息原因解决ModuleNotFoundError: No module named pymysql没装驱动pip install pymysqlDetachedInstanceErrorsession 关闭后访问懒加载属性用 selectinload 预加载或提前取出数据IntegrityError唯一约束冲突 / 外键约束失败检查数据提交前 try/rollbackOperationalError: (2002, ...)数据库服务未启动或地址不通检查服务与连接串TimeoutError: QueuePool limit连接池耗尽调大 pool_size / max_overflowUnicodeDecodeError / 乱码字符集不匹配连接串写charsetutf8mb4表也建 utf8mb4IntegrityError这个值得多说两句。唯一键冲突时session 会进入失败状态如果不 rollback后续操作都会异常。所以涉及到唯一约束的插入一定要 try 一下冲突时 rollback 再走更新或跳过逻辑。爬虫去重场景我就在 6.1 里给你一个可直接抄的写法。6. 实战用 SQLAlchemy 存储爬虫数据6.1 模型设计与去重逻辑配合SQLAlchemy 储存爬虫数据这个高频需求给一个可落地的完整示例。基本流程是爬虫拿到数据 → 清洗 → 写入 MySQL。假设爬的是公开的图书信息模型如下class Book(Base): __tablename__ books id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) isbn: Mapped[str] mapped_column(String(20), uniqueTrue) title: Mapped[str] mapped_column(String(200)) price: Mapped[float] mapped_column(nullableTrue) source_url: Mapped[str] mapped_column(String(500)) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.now)存储时注意两点一是用isbn做唯一键重复抓取时先查询再决定 insert 还是 skip今天爬到的和昨天重复的页面不会产生垃圾数据二是created_at默认取当前时间入库时不用手动传。判断重复的代码existing session.execute( select(Book.id).where(Book.isbn book[isbn]) ).first() if not existing: session.add(Book(**book)) else: session.rollback() # 或者什么都不做跳过这一条这里有个小技巧用Book(**book)直接把字典展开成关键字参数构造模型前提是字典的键跟模型字段名完全一致。所以爬虫解析完数据后先做一个字段名对齐的清洗步骤把书名改成title、价格改成price后面入库就顺了。别小看这一步字段对不齐是爬虫入库最常见的报错来源。6.2 批量写入与提交策略爬虫任务往往是持续抓取几百上千条积累起来可以积攒到一定量再批量入库。批量任务建议每 500 条 commit 一次避免长时间事务占用连接。如果整批写入后某条数据违反唯一约束会导致整批回滚所以去重最好前置——在入库前就用集合把已存在的 isbn 过滤一遍减少数据库回滚的次数。一个提升效率的写法先查一次库把已有的 isbn 取成一个 set然后内存里过滤掉重复项最后一次性批量插入。这种方式比每条都去数据库查一次快得多尤其适合一次性导入几千条的初始化任务。我做过一次爬虫数据迁移3 万条数据用这个思路导入去重加写入总共花了几秒比逐条判断快了一个量级。日常开发中我还习惯搭配 Navicat 这类图形化客户端配合排查。起一个简单的查询接口或者直接连库看表结构、确认数据落库情况比在代码里敲 Python 交互式查表方便。注意给只给需要的账号配只读权限避免手滑改坏数据这也是团队协作里数据库安全的基本习惯。结尾说到这儿聊点个人体会。SQLAlchemy 真正上手之后你会发现写数据相关代码的体验比裸 SQL 舒服太多不用再手动拼接查询条件、不用担心注入、切换数据库也很从容。但我建议你依然保持对 SQL 本身的敏感度——ORM 只是帮你生成 SQL你最好知道它生成的 SQL 长什么样。调试时开echoTrue看几眼时间久了就能建立对象操作 ↔ SQL的映射感遇到性能问题也更容易定位。最后再分享一个实用小技巧从select(User)这种查询里获取模型列表时记得scalars().all()和execute().all()的区别。前者返回User对象列表后者返回Row元组列表。很多新手困惑查出来怎么不是对象多半是用了后者没加scalars()。这两种方式各有适用场景淘清了写起来就顺了。SQLAlchemy 内容很深这篇覆盖的是我平时用得最多的部分。往后再碰到分库分表、异步 ORMasyncpg加 SQLAlchemy 异步模式、数据库迁移这类进阶主题再单独开篇聊。数据操作是几乎所有 Python 项目的底座把这层打稳后面做爬虫、做数据分析、做后端接口都会顺畅很多。
SQLAlchemy核心入门与实战：从裸SQL到ORM的Python数据库操作指南
写 Python 的人早晚会碰数据库。一开始你可能用 pymysql 裸写 SQL写几条 insert 还好等到表多了、字段改了、查询逻辑复杂了你就发现自己陷入一堆字符串拼接里改一个字段名得全局搜索查一个带条件的列表要拼半天的 where。这时候就该上 ORM 了而 SQLAlchemy 就是 Python 生态里最成熟、最完整的那一个也是我把爬虫数据、后台管理系统、分析项目存库时的默认选择。它到底解决了什么问题简单说把你脑子里对业务对象的理解类、属性和数据库里的表、字段之间搭一座桥。你定义一个User类它对应users表类属性对应字段。写代码时你操作的是对象SQLAlchemy 帮你翻译成 SQL再执行。你不用再记着每个字段的拼写不用担心手写的 SQL 字符串里混入意外的单引号导致语法错误更重要的是参数自动转义从根上规避了注入风险。谁适合读这篇刚入门 Python、想正经管数据的初学者爬虫写了不少但数据存在 JSON 文件里、想迁到 MySQL 的进阶玩家正在从裸 SQL 转向 ORM 的开发者。SQLAlchemy 的官方文档其实很全但组织方式比较工程化新手翻起来容易迷失。下面这篇是我按自己实际项目经验捋出来的版本该深入的深入该跳过的跳过照着操作基本能落地。1. 项目概述与核心思路拆解1.1 SQLAlchemy 的两层架构Core 和 ORMSQLAlchemy 跟其他 ORM 最大的不同是它从一开始就没打算只做对象映射。它分两层底层叫 Core是 SQL 表达式语言不依赖对象映射那一套上层叫 ORM在 Core 之上构建对象模型。这个设计带来的直接好处是学习曲线可以缓着走——你完全可以只用 Core 写参数化查询获得比裸 SQL 拼接更安全的体验等熟悉了再上 ORM 的对象玩法。我见过不少人一上来就懵在我该学 Core 还是 ORM这个问题上。其实不用纠结大多数业务场景直接用 ORM 就够了。Core 的价值在少数场景会体现出来比如大批量插入、复杂报表查询、需要精细控制 SQL 的时候。两者共用同一套连接和事务机制也就是说你可以在一个项目里混着用ORM 管常规增删改查Core 管批量写入和统计查询互不冲突。从为什么的角度多说一句ORM 不是用来消灭 SQL 的是让你把注意力放在业务模型上。SQL 依然要懂因为你会遇到需要手写原生 SQL 的场景但日常的增删改查、联表查询用 ORM 表达起来更清晰、更安全。调试时开echoTrue让 SQLAlchemy 把实际执行的 SQL 打印出来多看看你很快就能建立对象操作对应什么 SQL的映射直觉。1.2 为什么选 SQLAlchemy 而不是其他 ORMPython 生态里的 ORM 不止它一个Django 自带 ORMpeewee 也轻量好用。但我常年用 SQLAlchemy理由很实在第一数据库适配广。SQLite、MySQL、PostgreSQL、Oracle、SQL Server 都支持。最难得的是切换数据库时大部分业务代码不用改只改连接字符串。这对做项目的人来说太实用了本地开发用 SQLite 零配置跑起来部署到服务器换 PostgreSQL代码基本不动。我有个数据分析项目就是这样本地 SQLite 验证逻辑上线后切到 MySQL 就改了一行 URL。第二生态地位稳。Flask-SQLAlchemy、Alembic数据库迁移工具都是围绕它构建的资料多搜索引擎一找一大把遇到问题好查。相比之下 peewee 轻但功能少Django ORM 好使但绑死 Django 框架。如果你不想被某个 Web 框架绑架SQLAlchemy 几乎是唯一的主流选择。第三性能可控。ORM 并不意味着牺牲性能SQLAlchemy 给了你大量后门可以写原生 SQL、可以用 Core 批量操作、可以精细控制加载策略预加载、懒加载、只取需要的列。我自己实测过同样的批量插入场景用对方式后性能差距能达到 5 倍以上。这一点在第 5 节会展开讲。2. 环境准备与基础配置2.1 安装与版本选择先用 pip 装基础包如果连的是 MySQL还得装驱动。示例pip install sqlalchemy pip install pymysqlSQLAlchemy 2.0 系列是当前主力版本。2.0 相比 1.4 有比较大的 API 调整最明显的是Session.query()这种老写法虽然还兼容官方推荐的是select()函数式写法。这篇以 2.0 的推荐写法为主。装完验证一下版本python -c import sqlalchemy; print(sqlalchemy.__version__)如果你的环境是刚配好的Python 版本建议 3.9 以上SQLAlchemy 2.0 对 3.7 也支持但很多新特性用不上。Windows 用户如果在安装时遇到Microsoft Visual C Build Tools的报错多半是某些依赖需要本地编译换用较新的 pymysql 版本通常能解决或者直接改用纯 Python 的驱动。SQLite 不需要额外驱动这也是我推荐新手用它起步的原因。2.2 建立数据库连接连接数据库的入口是create_engine这步搞定之后所有操作都围绕 engine 进行。SQLite 和 MySQL 的连接示例from sqlalchemy import create_engine # SQLite适合本地练习、小项目 engine create_engine(sqlite:///blog.db, echoFalse) # MySQL需要安装 pymysql engine create_engine(mysqlpymysql://user:passwordlocalhost:3306/blog?charsetutf8mb4)engine 负责管理数据库连接池本身不执行业务逻辑你把它理解成一个总闸口就行。echoTrue会把 SQLAlchemy 生成的实际 SQL 打印到控制台调试阶段强烈建议开你能看到 ORM 到底执行了什么语句对排查问题帮助很大生产环境记得关掉。连接 URL 里包含用户名、密码、主机、端口、库名。密码里有特殊字符时注意转义问题比如要写成%40否则解析连接串时会截断这算是新手的常见坑。另外 MySQL 连接串里我通常都加charsetutf8mb4否则遇到 emoji 或生僻字容易乱码或报错。2.3 定义声明式模型从类到表2.0 声明式模型用Mapped和mapped_column定义字段类型更明确。看一下最简单的用户表from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Integer class Base(DeclarativeBase): pass class User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), nullableFalse) email: Mapped[str] mapped_column(String(100), uniqueTrue) age: Mapped[int] mapped_column(Integer, default0) def __repr__(self): return fUser(id{self.id}, name{self.name}, email{self.email})Base是所有模型的父类__tablename__指定表名。字段类型尽量用 SQLAlchemy 提供的String、Integer、DateTime等因为要翻译成不同数据库的方言不能直接用 Python 原生类型。nullableFalse对应数据库的 NOT NULLuniqueTrue对应唯一索引default0是 Python 侧的默认值这些约束都会在建表时生效。建表语句执行如下Base.metadata.create_all(engine)这句会把所有继承Base的模型对应表建出来。注意create_all只创建不存在的表对已存在的表不会做任何修改所以它只适合初次建库。生产环境表结构变更是迁移的活得用 Alembic 这种迁移工具生成变更脚本、记录版本、可回滚。我第一次做项目时手动 ALTER TABLE改几个字段就乱成一锅粥后来老老实实用迁移工具省心太多了。3. 核心操作详解增删改查3.1 会话机制所有操作的起点ORM 操作对象不是直接通过 engine而是通过 Session。Session 是工作单元的概念跟踪你操作的对象变更最后统一提交。你可以把它理解为一张工作台所有要操作的对象都摆在这张台上摆好了、确认没问题了一起推到数据库。from sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine) with SessionLocal() as session: user User(name张三, emailzhangsanexample.com, age30) session.add(user) session.commit()commit()提交事务事务提交后数据才真正落库。flush()可以把数据先发送到数据库但还没提交比如为了立刻取到自增 id 就需要 flush。实际项目里用with块基本就对了会话在块结束时自动关闭省去忘记 close 的隐患对新手极其友好。3.2 新增数据一次性加多条用add_allwith SessionLocal() as session: users [ User(name李四, emaillisiexample.com, age25), User(name王五, emailwangwuexample.com, age35), ] session.add_all(users) session.commit()值得提醒的是如果你批量插入几十万条数据一条条add再commit会很慢。SQLAlchemy 的 ORM 层有对象跟踪、状态维护这些开销大批量场景建议直接用 Core 的insert()或者session.bulk_insert_mappings()速度差别非常明显。批量场景我放到第 5 节专门讲这里先记住结论小批量和业务处理用 ORM大批量上 Core。3.3 查询数据最常用的部分2.0 推荐用select()来构造查询执行后用scalars()取出模型对象列表from sqlalchemy import select with SessionLocal() as session: # 查全部 users session.scalars(select(User)).all() # 条件过滤 users session.scalars(select(User).where(User.age 30)).all() # 排序 限制数量 users session.scalars( select(User).where(User.age 18).order_by(User.id.desc()).limit(10) ).all() # 按主键取单条 user session.get(User, 1)session.get(User, 1)是按主键取对象最简单也最常用没有匹配结果时返回None。where支持多个条件多个条件之间是 AND 关系。如果要 OR 条件用or_函数包起来from sqlalchemy import or_ users session.scalars( select(User).where(or_(User.age 20, User.age 60)) ).all()一个容易混淆的地方session.execute(select(User))返回的是Row对象列表要用.scalars()才能拿到User对象列表。如果只取几个列可以直接写session.execute(select(User.name, User.email))返回元组列表适合做列表展示不用把整个对象捞出来。统计场景用select(func.count()).select_from(User)配合group_by可以做分组统计。3.4 更新与删除更新在 ORM 里很直观查出来再改属性commit 即可with SessionLocal() as session: user session.get(User, 1) if user: user.age 31 session.commit()不查询直接批量更新用update()语句适合把满足条件的记录统一改个状态这种场景from sqlalchemy import update with SessionLocal() as session: session.execute(update(User).where(User.id 1).values(age32)) session.commit()删除同理user session.get(User, 1) session.delete(user) session.commit()说一个容易踩的坑更新对象后如果忘了 commit容易在不同代码块里看到改了但没生效的假象排查半天发现是事务没提交。Session 默认是自动提交关掉的所以一定要显式 commit。另一个坑是删除有外键关联的数据时如果关联表里还有引用会报IntegrityError这时候要先处理关联数据或者设计级联删除别硬删。4. 关系映射与进阶实战4.1 一对多与多对多关系表之间关系是 ORM 的核心价值也是它比手写 join 舒服的地方。一个用户有多篇文章典型的一对多。定义方式from sqlalchemy import ForeignKey from sqlalchemy.orm import relationship class Article(Base): __tablename__ articles id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200)) user_id: Mapped[int] mapped_column(ForeignKey(users.id)) user: Mapped[User] relationship(back_populatesarticles) # 在 User 类里补充 articles: Mapped[list[Article]] relationship(back_populatesuser)ForeignKey建立物理外键关联relationship建立 ORM 层面的对象关系。有了这个查询时可以直接访问user.articles拿文章列表或者article.user拿作者信息不用手写 join。这里注意back_populates要两边都写名字对得上才联得起来写漏一边关系就是单向的某些操作会失灵。多对多关系比如用户和文章的点赞关系需要一张中间关联表用Table定义两个模型分别声明relationship(secondary关联表)。实操中把关联表的唯一约束建好避免重复数据。关系定义本身不难难的是理解加载策略。4.2 常用查询技巧速查我把日常最常用的查询方法整理成表方便索引需求写法取所有记录select(User).all()条件过滤.where(User.age 18)多条件 AND.where(User.age 18, User.city 北京)或条件.where(or_(条件1, 条件2))排序.order_by(User.id.desc())分页.offset(20).limit(10)按主键取session.get(User, 1)统计总数select(func.count()).select_from(User)去重.distinct()分页这个值得展开.offset(20).limit(10)翻译成 SQL 是LIMIT 10 OFFSET 20数据量大时 offset 越深越慢因为数据库要扫描并跳过前面所有行。几万条内无所谓百万级以上可以考虑用主键游标分页where id 上次最后一条id。这个优化我在一个后台管理模块里实测过翻到第 100 页时查询时间从 1.2 秒降到 50 毫秒左右差距相当可观。4.3 事务与会话生命周期管理事务是数据库操作的基本保障。Session 的commit()提交事务rollback()回滚。一个典型事务处理场景with SessionLocal() as session: try: session.add(user) session.add(article) session.commit() except Exception: session.rollback() raise只要中间的某个操作失败rollback()会把之前所有未提交的改动撤销避免半截数据落库。做转账、订单这类涉及多张表的业务这个结构是底线。反过来如果你不 try 也不 rollback一个失败的操作可能让 session 处于半死不活的状态后续操作全部异常排查起来血压拉满。实际项目里建议给会话管理加一层统一封装提供一个获取 session 的入口。Flask 里用flask-sqlalchemy的db.sessionFastAPI 里用依赖注入拿 session。核心原则是每个请求对应一个 session请求结束关闭不要跨线程共享 session。跨线程共享 session 会出现各种诡异问题比如数据没提交就看不见、状态错乱新手很容易在这上面栽跟头。5. 常见问题与避坑指南5.1 懒加载与 N1 查询陷阱前面提过懒加载。再补充一个现象开启 ORM 的懒加载后一旦 session 关闭再访问未加载的属性会报DetachedInstanceError: Instance is not bound to a Session。这是新手踩得最多的坑之一。查出来的对象在with块里好好的一出来就报错原因就是 session 关闭后对象变成了游离态。解决思路有三种要么在 session 内提前用selectinload把关联数据加载好要么用session.expunge_all()把对象变成普通对象后保存必要字段要么干脆取需要的字段组建普通的 Python 对象或字典返回。我看很多教程推荐第三种虽然麻烦点但数据边界最清晰。别指望 session 关闭后还能像热数据一样随意访问这个认知是每个 ORM 使用者必须建立的。N1 查询的问题也在这里。访问user.articles时 SQLAlchemy 会即时发一条新 SQL 去查文章开发时方便但如果你循环打印 100 个用户的文章列表就会执行额外 100 条 SQL。解决办法是查询时用selectinload预加载from sqlalchemy.orm import selectinload users session.scalars( select(User).options(selectinload(User.articles)) ).all()这样一条主查询加一条批量查询就能拿到所有数据。建议在查询方法里显式声明需要的关联而不是全都依赖懒加载。5.2 大批量数据插入的性能优化ORM 插入几十万条数据会很慢因为每条都要经过对象创建、状态跟踪、Flush 等环节。我实测过50 万条数据用add_all大约需要 20-30 秒而用 Core 的insert批量模式只需要 3-5 秒差距非常明显。批量场景建议直接上 Corefrom sqlalchemy import insert data [ {name: fuser_{i}, email: fuser{i}example.com, age: 20} for i in range(10000) ] with engine.begin() as conn: conn.execute(insert(User), data)engine.begin()自动开启事务并在结束时 commit传入字典列表做批量写入效率比 ORM 逐条高得多。如果数据量真的特别大建议分批处理每批 1000-5000 条再加个进度打印方便观察任务跑到了哪。这里有个细节批量插入字典里的字段必须跟表完全对应传多了会报错传少了用默认值所以构造数据时要把字段对齐。5.3 连接池与超时问题engine 自带连接池默认pool_size5。高并发小项目经常遇到TimeoutError: QueuePool limit of size 5 overflow 10 reached说明连接被占满。处理方式加大连接池、启用pool_pre_ping每次取连接前先探测可用性。MySQL 默认的wait_timeout可能是 8 小时连接闲置过久会被服务端断开下次复用时直接报错。pool_pre_pingTrue能自动识别失效连接并重建这个参数我几乎是必开的。完整配置engine create_engine( mysqlpymysql://user:passlocalhost:3306/blog?charsetutf8mb4, pool_size10, max_overflow20, pool_pre_pingTrue, )另外如果你遇到Lost connection during query或者server has gone away除了连接池问题还要检查是不是单条 SQL 执行时间超过了数据库的超时阈值。做大数据量查询时该分批就分批别一条 SQL 跑几分钟。5.4 常见报错速查表报错信息原因解决ModuleNotFoundError: No module named pymysql没装驱动pip install pymysqlDetachedInstanceErrorsession 关闭后访问懒加载属性用 selectinload 预加载或提前取出数据IntegrityError唯一约束冲突 / 外键约束失败检查数据提交前 try/rollbackOperationalError: (2002, ...)数据库服务未启动或地址不通检查服务与连接串TimeoutError: QueuePool limit连接池耗尽调大 pool_size / max_overflowUnicodeDecodeError / 乱码字符集不匹配连接串写charsetutf8mb4表也建 utf8mb4IntegrityError这个值得多说两句。唯一键冲突时session 会进入失败状态如果不 rollback后续操作都会异常。所以涉及到唯一约束的插入一定要 try 一下冲突时 rollback 再走更新或跳过逻辑。爬虫去重场景我就在 6.1 里给你一个可直接抄的写法。6. 实战用 SQLAlchemy 存储爬虫数据6.1 模型设计与去重逻辑配合SQLAlchemy 储存爬虫数据这个高频需求给一个可落地的完整示例。基本流程是爬虫拿到数据 → 清洗 → 写入 MySQL。假设爬的是公开的图书信息模型如下class Book(Base): __tablename__ books id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) isbn: Mapped[str] mapped_column(String(20), uniqueTrue) title: Mapped[str] mapped_column(String(200)) price: Mapped[float] mapped_column(nullableTrue) source_url: Mapped[str] mapped_column(String(500)) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.now)存储时注意两点一是用isbn做唯一键重复抓取时先查询再决定 insert 还是 skip今天爬到的和昨天重复的页面不会产生垃圾数据二是created_at默认取当前时间入库时不用手动传。判断重复的代码existing session.execute( select(Book.id).where(Book.isbn book[isbn]) ).first() if not existing: session.add(Book(**book)) else: session.rollback() # 或者什么都不做跳过这一条这里有个小技巧用Book(**book)直接把字典展开成关键字参数构造模型前提是字典的键跟模型字段名完全一致。所以爬虫解析完数据后先做一个字段名对齐的清洗步骤把书名改成title、价格改成price后面入库就顺了。别小看这一步字段对不齐是爬虫入库最常见的报错来源。6.2 批量写入与提交策略爬虫任务往往是持续抓取几百上千条积累起来可以积攒到一定量再批量入库。批量任务建议每 500 条 commit 一次避免长时间事务占用连接。如果整批写入后某条数据违反唯一约束会导致整批回滚所以去重最好前置——在入库前就用集合把已存在的 isbn 过滤一遍减少数据库回滚的次数。一个提升效率的写法先查一次库把已有的 isbn 取成一个 set然后内存里过滤掉重复项最后一次性批量插入。这种方式比每条都去数据库查一次快得多尤其适合一次性导入几千条的初始化任务。我做过一次爬虫数据迁移3 万条数据用这个思路导入去重加写入总共花了几秒比逐条判断快了一个量级。日常开发中我还习惯搭配 Navicat 这类图形化客户端配合排查。起一个简单的查询接口或者直接连库看表结构、确认数据落库情况比在代码里敲 Python 交互式查表方便。注意给只给需要的账号配只读权限避免手滑改坏数据这也是团队协作里数据库安全的基本习惯。结尾说到这儿聊点个人体会。SQLAlchemy 真正上手之后你会发现写数据相关代码的体验比裸 SQL 舒服太多不用再手动拼接查询条件、不用担心注入、切换数据库也很从容。但我建议你依然保持对 SQL 本身的敏感度——ORM 只是帮你生成 SQL你最好知道它生成的 SQL 长什么样。调试时开echoTrue看几眼时间久了就能建立对象操作 ↔ SQL的映射感遇到性能问题也更容易定位。最后再分享一个实用小技巧从select(User)这种查询里获取模型列表时记得scalars().all()和execute().all()的区别。前者返回User对象列表后者返回Row元组列表。很多新手困惑查出来怎么不是对象多半是用了后者没加scalars()。这两种方式各有适用场景淘清了写起来就顺了。SQLAlchemy 内容很深这篇覆盖的是我平时用得最多的部分。往后再碰到分库分表、异步 ORMasyncpg加 SQLAlchemy 异步模式、数据库迁移这类进阶主题再单独开篇聊。数据操作是几乎所有 Python 项目的底座把这层打稳后面做爬虫、做数据分析、做后端接口都会顺畅很多。
