写Python的人早晚都要面对数据库。前两天还有个做爬虫的朋友跟我吐槽数据抓下来容易存进去反而麻烦。字段多了要拼接SQL字符串字段改了要回头改一大段插入语句一个引号没转义整条SQL直接报错好不容易跑通又发现数据重复入库。我说你早就该用SQLAlchemy ORM了。这篇文章就把我平时用的那套东西完整写出来从建表、会话管理、查询、爬虫入库到异步操作全部覆盖。如果你是刚接触Python数据库操作的初学者或者一直手写SQL想换一种更稳的方式再或者写爬虫需要一套能直接照搬的存储模板这份指南都可以直接参考。1. 从手写SQL到ORM数据库操作的核心痛点到底在哪1.1 一次SQL拼接事故引发的思考我早期写脚本时也干过这种事name 张三 age 28 sql fINSERT INTO user(name, age) VALUES({name}, {age}) cursor.execute(sql)看起来没什么问题但一旦name里带了个单引号SQL直接炸再往后字段越来越多增删字段时所有SQL语句都要跟着改一遍。最怕的还是让别人传参进来万一参数里带点恶意内容整条语句就会变成注入入口。后来我换成了参数化查询问题少了但代码量依然可观每个表都要写一套重复的CRUD。真正让我下定决心用ORM的是一次字段变更。业务方要求给用户表加一个avatar_url字段手写SQL的项目里要改建表脚本、插入语句、查询语句、更新语句稍漏一处就是线上事故。而用ORM的项目里只需要改一行模型类配合迁移工具自动生成同步脚本十分钟搞定。1.2 ORM不是把SQL藏起来而是把重复劳动收走很多新手误以为学了ORM就不用懂SQL了这是最大的误解。SQLAlchemy ORM做的事情是把数据库里的表映射成类把行记录映射成对象把列映射成属性。你在代码里操作对象框架在底层自动生成SQL。可以把它理解成“翻译官”你还是得知道SQL大概是怎么执行、索引怎么生效、事务怎么提交只是不用再亲手去写那些极其重复的INSERT、SELECT、UPDATE。对比维度手写SQLSQLAlchemy ORM开发速度慢字符串拼接易错快面向对象操作字段变更所有SQL同步改只改模型类迁移工具管理SQL注入风险容易踩坑内置参数化查询默认安全跨数据库切换每个数据库语法不同改动大统一ORM接口切换成本低复杂查询控制力完全可控支持原生SQL兜底1.3 为什么是SQLAlchemy而不是其他ORMPython生态里还有Django ORM、Peewee等但它们大多和框架绑定或者面向轻量场景。SQLAlchemy最大的优势是不绑定任何Web框架Flask、FastAPI、纯脚本、爬虫项目都能用。它同时提供了Core和ORM两层前者偏SQL构建后者偏对象映射想精细控制时还能用text()写原生SQL灵活度很高。所以很多数据量不大不小、框架自由度高的项目最终都会选它。2. 环境准备与第一张表Engine、模型类、建表动作的完整拆解2.1 安装与连接串先从SQLite起步SQLAlchemy的安装非常简单pip install sqlalchemy没有额外依赖时默认支持SQLite。这也是我最推荐的入门路径不需要安装数据库服务一个文件就能把整个机制跑通。from sqlalchemy import create_engine engine create_engine(sqlite:///demo.db, echoTrue)这里的engine是数据库连接的总入口它并不真正马上建立连接而是维护了一个连接池等你要执行SQL时才从池里取连接。参数echoTrue会把生成的SQL打印到控制台调试阶段强烈建议打开你会直观看到ORM在背后生成了什么语句。如果是MySQL或PostgreSQL连接串要写成这样# MySQL engine create_engine(mysqlpymysql://root:passwordlocalhost:3306/mydb) # PostgreSQL engine create_engine(postgresqlpsycopg://root:passwordlocalhost:5432/mydb)格式很好记数据库驱动://用户名:密码主机地址:端口/库名。2.2 声明式模型用类来定义表结构SQLAlchemy 2.0推荐用声明式Base来定义模型。先创建Base再让每个模型类继承它from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String class Base(DeclarativeBase): pass class User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue) name: Mapped[str] mapped_column(String(50), uniqueTrue) age: Mapped[int] mapped_column(default0)2.0风格里Mapped[int]、Mapped[str]这种类型注解让IDE能正确推断出user.name是字符串代码补全和静态检查都舒服很多。mapped_column()里再补充列参数比如主键、唯一约束、默认值。老版本的写法是Column(name, String(50))现在依然兼容但新项目我建议直接用2.0风格代码更简洁。2.3 建表create_all不是银弹定义好模型之后建表只要一行Base.metadata.create_all(engine)它会把所有还没有创建的表全部创建出来。但这里有个大坑如果表已经存在create_all不会去修改这个表的任何结构。你往模型类里加了一个新字段再次运行create_all数据库里的表并不会自动加列。这是新手最容易困惑的地方以为跑一下建表代码就同步了实际完全没有。所以我的用法是原型验证阶段用create_all省事正式项目一定要配Alembic迁移工具通过生成迁移脚本来管理表结构的演进。这个结论我踩了几次坑后才真正接受。2.4 第一个CRUD别把commit当摆设有了表之后来一次完整的插入、查询、修改、删除from sqlalchemy.orm import Session with Session(engine) as session: user User(name张三, age28) session.add(user) session.commit() print(user.id)Session是执行ORM操作的核心对象它负责跟踪所有对象的状态变化。add只是把对象放进session的跟踪列表真正生成INSERT SQL并执行的是commit()。如果不调用commitsession关闭时所有改动都会悄无声息地丢弃。如果想让对象马上获得自增主键可以先session.flush()它会立即发送SQL把数据写入数据库但事务还不会结束因此还能灵活回滚。刚开始学习时只需记住一次完整的写操作以add开始以commit收尾。3. Session生命周期事务提交、回滚和那些“数据神秘消失”的瞬间3.1 Session不是连接池是工作单元很多人会把Session和数据库连接搞混。实际上Session更像是“工作单元”它内部从engine的连接池里拿连接并负责把多个对象操作打包成一个事务。打个比方Session像一张草稿纸你在纸上涂涂改改数据库完全不知道只有把草稿纸交给老师批改commit才算正式生效。如果不想要这页内容可以直接撕掉rollback老师那边自然什么都没发生。这个机制带来一个很大的好处一次业务操作如果涉及多张表的写入只要最后统一commit要么全部成功要么全部回滚不会出现半个状态。3.2 用上下文管理器把生命周期固定住我强烈建议不要手写session Session(engine)然后到处传参而是用with上下文管理with Session(engine) as session: user User(name李四, age30) session.add(user) session.commit()with块退出时session会自动关闭。如果你还需要事务边界更清晰可以这样with Session(engine) as session: with session.begin(): session.add(user1) session.add(user2)内层begin()的作用是在这个块内除非代码主动commit()否则块结束时字段自动回滚。这种方式对“要么都成功要么都失败”的业务非常合适。3.3 三个高频坑位第一个坑是忘记commit。代码跑完没有报错数据却没写进去就是少了commit。在交互式环境里尤其容易发生。第二个坑是访问已过期或已脱离session的对象。默认情况下commit()之后session会把对象标记为“expired”下次访问它的属性时会自动发SQL回去查库。这本来是为了获取最新数据但如果对象已经被移出sessiondetached状态再访问属性就会抛出DetachedInstanceError。解决办法是在配置session时设置expire_on_commitFalse或者确保在session内完成所有属性访问。第三个坑是长事务。爬虫常犯开启一个session循环抓取几十上百条数据每抓一条入库一次整个session持续很长时间。这会导致数据库连接一直被占用并发一高就容易堆积连接还会让锁的持有时间变长。正确做法是快进快出每一次小批量任务用独立的session或者至少定时commit并在完成后立刻关闭。3.4 并发写入与数据丢失两个线程同时读取同一条记录各自修改后提交后提交的会覆盖先提交的这就是经典的丢失更新问题。ORM里可以用乐观锁解决在模型里加一个version_id_col提交时版本号不匹配就抛异常重试即可。也可以使用悲观锁SELECT ... FOR UPDATE但需要底层数据库支持。这部分内容很容易被人忽略等真正出问题再来补就痛苦了。4. 查询API实战过滤、联表、分页与N1的根治方法4.1 用select().where()替代传统的query写法SQLAlchemy 2.0推荐使用select()构造查询from sqlalchemy import select with Session(engine) as session: stmt select(User).where(User.age 18).order_by(User.id.desc()) users session.scalars(stmt).all()关键点在于session.scalars(stmt)。它返回一个ScalarResult对象.all()会把结果以模型对象列表返回。如果你写的是session.execute(stmt)拿到的则是包含元组的Result还要再取.scalars()转换。很多教程没讲清这一步导致新手总是看到一堆元组而不是对象。4.2 filter_by和filter别再傻傻分不清SQLAlchemy查询里有两个长得很像的过滤方法。我用一句话总结filter_by()只适合简单等值判断参数不带模型类名filter_by(name张三)filter()更全能可以写任意表达式filter(User.name 张三, User.age 18)实际项目中我基本只用filter()因为迟早会遇到大于、小于、IN、LIKE这类条件统一用filter()写法更一致也避免两种API混用导致的混乱。4.3 联表与懒加载的N1问题这个坑几乎每个用ORM的人都会踩。假如有Blog和User两张表Blog.author_id关联User.id你通常会这样关联查询with Session(engine) as session: blogs session.scalars(select(Blog).limit(10)).all() for blog in blogs: print(blog.author.name)表面看只执行了一条查询实际却执行了11条第一条查博客列表后面10条分别在访问blog.author时去查对应的用户。这就是“N1查询”。数据量到一定规模接口会肉眼可见地变慢。根治方案是在主查询里用joinedload或selectinload预加载关联对象from sqlalchemy.orm import selectinload stmt select(Blog).options(selectinload(Blog.author)).limit(10)这样一个主查询加一个WHERE id IN (...)的副查询就把关联数据全取回来了。至于怎么选joinedload用SQL的LEFT JOIN适合一对一或少量关联selectinload用IN查询适合一对多且关联数据较多的场景。我默认优先selectinload它在大多数情况下更可控。4.4 聚合、分组与分页查询统计时用func:from sqlalchemy import func stmt ( select(User.age, func.count(User.id)) .group_by(User.age) ) rows session.execute(stmt).all()分页方面最常用的还是stmt select(User).order_by(User.id).offset(20).limit(10)但要注意offset越大数据库扫描并丢弃的行越多。数据量较大时建议用“keyset分页”记录上一页最后一条记录的id下一页直接where(User.id last_id)。虽然写法上稍微复杂一点但性能稳定得多。5. 爬虫数据存储模板去重、批量写入和多线程会话隔离5.1 设计一张带唯一约束的存储表热搜词里频繁出现“sqlalchemy储存爬虫数据”说明这确实是很多人的刚需。爬虫数据的核心诉求通常有三个接口字段不固定、数据量大、重复抓取。用一个Article模型示例from datetime import datetime from sqlalchemy import String, Text, DateTime class Article(Base): __tablename__ articles id: Mapped[int] mapped_column(primary_keyTrue) url: Mapped[str] mapped_column(String(500), uniqueTrue) title: Mapped[str] mapped_column(String(200)) content: Mapped[str] mapped_column(Text) crawled_at: Mapped[datetime] mapped_column(defaultdatetime.utcnow)url字段设置了uniqueTrue这一步就是数据库层面的“去重保险”比单纯的代码判断可靠得多。5.2 时间去重与批量插入最朴素的“先查再插”方式是这样def save_articles(session, items): for item in items: exists session.scalar( select(Article).where(Article.url item[url]) ) if not exists: session.add(Article(**item)) session.commit()但每条数据都要先查一次效率不算高。数据量大时我建议改用数据库方言的“冲突忽略”或“冲突更新”能力。比如SQLite可以使用from sqlalchemy.dialects.sqlite import insert as sqlite_insert stmt sqlite_insert(Article).values(list_of_dicts) stmt stmt.on_conflict_do_nothing(index_elements[url]) session.execute(stmt) session.commit()这一条语句就能把整批数据插入遇到url重复的直接跳过不需要先查再插。如果你用的是PostgreSQL对应使用postgresql.insert的on_conflict_do_nothing。这属于方言级API但为了性能完全值得学。5.3 多线程爬虫的会话隔离写并发爬虫时最大的一个坑就是Session对象不是线程安全的。同一个session被多个线程同时add、commit轻则报错重则数据错乱。正确做法是每个线程独立创建自己的session。使用sessionmaker工厂from sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine) def crawl_worker(urls): session SessionLocal() try: for url in urls: # 抓取、清洗、入库 pass session.commit() except Exception: session.rollback() raise finally: session.close()也可以配合scoped_session来按线程或协程隔离session但本质上还是要记住session是脆弱的别共享。5.4 一个带网络请求的完整流程参考结合httpx抓取公开测试页面的流程可以这样设计import httpx from sqlalchemy import create_engine, select engine create_engine(sqlite:///crawler.db) Base.metadata.create_all(engine) SessionLocal sessionmaker(bindengine) def fetch_and_store(): with httpx.Client(timeout10) as client: resp client.get(https://example.com/api/list) resp.raise_for_status() data resp.json() with SessionLocal() as session: for row in data.get(items, []): if session.scalar(select(Article).where(Article.url row[url])): continue session.add(Article(**row)) session.commit()这个模板足够改造成大多数内容型爬虫的存储层。需要注意的是正式项目建议加异常捕获和重试逻辑不要把httpx的请求异常直接抛到入库环节。6. 异步数据库操作同步与异步的适用边界与坑位盘点6.1 为什么异步话题越来越热看了后台的热搜词“python postgresql sqlalchemy 异步 同步 比较”和“sqlalchemy psycopg3 异步 同步 比较”频繁出现。这背后的原因很现实FastAPI等异步框架越来越主流很多接口本身是异步的如果数据库操作还是同步阻塞就等于把异步带来的并发收益全部丢掉。SQLAlchemy从1.4开始引入异步支持到2.0已经相当成熟。基本依赖是pip install sqlalchemy[asyncio] greenlet同时数据库驱动要选带异步能力的PostgreSQL用asyncpg或psycopg3的异步模式MySQL用aiomysqlSQLite用aiosqlite。6.2 异步engine与异步session的写法异步版本和同步版非常相似只是把create_engine换成create_async_engine把Session换成AsyncSession执行时加上awaitfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession engine create_async_engine(postgresqlasyncpg://user:passlocalhost:5432/mydb) async def main(): async with AsyncSession(engine) as session: result await session.execute(select(User).where(User.age 18)) users result.scalars().all()注意连接串里驱动名变了同时面对asyncpg和psycopg3时我建议优先选psycopg3的异步模式因为它在PostgreSQL生态里更活跃和SQLAlchemy的兼容性也更好。6.3 同步还是异步一张表看明白场景同步异步一次性脚本、定时任务推荐简单直接没必要徒增复杂度普通爬虫并发不高推荐配合线程池可以但收益有限高并发爬虫大量IO等待线程占用高成本大推荐协程开销低FastAPI等异步Web接口会阻塞事件循环推荐数据量小而逻辑复杂的业务同步更易调试异步排错成本高我的建议很明确不要为了异步而异步。如果你的程序本身是同步的硬改成异步会让错误排查变得更麻烦。真正的场景一定是IO密集、并发高或者项目框架本身就是异步模型。6.4 异步中的两个经典坑第一个坑AsyncSession里触发懒加载会直接报错。类似blog.author.name这种访问底层会同步执行SQL但异步session里根本没有同步连接可用。解决方法是查询时就用selectinload把关联对象预加载好这一点比同步场景更加严格。第二个坑异步协程里混用同步session。有人会在异步任务里调用一个普通函数这个普通函数内部又创建了同步Session去查库。一旦这个函数被多个协程同时调用数据库连接可能会被打满而且事件循环会被阻塞。如果用了异步框架数据库访问层就要保持全异步。从我个人的实践经验来说先踏踏实实把同步版跑通再过渡到异步版比一上来直接写异步要稳得多。毕竟大多数爬虫和脚本场景同步代码可读性更好排查问题也更省心。等真正需要面对高并发的API服务时再把异步三件套——create_async_engine、AsyncSession、selectinload——熟练用起来你会觉得SQLAlchemy这套设计确实香。