1. 先搞清楚SQLAlchemy凭什么成为Python ORM的头号选择1.1 ORM到底帮你省了什么事很多刚开始接触Python的朋友写了两三个月脚本数据库操作还在靠拼 SQL 字符串cursor.execute(SELECT * FROM user WHERE age %s % age)这种写法最难受的地方有两个一个是SQL语句和Python代码互相嵌套逻辑一复杂就分不清谁是谁另一个是换数据库的时候比如从SQLite换到MySQLSQL方言的差异会让你改一堆代码。ORM对象关系映射做的事情其实很简单把数据库里的表映射成Python里的类把表里的一行记录映射成类的一个实例把表之间的外键关系映射成对象之间的引用关系。SQLAlchemy作为Python生态里最流行的ORM框架最大的功劳就是让开发者可以用写Python对象的方式去操作数据库而不需要整天惦记着CREATE TABLE、INSERT INTO这些原生SQL语法。你定义好一个类它就能帮你把表建好你操作一个对象它就能帮你生成对应的增删改查语句。有人可能会说ORM多了一层转换性能一定差吧这话对了一半。SQLAlchemy的底层并不是一个万能翻译器它本身就有非常优秀的SQL生成和连接池管理机制。你可以用ORM写业务逻辑遇到复杂的查询场景还能随时切换到SQLAlchemy Core甚至直接用原生SQL。这一点是很多轻量级ORM做不到的。1.2 SQLAlchemy能流行靠的不只是资历老SQLAlchemy第一个正式版本发布已经很多年了在Python的ORM生态里算是元老级的存在。但资历老并不是它站稳脚跟的核心原因。仔细观察市面上其他Python ORM比如Django ORM、Peewee、Tortoise ORM你会发现它们各自都有明显的倾向性Django ORM是跟着Django框架走的脱离Django很难单独用模型定义也和Django的生态深度绑定。Peewee走的是轻量路线上手快但复杂查询和模型关系处理能力偏弱适合小型工具项目。Tortoise ORM主打异步与FastAPI这类异步框架配合得不错但生态和文档成熟度跟SQLAlchemy比还是有差距。SQLAlchemy走的是一条通吃路线。它既有Core这种偏底层、接近SQL的查询构建方式又有ORM这种完全面向对象的操作方式你可以在同一个项目里按需混用。数据库方面更是几乎覆盖了主流关系型数据库PostgreSQL、MySQL、SQLite、Oracle、SQL Server、甚至ClickHouse都有对应的方言实现。这意味着你在一家公司用SQLAlchemy写的数据库访问代码换到另一家换了数据库的公司几乎可以做到零成本迁移只需要改一下连接字符串。我从很多年前开始就不再手动拼SQL了个人感受最深的不是省了多少代码量而是代码的可维护性提升了几个档次。模型即表结构字段写了就是列关系用relationship声明一下就能自动管理外键整个项目的数据层看起来就像一个标准的Python模块新人接手也容易上手。1.3 Core和ORM双轨设计才是真正的分水岭SQLAlchemy内部其实分两层Core层提供Table、Column、select()、insert()等组件让你用Python表达式构造SQL语句但返回的仍然是类似结果集的东西不涉及对象映射。ORM层在Core之上构建了Mapped、relationship、Session等机制把表结构映射为Python类把查询结果映射为对象实例。很多用了一段时间SQLAlchemy的人可能只接触过ORM层根本没意识到Core的存在。但实际上ORM的底层就是Core你通过ORM写的操作最终都要经过Core来生成SQL。这种双轨设计最大的好处是灵活性和性能之间的平衡。比如一个统计报表需要聚合多张表、嵌套子查询、按年月分组用ORM写会非常别扭生成的SQL也不一定高效。这时候直接用Core的select()写法从代码上看还是一套风格但你已经完全掌控了SQL的生成过程from sqlalchemy import select, func stmt ( select(User.name, func.count(Order.id).label(order_count)) .join(Order, Order.user_id User.id) .group_by(User.id) .having(func.count(Order.id) 10) ) result session.execute(stmt)这段代码既能让你享受Python表达式带来的语法高亮和自动参数绑定又不会像拼字符串那样有SQL注入的风险。不少从Java那边转过来的朋友看到这个设计都会联想到MyBatis和Hibernate的关系——如果你需要贴近SQL就用Core如果你想快速写业务就用ORM。一个框架覆盖两种诉求这在Python生态里几乎没有对手。2. 模型定义、Session生命周期和查询套路2.1 用声明式方式快速定义一张表SQLAlchemy 2.x之后官方推荐的写法是声明式映射Declarative Mapping最直观的变化就是配合类型注解来定义模型from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Integer, DateTime, func from datetime import datetime class Base(DeclarativeBase): pass class Article(Base): __tablename__ articles id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200), nullableFalse, comment标题) author: Mapped[str] mapped_column(String(50), default匿名) view_count: Mapped[int] mapped_column(Integer, default0) created_at: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now())Mapped[str]这种写法第一次见可能会觉得有点绕但实际上就是告诉SQLAlchemy这个属性对应数据库里的什么类型。类型注解本身就是文档IDE也能自动补全比老式的Column(String(200))清爽不少。mapped_column里支持的参数基本上和传统Column一致nullable、default、server_default、comment、index这些都能用。有一点要注意default是Python端的默认值在ORM插入时会自动填充server_default是数据库端的默认值比如func.now()表示由数据库在插入时生成时间两者概念不同混用的时候要想清楚。定义好模型之后创建表只需要一句话from sqlalchemy import create_engine engine create_engine(sqlite:///./blog.db) Base.metadata.create_all(engine)create_all会检查表是否存在如果不存在就自动建表。对于开发环境非常方便但生产环境建议还是用Alembic这类迁移工具来管理表结构变更避免某次改动模型后create_all不会自动同步已有表结构的问题。2.2 Session理解它就是理解SQLAlchemy的事务模型Session是SQLAlchemy ORM里最容易让新手困惑的东西。我见过很多初学者把Session当成连接池里的一个连接这是一个很有迷惑性的类比。更准确的说法是Session是一个工作单元Unit of Work的容器它管理着一批对象的状态并在你需要的时候把对象的变更通过事务写入数据库。你在Session里add()一个对象它并不会立刻执行INSERT你在Session里修改一个对象的属性它也不会立刻执行UPDATE。直到你调用session.commit()SQLAlchemy才会把Session里所有待处理的对象变化统一生成SQL语句发送到数据库。这种延迟写入机制好处是你可以把一个业务流程里的多处数据修改攒在一起最后一次性提交既减少数据库往返又能保证事务的一致性。坏处是如果你不熟悉这个机制很容易出现我以为数据已经保存了结果退出程序数据丢了的尴尬——因为你忘了commit()。一个比较标准的使用方式是上下文管理器from sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine) with SessionLocal() as session: article Article(titleSQLAlchemy入门, author张三) session.add(article) session.commit()with session.commit()其实还隐含了异常时rollback()的逻辑。如果用普通的sessionmaker()而不加context就必须自己在try/finally里处理回滚。注意Session不是线程安全的。在Web框架里正确的做法是为每个请求创建一个新的Session而不是在全局共享同一个Session。Flask-SQLAlchemy和FastAPI的依赖注入机制都是基于这个原则设计的。2.3 查询接口select语法和链式调用SQLAlchemy 2.x的查询风格已经全面转向select()函数from sqlalchemy import select # 查询所有 stmt select(Article) articles session.scalars(stmt).all() # 带条件 stmt select(Article).where(Article.view_count 1000).order_by(Article.created_at.desc()) articles session.scalars(stmt).all() # 分页 stmt select(Article).limit(20).offset(40) articles session.scalars(stmt).all()session.scalars()返回的是标量结果每条记录对应一个模型对象如果你用session.execute()拿到的是Row对象需要通过row[0]或者row.XXX来取值。这个区别要记牢不然会经常遇到为什么我查出来的是元组不是对象的困惑。条件筛选时多个where()之间是AND关系如果遇到需要OR的场景要用or_from sqlalchemy import or_ stmt select(Article).where( or_(Article.title.like(%Python%), Article.author 张三) )有一点我特别推荐在调试的时候可以直接把SQLAlchemy生成的SQL打印出来方法是在连接串里加echoTrue或者用str(stmt)查看编译后的SQL。engine create_engine(sqlite:///./blog.db, echoTrue)篇幅所限这里不展开join、子查询、聚合等高级写法后面实操章节会重点演示几个真实场景。先说这么多是为了让大家对查询套路有个整体印象。3. 实操用SQLAlchemy搭一个爬虫数据存储项目3.1 环境准备Python、数据库、依赖网上关于python安装教程的热度一直很高这里简单提一下关键点。建议直接安装Python 3.10以上的版本在Windows上记得勾选Add Python to PATH否则后面用命令行执行pip会很痛苦。如果你用的是Linux服务器注意系统自带的Python版本可能偏低不要随便动系统Python用pyenv或者直接装个独立的Python环境是更稳妥的做法。依赖安装用pip就行pip install sqlalchemy pip install psycopg2-binary # 如果连PostgreSQL如果你只是本地练习SQLite不需要额外安装驱动Python自带sqlite3模块SQLAlchemy可以直接连接。如果练习的生产场景是PostgreSQL就需要psycopg2-binary或者新版psycopg3。psycopg3安装方式略有不同需要pip install psycopg[binary]然后连接串里写postgresqlpsycopg://user:passhost:port/dbname。很多新手在连接PostgreSQL时报错十有八九是用户名密码权限问题或者PostgreSQL默认端口5432被占用可以先在命令行用psql测试一下能否正常登录再回头查SQLAlchemy的连接串。3.2 设计一个爬虫数据表结构爬虫场景最典型的存储需求是抓取文章列表、去重、记录抓取时间、支持按来源和关键词检索。我直接演示一张articles表和一张crawl_records表的设计。from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship from sqlalchemy import String, Integer, DateTime, ForeignKey, Text, UniqueConstraint from datetime import datetime class Base(DeclarativeBase): pass class Article(Base): __tablename__ articles __table_args__ ( UniqueConstraint(source, url, nameuq_source_url), ) id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200), nullableFalse) url: Mapped[str] mapped_column(String(500), nullableFalse) source: Mapped[str] mapped_column(String(50), nullableFalse) content: Mapped[str] mapped_column(Text, default) author: Mapped[str] mapped_column(String(50), default) published_at: Mapped[datetime | None] mapped_column(DateTime, nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now()) crawl_records: Mapped[list[CrawlRecord]] relationship(back_populatesarticle) class CrawlRecord(Base): __tablename__ crawl_records __table_args__ ( UniqueConstraint(article_id, crawl_time, nameuq_article_crawl_time), ) id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) article_id: Mapped[int] mapped_column(ForeignKey(articles.id), nullableFalse) status: Mapped[str] mapped_column(String(20), defaultsuccess) crawl_time: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now()) duration_ms: Mapped[int] mapped_column(Integer, default0) article: Mapped[Article] relationship(back_populatescrawl_records)这里有两个设计细节值得说明用UniqueConstraint(source, url)给文章数据加了唯一约束这样即使爬虫脚本写得不小心重复抓取靠数据库层面也能挡住一部分重复数据。单独建一张crawl_records表记录每次爬取的耗时和状态好处是后续分析爬虫健康度、统计抓取成功率都有数据可查。这种主表记录表的结构在真实项目中非常常见比把所有字段塞进一张表优雅很多。3.3 连接数据库并写入数据选SQLite练习连接串最简单from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine create_engine(sqlite:///./crawler.db) SessionLocal sessionmaker(bindengine, autoflushFalse, expire_on_commitFalse) Base.metadata.create_all(engine)autoflushFalse是一个细节。默认情况下Session会在查询前自动把待处理的变更发送到数据库flush这有可能导致某些瞬间数据状态不是你预期的那样。很多团队习惯把autoflush关掉让flush只发生在显式调用session.flush()或session.commit()的时候行为更可控。写入数据的核心动作是批量插入def save_articles(session, items): items: list of dict for item in items: article Article( titleitem[title], urlitem[url], sourceitem[source], contentitem.get(content, ), authoritem.get(author, ), published_atitem.get(published_at), ) session.add(article) session.commit()如果数据量很大几个G的数据一次性add再commit会非常慢因为SQLAlchemy要跟踪每个对象的状态。更高效的做法是分批提交每500条或者每1000条提交一次def save_articles_batch(session, items, batch_size500): for i in range(0, len(items), batch_size): batch items[i:i batch_size] for item in batch: session.add(Article(**{k: v for k, v in item.items() if hasattr(Article, k)})) session.commit()这里用了hasattr(Article, k)做过滤防止传入的字典里有模型不认识的字段导致报错。批量插入还有一种方式是走Core层的insert()速度会更快因为绕过了ORM的实例化过程from sqlalchemy import insert stmt insert(Article).values([ {title: item[title], url: item[url], source: item[source]} for item in items ]) session.execute(stmt) session.commit()实际项目里我通常先用ORM方式写业务代码确定逻辑没问题后再把真正需要高性能的批量导入部分改成Core的insert()。这种先求正确、再求性能的思路能帮你避免一开始就为了优化性能而写出难以维护的代码。3.4 数据去重逻辑数据库唯一约束应用层判断爬虫最烦的问题就是重复抓取。我见过很多项目只靠先查数据库再决定是否插入来去重这在数据量小的时候还能凑合一旦并发量上来或者表数据过亿每次查询的代价就很高。最佳实践是唯一约束放在数据库层面兜底应用层只做轻量的查重缓存。from sqlalchemy.exc import IntegrityError seen_urls set() def save_articles_dedup(session, items): for item in items: if item[url] in seen_urls: continue article Article( titleitem[title], urlitem[url], sourceitem[source], ) session.add(article) seen_urls.add(item[url]) try: session.commit() except IntegrityError: session.rollback() # 说明有并发冲突或缓存失效可以再查一遍数据库 for item in items: exists session.execute( select(Article.id).where(Article.url item[url]) ).first() if not exists: session.add(Article( titleitem[title], urlitem[url], sourceitem[source], )) session.commit()这个写法看着繁琐但实际运行中能省掉大量无谓的查询。seen_urls集合在内存中处理相同URL的重复时几乎是零成本万一有漏网之鱼靠数据库的UniqueConstraint抛出的IntegrityError兜底回滚之后再做精确查重既保证数据不重复又不会因为并发冲突直接把整个爬虫程序搞崩。3.5 查询报表统计爬取趋势和热门来源数据存进去不是目的能查出来才有价值。用SQLAlchemy写两个报表查询先看每个来源的文章数量from sqlalchemy import select, func stmt ( select(Article.source, func.count(Article.id).label(cnt)) .group_by(Article.source) .order_by(func.count(Article.id).desc()) ) rows session.execute(stmt).all() for source, cnt in rows: print(source, cnt)再看最近一周每天抓取的文章量from datetime import datetime, timedelta week_ago datetime.now() - timedelta(days7) stmt ( select(func.date(Article.created_at).label(day), func.count(Article.id)) .where(Article.created_at week_ago) .group_by(func.date(Article.created_at)) .order_by(func.date(Article.created_at)) ) rows session.execute(stmt).all() for day, cnt in rows: print(day, cnt)func.date()是数据库函数在SQLite和PostgreSQL里都能正常工作。SQLAlchemy会在生成SQL时把它编译成对应数据库的日期函数这种跨数据库的能力就是我一直推荐它的原因。4. 常见问题与排查技巧实录4.1 N1查询ORM最容易踩的性能坑用ORM查列表时最常见的问题不是连接慢而是N1查询。看起来只发了一条查询实际上ORM在访问每个对象的关联属性时又各自发了一条SQL去数据库。举个例子stmt select(Article).limit(10) articles session.scalars(stmt).all() for article in articles: print(article.crawl_records[-1]) # 每访问一次都会发起一次查询这段代码会产生1条查询articles表的SQL加上10条查询crawl_records表的SQL总共11条。数据量小看不出问题数据量大了数据库会被拖垮。解决方案是使用selectinload或joinedload主动加载关联关系from sqlalchemy.orm import selectinload stmt select(Article).options(selectinload(Article.crawl_records)).limit(10) articles session.scalars(stmt).all()selectinload会先用一条IN查询把关联记录一次性取出来加载到内存里后续访问article.crawl_records就不会再触发SQL了。建议凡是能预见到要访问关联属性的场景查询时直接把关联关系用selectinload带上。宁可多查一条关联SQL也不要让ORM在循环里偷偷发N条查询。4.2 Session用完之后必须关闭或重置Web应用里最容易出现的内存泄漏问题之一就是把Session当成全局单例来用。Session内部持有大量已加载的对象引用如果长生命周期服务里一直复用同一个Session内存占用会只增不减。正确做法是依赖注入或中间件在每个请求开始时创建Session请求结束或异常时关闭。FastAPI里可以这么写from fastapi import Depends def get_db(): db SessionLocal() try: yield db finally: db.close()除了关闭还有一个很多人不知道的细节expire_on_commit。默认情况下session.commit()之后所有对象的属性都被标记为过期下次访问属性时会重新查询数据库。在sessionmaker里设置expire_on_commitFalse可以避免这种隐式查询尤其是在模板渲染场景里避免提交之后页面展示又要触发一次数据库查询的奇怪现象。4.3 模型字段名与数据库关键字的冲突给模型字段起名的时候如果取了order、group、desc这类在SQL里是关键字的名字生成SQL时很可能会报语法错误。SQLAlchemy对大多数情况下可以自动转义但在某些数据库方言下依然会出问题。建议一开始就给字段加上显式的列名映射from sqlalchemy import Column, String class Product(Base): __tablename__ products id: Mapped[int] mapped_column(Integer, primary_keyTrue) name: Mapped[str] mapped_column(String(100)) # 如果字段名叫 order则映射到 order_no 列 order_no: Mapped[str] mapped_column(order, String(50))这样Python里用product.order_no数据库里实际列名是order看起来绕了一圈但能避开关键字冲突。4.4 同步不是终点异步SQLAlchemy怎么选搜索热词里有比较多的sqlalchemy psycopg3 异步 同步比较可见不少人在纠结要不要上异步。结论可以直接说如果你的项目是FastAPI这类异步框架并且数据库交互是IO密集场景用异步SQLAlchemy可以提升并发吞吐量如果项目是传统Django、Flask同步接口或者主要写爬虫脚本就没必要硬上异步同步写法更简单调试成本也更低。SQLAlchemy 2.x的异步用法主要涉及两个地方from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker from sqlalchemy import text engine create_async_engine(postgresqlpsycopg://user:passlocalhost/db) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse) async def get_one(): async with AsyncSessionLocal() as session: result await session.execute(text(SELECT 1)) return result.scalar()注意异步的连接串要用postgresqlpsycopg或postgresqlasyncpg这样的异步方言同步的psycopg2-binary不支持异步。很多人换了异步写法却不换驱动结果死活连不上数据库报错信息还特别隐晦。4.5 调试技巧从日志到SQL编译最后分享一个实用技巧。排查SQLAlchemy性能问题的时候我通常会先开启SQL日志import logging logging.basicConfig() logging.getLogger(sqlalchemy.engine).setLevel(logging.INFO)这样控制台会打印出SQLAlchemy实际执行的每一条SQL语句包括耗时。如果你发现一个接口执行了十几条SQL基本就是N1问题没跑了。有时候想单独看某条语句编译后的SQL不用启动日志直接stmt select(Article).where(Article.title.like(%Python%)) print(stmt.compile(engineengine))这个方式特别适合调试动态查询确认自己拼出来的条件是不是符合预期。我个人在实际项目里遇到最多次的坑基本都是Session生命周期和N1问题。前者会让你的服务看起来偶尔卡顿后者会让你的数据库在数据量上来之后突然扛不住。如果你现在正在迁移到SQLAlchemy我建议先把这两个概念吃透再去研究各种高级查询写法。最后顺手推荐一个思路如果你只是写脚本、做数据分析用SQLite加同步SQLAlchemy就够了如果你的服务要扛线上流量至少把连接池、批量提交、异步驱动这几件事一起考虑进去。数据层是大多数应用的底座这一层稳了上面的业务逻辑才有得谈。
1. 先搞清楚SQLAlchemy凭什么成为Python ORM的头号选择1.1 ORM到底帮你省了什么事很多刚开始接触Python的朋友写了两三个月脚本数据库操作还在靠拼 SQL 字符串cursor.execute(SELECT * FROM user WHERE age %s % age)这种写法最难受的地方有两个一个是SQL语句和Python代码互相嵌套逻辑一复杂就分不清谁是谁另一个是换数据库的时候比如从SQLite换到MySQLSQL方言的差异会让你改一堆代码。ORM对象关系映射做的事情其实很简单把数据库里的表映射成Python里的类把表里的一行记录映射成类的一个实例把表之间的外键关系映射成对象之间的引用关系。SQLAlchemy作为Python生态里最流行的ORM框架最大的功劳就是让开发者可以用写Python对象的方式去操作数据库而不需要整天惦记着CREATE TABLE、INSERT INTO这些原生SQL语法。你定义好一个类它就能帮你把表建好你操作一个对象它就能帮你生成对应的增删改查语句。有人可能会说ORM多了一层转换性能一定差吧这话对了一半。SQLAlchemy的底层并不是一个万能翻译器它本身就有非常优秀的SQL生成和连接池管理机制。你可以用ORM写业务逻辑遇到复杂的查询场景还能随时切换到SQLAlchemy Core甚至直接用原生SQL。这一点是很多轻量级ORM做不到的。1.2 SQLAlchemy能流行靠的不只是资历老SQLAlchemy第一个正式版本发布已经很多年了在Python的ORM生态里算是元老级的存在。但资历老并不是它站稳脚跟的核心原因。仔细观察市面上其他Python ORM比如Django ORM、Peewee、Tortoise ORM你会发现它们各自都有明显的倾向性Django ORM是跟着Django框架走的脱离Django很难单独用模型定义也和Django的生态深度绑定。Peewee走的是轻量路线上手快但复杂查询和模型关系处理能力偏弱适合小型工具项目。Tortoise ORM主打异步与FastAPI这类异步框架配合得不错但生态和文档成熟度跟SQLAlchemy比还是有差距。SQLAlchemy走的是一条通吃路线。它既有Core这种偏底层、接近SQL的查询构建方式又有ORM这种完全面向对象的操作方式你可以在同一个项目里按需混用。数据库方面更是几乎覆盖了主流关系型数据库PostgreSQL、MySQL、SQLite、Oracle、SQL Server、甚至ClickHouse都有对应的方言实现。这意味着你在一家公司用SQLAlchemy写的数据库访问代码换到另一家换了数据库的公司几乎可以做到零成本迁移只需要改一下连接字符串。我从很多年前开始就不再手动拼SQL了个人感受最深的不是省了多少代码量而是代码的可维护性提升了几个档次。模型即表结构字段写了就是列关系用relationship声明一下就能自动管理外键整个项目的数据层看起来就像一个标准的Python模块新人接手也容易上手。1.3 Core和ORM双轨设计才是真正的分水岭SQLAlchemy内部其实分两层Core层提供Table、Column、select()、insert()等组件让你用Python表达式构造SQL语句但返回的仍然是类似结果集的东西不涉及对象映射。ORM层在Core之上构建了Mapped、relationship、Session等机制把表结构映射为Python类把查询结果映射为对象实例。很多用了一段时间SQLAlchemy的人可能只接触过ORM层根本没意识到Core的存在。但实际上ORM的底层就是Core你通过ORM写的操作最终都要经过Core来生成SQL。这种双轨设计最大的好处是灵活性和性能之间的平衡。比如一个统计报表需要聚合多张表、嵌套子查询、按年月分组用ORM写会非常别扭生成的SQL也不一定高效。这时候直接用Core的select()写法从代码上看还是一套风格但你已经完全掌控了SQL的生成过程from sqlalchemy import select, func stmt ( select(User.name, func.count(Order.id).label(order_count)) .join(Order, Order.user_id User.id) .group_by(User.id) .having(func.count(Order.id) 10) ) result session.execute(stmt)这段代码既能让你享受Python表达式带来的语法高亮和自动参数绑定又不会像拼字符串那样有SQL注入的风险。不少从Java那边转过来的朋友看到这个设计都会联想到MyBatis和Hibernate的关系——如果你需要贴近SQL就用Core如果你想快速写业务就用ORM。一个框架覆盖两种诉求这在Python生态里几乎没有对手。2. 模型定义、Session生命周期和查询套路2.1 用声明式方式快速定义一张表SQLAlchemy 2.x之后官方推荐的写法是声明式映射Declarative Mapping最直观的变化就是配合类型注解来定义模型from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, Integer, DateTime, func from datetime import datetime class Base(DeclarativeBase): pass class Article(Base): __tablename__ articles id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200), nullableFalse, comment标题) author: Mapped[str] mapped_column(String(50), default匿名) view_count: Mapped[int] mapped_column(Integer, default0) created_at: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now())Mapped[str]这种写法第一次见可能会觉得有点绕但实际上就是告诉SQLAlchemy这个属性对应数据库里的什么类型。类型注解本身就是文档IDE也能自动补全比老式的Column(String(200))清爽不少。mapped_column里支持的参数基本上和传统Column一致nullable、default、server_default、comment、index这些都能用。有一点要注意default是Python端的默认值在ORM插入时会自动填充server_default是数据库端的默认值比如func.now()表示由数据库在插入时生成时间两者概念不同混用的时候要想清楚。定义好模型之后创建表只需要一句话from sqlalchemy import create_engine engine create_engine(sqlite:///./blog.db) Base.metadata.create_all(engine)create_all会检查表是否存在如果不存在就自动建表。对于开发环境非常方便但生产环境建议还是用Alembic这类迁移工具来管理表结构变更避免某次改动模型后create_all不会自动同步已有表结构的问题。2.2 Session理解它就是理解SQLAlchemy的事务模型Session是SQLAlchemy ORM里最容易让新手困惑的东西。我见过很多初学者把Session当成连接池里的一个连接这是一个很有迷惑性的类比。更准确的说法是Session是一个工作单元Unit of Work的容器它管理着一批对象的状态并在你需要的时候把对象的变更通过事务写入数据库。你在Session里add()一个对象它并不会立刻执行INSERT你在Session里修改一个对象的属性它也不会立刻执行UPDATE。直到你调用session.commit()SQLAlchemy才会把Session里所有待处理的对象变化统一生成SQL语句发送到数据库。这种延迟写入机制好处是你可以把一个业务流程里的多处数据修改攒在一起最后一次性提交既减少数据库往返又能保证事务的一致性。坏处是如果你不熟悉这个机制很容易出现我以为数据已经保存了结果退出程序数据丢了的尴尬——因为你忘了commit()。一个比较标准的使用方式是上下文管理器from sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(bindengine) with SessionLocal() as session: article Article(titleSQLAlchemy入门, author张三) session.add(article) session.commit()with session.commit()其实还隐含了异常时rollback()的逻辑。如果用普通的sessionmaker()而不加context就必须自己在try/finally里处理回滚。注意Session不是线程安全的。在Web框架里正确的做法是为每个请求创建一个新的Session而不是在全局共享同一个Session。Flask-SQLAlchemy和FastAPI的依赖注入机制都是基于这个原则设计的。2.3 查询接口select语法和链式调用SQLAlchemy 2.x的查询风格已经全面转向select()函数from sqlalchemy import select # 查询所有 stmt select(Article) articles session.scalars(stmt).all() # 带条件 stmt select(Article).where(Article.view_count 1000).order_by(Article.created_at.desc()) articles session.scalars(stmt).all() # 分页 stmt select(Article).limit(20).offset(40) articles session.scalars(stmt).all()session.scalars()返回的是标量结果每条记录对应一个模型对象如果你用session.execute()拿到的是Row对象需要通过row[0]或者row.XXX来取值。这个区别要记牢不然会经常遇到为什么我查出来的是元组不是对象的困惑。条件筛选时多个where()之间是AND关系如果遇到需要OR的场景要用or_from sqlalchemy import or_ stmt select(Article).where( or_(Article.title.like(%Python%), Article.author 张三) )有一点我特别推荐在调试的时候可以直接把SQLAlchemy生成的SQL打印出来方法是在连接串里加echoTrue或者用str(stmt)查看编译后的SQL。engine create_engine(sqlite:///./blog.db, echoTrue)篇幅所限这里不展开join、子查询、聚合等高级写法后面实操章节会重点演示几个真实场景。先说这么多是为了让大家对查询套路有个整体印象。3. 实操用SQLAlchemy搭一个爬虫数据存储项目3.1 环境准备Python、数据库、依赖网上关于python安装教程的热度一直很高这里简单提一下关键点。建议直接安装Python 3.10以上的版本在Windows上记得勾选Add Python to PATH否则后面用命令行执行pip会很痛苦。如果你用的是Linux服务器注意系统自带的Python版本可能偏低不要随便动系统Python用pyenv或者直接装个独立的Python环境是更稳妥的做法。依赖安装用pip就行pip install sqlalchemy pip install psycopg2-binary # 如果连PostgreSQL如果你只是本地练习SQLite不需要额外安装驱动Python自带sqlite3模块SQLAlchemy可以直接连接。如果练习的生产场景是PostgreSQL就需要psycopg2-binary或者新版psycopg3。psycopg3安装方式略有不同需要pip install psycopg[binary]然后连接串里写postgresqlpsycopg://user:passhost:port/dbname。很多新手在连接PostgreSQL时报错十有八九是用户名密码权限问题或者PostgreSQL默认端口5432被占用可以先在命令行用psql测试一下能否正常登录再回头查SQLAlchemy的连接串。3.2 设计一个爬虫数据表结构爬虫场景最典型的存储需求是抓取文章列表、去重、记录抓取时间、支持按来源和关键词检索。我直接演示一张articles表和一张crawl_records表的设计。from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship from sqlalchemy import String, Integer, DateTime, ForeignKey, Text, UniqueConstraint from datetime import datetime class Base(DeclarativeBase): pass class Article(Base): __tablename__ articles __table_args__ ( UniqueConstraint(source, url, nameuq_source_url), ) id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(200), nullableFalse) url: Mapped[str] mapped_column(String(500), nullableFalse) source: Mapped[str] mapped_column(String(50), nullableFalse) content: Mapped[str] mapped_column(Text, default) author: Mapped[str] mapped_column(String(50), default) published_at: Mapped[datetime | None] mapped_column(DateTime, nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now()) crawl_records: Mapped[list[CrawlRecord]] relationship(back_populatesarticle) class CrawlRecord(Base): __tablename__ crawl_records __table_args__ ( UniqueConstraint(article_id, crawl_time, nameuq_article_crawl_time), ) id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue) article_id: Mapped[int] mapped_column(ForeignKey(articles.id), nullableFalse) status: Mapped[str] mapped_column(String(20), defaultsuccess) crawl_time: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now()) duration_ms: Mapped[int] mapped_column(Integer, default0) article: Mapped[Article] relationship(back_populatescrawl_records)这里有两个设计细节值得说明用UniqueConstraint(source, url)给文章数据加了唯一约束这样即使爬虫脚本写得不小心重复抓取靠数据库层面也能挡住一部分重复数据。单独建一张crawl_records表记录每次爬取的耗时和状态好处是后续分析爬虫健康度、统计抓取成功率都有数据可查。这种主表记录表的结构在真实项目中非常常见比把所有字段塞进一张表优雅很多。3.3 连接数据库并写入数据选SQLite练习连接串最简单from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine create_engine(sqlite:///./crawler.db) SessionLocal sessionmaker(bindengine, autoflushFalse, expire_on_commitFalse) Base.metadata.create_all(engine)autoflushFalse是一个细节。默认情况下Session会在查询前自动把待处理的变更发送到数据库flush这有可能导致某些瞬间数据状态不是你预期的那样。很多团队习惯把autoflush关掉让flush只发生在显式调用session.flush()或session.commit()的时候行为更可控。写入数据的核心动作是批量插入def save_articles(session, items): items: list of dict for item in items: article Article( titleitem[title], urlitem[url], sourceitem[source], contentitem.get(content, ), authoritem.get(author, ), published_atitem.get(published_at), ) session.add(article) session.commit()如果数据量很大几个G的数据一次性add再commit会非常慢因为SQLAlchemy要跟踪每个对象的状态。更高效的做法是分批提交每500条或者每1000条提交一次def save_articles_batch(session, items, batch_size500): for i in range(0, len(items), batch_size): batch items[i:i batch_size] for item in batch: session.add(Article(**{k: v for k, v in item.items() if hasattr(Article, k)})) session.commit()这里用了hasattr(Article, k)做过滤防止传入的字典里有模型不认识的字段导致报错。批量插入还有一种方式是走Core层的insert()速度会更快因为绕过了ORM的实例化过程from sqlalchemy import insert stmt insert(Article).values([ {title: item[title], url: item[url], source: item[source]} for item in items ]) session.execute(stmt) session.commit()实际项目里我通常先用ORM方式写业务代码确定逻辑没问题后再把真正需要高性能的批量导入部分改成Core的insert()。这种先求正确、再求性能的思路能帮你避免一开始就为了优化性能而写出难以维护的代码。3.4 数据去重逻辑数据库唯一约束应用层判断爬虫最烦的问题就是重复抓取。我见过很多项目只靠先查数据库再决定是否插入来去重这在数据量小的时候还能凑合一旦并发量上来或者表数据过亿每次查询的代价就很高。最佳实践是唯一约束放在数据库层面兜底应用层只做轻量的查重缓存。from sqlalchemy.exc import IntegrityError seen_urls set() def save_articles_dedup(session, items): for item in items: if item[url] in seen_urls: continue article Article( titleitem[title], urlitem[url], sourceitem[source], ) session.add(article) seen_urls.add(item[url]) try: session.commit() except IntegrityError: session.rollback() # 说明有并发冲突或缓存失效可以再查一遍数据库 for item in items: exists session.execute( select(Article.id).where(Article.url item[url]) ).first() if not exists: session.add(Article( titleitem[title], urlitem[url], sourceitem[source], )) session.commit()这个写法看着繁琐但实际运行中能省掉大量无谓的查询。seen_urls集合在内存中处理相同URL的重复时几乎是零成本万一有漏网之鱼靠数据库的UniqueConstraint抛出的IntegrityError兜底回滚之后再做精确查重既保证数据不重复又不会因为并发冲突直接把整个爬虫程序搞崩。3.5 查询报表统计爬取趋势和热门来源数据存进去不是目的能查出来才有价值。用SQLAlchemy写两个报表查询先看每个来源的文章数量from sqlalchemy import select, func stmt ( select(Article.source, func.count(Article.id).label(cnt)) .group_by(Article.source) .order_by(func.count(Article.id).desc()) ) rows session.execute(stmt).all() for source, cnt in rows: print(source, cnt)再看最近一周每天抓取的文章量from datetime import datetime, timedelta week_ago datetime.now() - timedelta(days7) stmt ( select(func.date(Article.created_at).label(day), func.count(Article.id)) .where(Article.created_at week_ago) .group_by(func.date(Article.created_at)) .order_by(func.date(Article.created_at)) ) rows session.execute(stmt).all() for day, cnt in rows: print(day, cnt)func.date()是数据库函数在SQLite和PostgreSQL里都能正常工作。SQLAlchemy会在生成SQL时把它编译成对应数据库的日期函数这种跨数据库的能力就是我一直推荐它的原因。4. 常见问题与排查技巧实录4.1 N1查询ORM最容易踩的性能坑用ORM查列表时最常见的问题不是连接慢而是N1查询。看起来只发了一条查询实际上ORM在访问每个对象的关联属性时又各自发了一条SQL去数据库。举个例子stmt select(Article).limit(10) articles session.scalars(stmt).all() for article in articles: print(article.crawl_records[-1]) # 每访问一次都会发起一次查询这段代码会产生1条查询articles表的SQL加上10条查询crawl_records表的SQL总共11条。数据量小看不出问题数据量大了数据库会被拖垮。解决方案是使用selectinload或joinedload主动加载关联关系from sqlalchemy.orm import selectinload stmt select(Article).options(selectinload(Article.crawl_records)).limit(10) articles session.scalars(stmt).all()selectinload会先用一条IN查询把关联记录一次性取出来加载到内存里后续访问article.crawl_records就不会再触发SQL了。建议凡是能预见到要访问关联属性的场景查询时直接把关联关系用selectinload带上。宁可多查一条关联SQL也不要让ORM在循环里偷偷发N条查询。4.2 Session用完之后必须关闭或重置Web应用里最容易出现的内存泄漏问题之一就是把Session当成全局单例来用。Session内部持有大量已加载的对象引用如果长生命周期服务里一直复用同一个Session内存占用会只增不减。正确做法是依赖注入或中间件在每个请求开始时创建Session请求结束或异常时关闭。FastAPI里可以这么写from fastapi import Depends def get_db(): db SessionLocal() try: yield db finally: db.close()除了关闭还有一个很多人不知道的细节expire_on_commit。默认情况下session.commit()之后所有对象的属性都被标记为过期下次访问属性时会重新查询数据库。在sessionmaker里设置expire_on_commitFalse可以避免这种隐式查询尤其是在模板渲染场景里避免提交之后页面展示又要触发一次数据库查询的奇怪现象。4.3 模型字段名与数据库关键字的冲突给模型字段起名的时候如果取了order、group、desc这类在SQL里是关键字的名字生成SQL时很可能会报语法错误。SQLAlchemy对大多数情况下可以自动转义但在某些数据库方言下依然会出问题。建议一开始就给字段加上显式的列名映射from sqlalchemy import Column, String class Product(Base): __tablename__ products id: Mapped[int] mapped_column(Integer, primary_keyTrue) name: Mapped[str] mapped_column(String(100)) # 如果字段名叫 order则映射到 order_no 列 order_no: Mapped[str] mapped_column(order, String(50))这样Python里用product.order_no数据库里实际列名是order看起来绕了一圈但能避开关键字冲突。4.4 同步不是终点异步SQLAlchemy怎么选搜索热词里有比较多的sqlalchemy psycopg3 异步 同步比较可见不少人在纠结要不要上异步。结论可以直接说如果你的项目是FastAPI这类异步框架并且数据库交互是IO密集场景用异步SQLAlchemy可以提升并发吞吐量如果项目是传统Django、Flask同步接口或者主要写爬虫脚本就没必要硬上异步同步写法更简单调试成本也更低。SQLAlchemy 2.x的异步用法主要涉及两个地方from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker from sqlalchemy import text engine create_async_engine(postgresqlpsycopg://user:passlocalhost/db) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse) async def get_one(): async with AsyncSessionLocal() as session: result await session.execute(text(SELECT 1)) return result.scalar()注意异步的连接串要用postgresqlpsycopg或postgresqlasyncpg这样的异步方言同步的psycopg2-binary不支持异步。很多人换了异步写法却不换驱动结果死活连不上数据库报错信息还特别隐晦。4.5 调试技巧从日志到SQL编译最后分享一个实用技巧。排查SQLAlchemy性能问题的时候我通常会先开启SQL日志import logging logging.basicConfig() logging.getLogger(sqlalchemy.engine).setLevel(logging.INFO)这样控制台会打印出SQLAlchemy实际执行的每一条SQL语句包括耗时。如果你发现一个接口执行了十几条SQL基本就是N1问题没跑了。有时候想单独看某条语句编译后的SQL不用启动日志直接stmt select(Article).where(Article.title.like(%Python%)) print(stmt.compile(engineengine))这个方式特别适合调试动态查询确认自己拼出来的条件是不是符合预期。我个人在实际项目里遇到最多次的坑基本都是Session生命周期和N1问题。前者会让你的服务看起来偶尔卡顿后者会让你的数据库在数据量上来之后突然扛不住。如果你现在正在迁移到SQLAlchemy我建议先把这两个概念吃透再去研究各种高级查询写法。最后顺手推荐一个思路如果你只是写脚本、做数据分析用SQLite加同步SQLAlchemy就够了如果你的服务要扛线上流量至少把连接池、批量提交、异步驱动这几件事一起考虑进去。数据层是大多数应用的底座这一层稳了上面的业务逻辑才有得谈。
