企业官网建设流程全解析

直接操作数据库的时候，大家是不是都有过这样的经历：项目初期用原生SQL一把梭，增删改查写得飞起，可到了后面需求一复杂，SQL语句越拼越长，改一个字段要牵扯十来个查询，而且还得天天提防SQL注入。后来接触到了SQLAlchemy ORM，才意识到数据库操作还能这么舒服——直接用Python类操作数据表，查询语法接近原生表达，事务管理清清楚楚。这篇指南就把我从裸SQL转向SQLAlchemy ORM过程中沉淀下来的实战经验一次性讲透，大部分内容都是官方文档里不会细说、但在真实项目里绝对会踩到的点，适合刚开始接触ORM、或者已经用了但总觉得哪里别扭的同学参考。

1. 我为什么从裸SQL转向ORM：一次实际重构的启发

先说个我自己的项目经历。之前维护一个内部管理系统，数据库用的是SQLite起步、后来迁移到PostgreSQL，数据表大概20来张，业务逻辑集中在订单、库存和用户这三块。早期图省事，所有查询都直接用cursor.execute()拼SQL字符串，查出来的行用fetchall()变成元组列表交给上层处理。

表面上看没什么问题，但等表结构改了一轮之后痛点全出来了：

1.1 字段名散落各处，改表结构等于上刑

比如订单表里原来有个字段叫remark，后来业务调整要改成order_remark。听起来就是个改名，对吧？结果全项目里搜出来几十处SQL引用，有些是拼接在SELECT里的、有些是写在WHERE条件里的、还有些在INSERT语句中。逐一手工修改，漏掉任何一处，线上运行的时候就直接报undefined column错误。这种事情我没少干，而且每次改完还得安慰自己"这是正常维护成本"。

用SQLAlchemy以后，字段名变成了Python类属性，改字段名只要动模型定义一处，再配合迁移工具自动生成ALTER TABLE语句，运行时代码里几乎没有字符串形式的字段名，这个痛点直接从根源上消失了。

1.2 原生SQL的复用性太差，ORM让查询变成了组装

当时系统里有个统计报表需求——按时间区间、按销售渠道、按订单状态三个维度动态组合查询。用原生SQL实现的时候，我写了一个通用函数，里面用一大堆if判断拼条件、拼GROUP BY字段。写法大致是这样：

def build_report_query(start, end, channel=None, status=None): sql = "SELECT date, channel, COUNT(*) FROM orders WHERE created_at >= ? AND created_at <= ?" if channel: sql += " AND channel = ?" if status: sql += " AND status = ?" sql += " GROUP BY date, channel"

这段代码的逻辑没有错，但每加一个过滤维度，函数的复杂度就上升一截，测试用例也得多写好几个分支。换成ORM之后，同样的需求用表达式组装就清爽很多，而且每个条件的加入与否天然是Python层的逻辑判断，不需要在字符串上做文章。

1.3 可读性与可维护性的最终差异

说实话，ORM生成的SQL语句，在性能上和手写SQL基本没有差别——执行计划该走索引走索引，该顺序扫描顺序扫描。不一样的是写代码的人怎么理解这段逻辑。裸SQL的核心问题是数据库的行，在Python里是一堆无名的元组；ORM的核心问题是数据库的行，被完整映射成对象，你可以直接通过order.user.name这样的属性链去访问关联数据，心智负担小很多。

那次重构之后，我对ORM的态度从"花架子"变成了"生产级工具"。但同时也意识到，SQLAlchemy作为最成熟的Python ORM库，它的学习曲线其实是"先甜后苦"——前期建映射、写查询都挺顺利，一旦涉及事务边界、关联加载、并发控制这些深层机制，坑就一个接一个地冒出来。接下来的篇幅，我按实际项目里会遇到的顺序，把这套东西完整过一遍。

2. 映射声明的取舍：如何设计你的第一个建表类

用SQLAlchemy的第一步不是写查询，而是定义模型。这里的模型映射方式选择，直接决定了后面所有代码的写法风格。

2.1 经典映射与现代映射：不是非此即彼

SQLAlchemy有两大类映射风格：一种是早期的经典映射，显式定义一个Table对象，再定义一个普通Python类，通过mapper()函数把两者关联起来；另一种是现在最常见、也被官方推荐的方式——声明式映射，直接在类定义里用Column声明字段，类本身既承担表结构定义又承担数据对象角色。

我个人的建议是：新项目直接用声明式，经典映射只在你需要对某个已有类（比如ORM之外的普通数据类）做持久化时才需要考虑。

声明式写法看起来是这样的：

from sqlalchemy.orm import declarative_base, relationship from sqlalchemy import Column, Integer, String, DateTime, ForeignKey from datetime import datetime Base = declarative_base() class Order(Base): __tablename__ = "orders" id = Column(Integer, primary_key=True, autoincrement=True) order_no = Column(String(32), unique=True, nullable=False, index=True) channel = Column(String(16), nullable=False, default="web") status = Column(Integer, nullable=False, default=0) created_at = Column(DateTime, default=datetime.now) updated_at = Column(DateTime, onupdate=datetime.now) items = relationship("OrderItem", back_populates="order")

关键点在于declarative_base()生成的这个Base类，它是所有模型类的父类，也是SQLAlchemy维护"类与表映射关系"的注册中心。只要你把这个类导入到某个模块里，之后调用Base.metadata.create_all(engine)就能根据所有继承自它的模型自动建表。

2.2 主键策略：自增主键和UUID主键怎么选

建表类里第一个要拿主意的是主键。

• 自增整数主键：最省心，数据库自带的autoincrement机制负责分配，连表也方便。缺点是如果系统未来要做分库分表，跨库唯一性会出问题。
• UUID字符串主键：分布式的天然朋友，任何一个服务都能独立生成全局唯一ID，不需要访问数据库中心。缺点是主键值长，索引空间占用变大，排序性能不如整数。

我实际项目里通常是混合方案：核心业务表用自增整数主键，对外暴露的标识（比如订单号、用户唯一标识）用独立字符串字段加唯一索引。这样既保证了内部连接的效率，又不会把内部ID暴露给客户端，安全性和实用性都兼顾了。

2.3 字段类型踩坑记录：Integer存Boolean的小心机

初学ORM的时候，我犯过一个低级错误——把所有布尔字段定义成Boolean，结果后续在某个数据库迁移场景下出现了兼容性问题。后来团队统一的做法是：布尔字段存SmallInteger，0和1表示False和True。这样看起来不如Boolean优雅，但好处在于，无论你底层用的是SQLite、MySQL还是PostgreSQL，语义都不会变，而且后续如果要扩展多状态枚举也不会被布尔类型卡住。

日期时间字段也值得注意。DateTime映射到不同数据库后，SQLAlchemy会自动适配对应的时间类型。但有两点坑：

1. default=datetime.now是在Python层调用，也就是说数据插入前由Python进程生成时间；如果你想完全依赖数据库的CURRENT_TIMESTAMP，可以用server_default=func.now()。
2. onupdate=datetime.now只会在ORM层更新时触发，如果你绕过ORM直接执行UPDATE语句，这个字段不会自动更新。谨慎依赖这个属性。

2.4 命名策略与约束：让数据库DBA朋友少找你麻烦

很多人喜欢给每个字段加nullable=False来保证非空，但过度使用会导致业务逻辑失去灵活性。我的经验是：核心业务字段（金额、状态、外键）一定非空，可选信息字段（备注、扩展属性）允许为空，同时用Python层的default提供合理的默认值。

表名和字段名建议统一使用小写下划线风格，不要驼峰。大多数数据库对大小写敏感程度不同，驼峰命名在MySQL的Linux环境下会引发一些莫名其妙的"表不存在"问题，而小写下划线天然规避这类麻烦。

3. 会话与事务的边界：最容易被新手写崩的地方

模型定义好了，接下来就是往数据库里塞数据、改数据、查数据。但ORM里所有操作都离不开Session——一个大多数人只学会了用、没学会管的东西。

3.1 Session到底是个什么东西，为什么要小心它

Session是SQLAlchemy ORM中的一个"工作区"概念。简单理解，它维护着一个从数据库加载出来的对象集合，以及这些对象在内存中的状态变化。当你调用session.add(obj)时，对象其实还停留在内存里，只有执行session.commit()后，所有Pending状态的变化才会真正同步到数据库。

这里有个新手最容易踩的坑：Session对象不是线程安全的。在Web应用里如果用一个全局Session实例被多个请求线程共用，轻则脏读，重则直接抛异常。更常见的做法是：每次请求来了创建一个新的Session，请求结束了就关闭Session。

3.2 手动管理Session：scoped_session与with语句

最稳妥的写法是显式控制Session的生命周期：

from sqlalchemy.orm import sessionmaker SessionFactory = sessionmaker(bind=engine) def get_session(): return SessionFactory()

在业务代码里这样用：

session = get_session() try: order = session.query(Order).filter_by(order_no="SO2024001").first() order.status = 2 session.commit() except Exception: session.rollback() raise finally: session.close()

这个模板看起来啰嗦，但它是所有正确Session管理的基石。try/except/finally三段式确保了三件事：正常提交、异常回滚、最后关闭。漏掉任何一段，都可能让数据库连接池被占满，或者事务长时间挂起不释放。

3.3 commit与flush的区别：别等报了错才理解

session.flush()与session.commit()的区别是很多人学了很长时间才搞明白的。一句话概括：

• flush()：把当前内存中的对象状态同步到数据库，生成SQL执行，但事务并未提交。
• commit()：先执行flush，然后提交事务，让数据永久生效。

为什么需要单独调用flush()？一个典型场景是：你插入一条订单记录后，需要拿到订单的自增主键ID，用来作为子表的外键。此时还没commit，但可以调用flush()，让数据库生成ID，赋回到对象上。之后在同一个事务里继续操作子表数据，最后统一commit()。

注意：不调用flush()，直接访问order.id，在部分数据库驱动下可能拿到的是None，因为SQL还没真正执行，数据库还没分配ID。这个问题很隐蔽，排查起来也比较费时。

3.4 事务隔离级别的选择：RepeatableRead还是ReadCommitted

SQLAlchemy允许你在创建engine的时候指定隔离级别：

engine = create_engine(db_url, isolation_level="READ COMMITTED")

默认情况下，大多数数据库的隔离级别是REPEATABLE READ或READ COMMITTED。事务隔离级别直接影响并发行为：隔离级别越高，并发冲突越少，但锁竞争越大，吞吐量下降。

我遇到过一个线上事故：某报表统计模块用长事务一次处理几千条记录，同时前端订单创建也在高频写入。结果因为默认的REPEATABLE READ导致两个事务并发更新同一行时频繁锁等待，接口响应从80ms恶化到3秒。后来改成了READ COMMITTED，把长事务拆短，问题就解决了。

现实项目的建议：读多写少的系统用READ COMMITTED，需要强一致性的金融类场景用REPEATABLE READ，没有万能的默认值，你得根据业务去取舍。

4. 查询表达式：写复杂查询之前要搞懂的四件事

ORM的查询写法，看起来是链式调用的语法糖，但背后涉及惰性加载、SQL生成时机、表达式求值顺序等机制。搞懂下面四件事，写复杂查询才能不迷路。

4.1 query.filter和query.filter_by不是一回事

初学者最容易混淆的两个API：

• filter(Order.channel == "web")：条件表达式风格，能写出各种复杂条件，比如Order.created_at >= datetime(2024,1,1)。
• filter_by(channel="web")：简化关键字风格，只适合等值判断。

复杂查询统一用filter；简单查询filter_by写起来更短。两者可以混用链式调用：

orders = ( session.query(Order) .filter_by(channel="web") .filter(Order.created_at >= datetime(2024, 6, 1)) .all() )

4.2 惰性查询：生成SQL的时机比你想象的晚

session.query(Order)不会立即执行SQL，只有调用.all()、.first()、.one()或者迭代、.count()这些终止方法时，SQL才真正发送到数据库。

这个特性带来一个常见bug：你在查询语句后面拿到的Query对象，被当作数据集合到处传递，直到某个不相关的地方意外触发了SQL执行。为了tracking这类问题，SQLAlchemy提供了enable_eagerloads()和execution_options()这类高级选项，但对绝大多数场景，处理方式就是：明确区分"构建查询"和"执行查询"两个阶段，构建阶段不要做耗时操作，执行阶段尽量早执行，不要长时间持有Query对象。

4.3 聚合查询：group_by、having与func的正确姿势

报表类需求绕不开聚合。SQLAlchemy的方法签名还算直白，但要注意func的用法：

from sqlalchemy import func result = ( session.query( Order.channel, func.count(Order.id).label("order_count"), func.sum(Order.amount).label("total_amount"), ) .filter(Order.status == 1) .group_by(Order.channel) .having(func.count(Order.id) > 100) .all() )

需要注意的一点是：func.count(Order.id)不会统计NULL值，如果业务上想统计行数，哪怕该列是NULL，请用func.count()不传参数，或者用func.count(1)。

聚合查询结果返回的是Row对象元组形式，不是ORM对象，此时的Order.channel只是查询中的一个列表达式。这是新手容易困惑的点，需要记住：模型类和查询结果不一定是一回事。

4.4 原生SQL还是ORM：什么时候该混搭

说个反直觉的结论：ORM并不排斥原生SQL，SQLAlchemy提供了text()函数让你随时写原生SQL片段，并和ORM查询混合使用：

from sqlalchemy import text orders = ( session.query(Order) .filter(Order.status == 1) .filter(text("created_at >= '2024-06-01'")) .all() )

复杂分析型SQL（比如多层子查询嵌套、窗口函数计算）用原生SQL更直白，ORM负责外层对象组装；简单CRUD和过滤查询用ORM；千万别一律不用ORM，也别一律SQL化——取舍的关键是代码可读性和维护成本。

5. 关系映射：Table之间的连接不是拼字符串

关系映射是ORM最值钱的部分，它把外键连接变成Python属性访问。但使用不当，也是性能水平差异的分水岭。

5.1 relationship与外键：一对多和多对多的双向绑定

先说一对多。订单表与订单明细表就是典型的一对多关系。模型定义里，父表声明relationship("OrderItem", back_populates="order")，子表声明外键字段和反向属性：

class OrderItem(Base): __tablename__ = "order_items" id = Column(Integer, primary_key=True) order_id = Column(Integer, ForeignKey("orders.id"), nullable=False) item_name = Column(String(64)) price = Column(Integer) order = relationship("Order", back_populates="items")

加上back_populates双向关联后，你既能通过order.items拿到这个订单下的所有明细，也能通过item.order反查订单对象。

多对多关系复杂一些，需要中间表：

from sqlalchemy import Table, Column, ForeignKey order_tag = Table( "order_tag", Base.metadata, Column("order_id", ForeignKey("orders.id")), Column("tag_id", ForeignKey("tags.id")), )

然后在模型里用secondary参数声明：

tags = relationship("Tag", secondary=order_tag, back_populates="orders")

5.2 lazy加载策略：select、joined、subquery和dynamic选哪个

关系映射写好了，紧接着就要面对加载策略。这是ORM性能优化里最绕不开的话题，用错一个，数据库就会被大量重复查询拖垮。

最经典的坑是默认的select策略加上循环访问关系属性：

# 错误示范：循环内访问order.items，每条订单都会触发一次额外SELECT all_orders = session.query(Order).all() for order in all_orders: print(len(order.items))

如果orders有1000条，就额外产生1000次SELECT order_items WHERE order_id = ?，这就是大家常说的N+1问题。改进方式很简单：查询时显式指定加载策略。

from sqlalchemy.orm import joinedload all_orders = session.query(Order).options(joinedload(Order.items)).all()

这样一次查询就把关联明细全带上来了。

5.3 级联删除：别让数据库外键约束背锅

关系映射的级联行为由cascade参数控制。默认情况下，删除一个Order对象时，SQLAlchemy不会自动删除关联的OrderItem记录，但如果数据库层没有外键ON DELETE CASCADE约束，就会因外键约束违反而报错；如果设置了数据库层约束，ORM删除时反而可能再发一条UPDATE置空外键，行为更复杂。

我一般会在模型里明确声明：

items = relationship("OrderItem", back_populates="order", cascade="all, delete-orphan")

这样删除Order时，关联的OrderItem会自动删除。delete-orphan保证父对象与子对象断开关联时，子对象同时被删除，避免产生游离数据。

注意：cascade和passive_deletes的搭配需要仔细设计。如果你在数据库层设置了ON DELETE CASCADE，可以在关系映射里加passive_deletes=True，让数据库自己处理删除，而不是让ORM帮你逐条DELETE，性能能高不少。

6. N+1查询、锁与性能：真实项目里的优化记录

前面提到N+1问题，其实只是一个开始。真实的性能优化往往在你看不到的地方——ORM生成的SQL是否合理、锁策略是否碰撞、大批量操作是否过于低效。

6.1 从SQLAlchemy日志捕捉慢查询

排查性能问题，第一步是看到底生成了什么SQL。开发环境可以简单配置：

import logging logging.basicConfig() logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

这样一来，control台会打印出每次ORM操作对应的实际SQL语句和参数。我在调优复杂报表时，经常盯着这些日志逐条分析，发现很多情况下问题不是ORM封装的错，而是我写的业务代码在无关紧要的地方触发了大量冗余查询。

6.2 防止N+1的三种套路

• 套路一：joinedload强制JOIN一次性加载，适合一对多且集合规模适中的场景。

• 套路二：subqueryload在需要避免JOIN导致结果集膨胀时使用——一次性查出主记录后，把主记录ID集合传给第二个查询，再把结果按外键分组回填到各主记录对象。N+1变成了2次查询。

• 套路三：彻底关掉懒加载——在关系映射中设置lazy=False或在查询时用noload()。这只适合确定某个关系在当前场景下完全不需要，用了能让初始查询更快。

实际项目里没有银弹，我的经验是先确认业务数据规模，再选策略。如果一对多的集合平均只有三五条，joinedload很好；如果集合可能上千条，joinedload会让主查询的行数膨胀，反而拖垮性能，这种情况用selectinload或subqueryload更合适。

6.3 with_for_update：数据库行锁Use Case

并发场景下需要防止超卖、防止余额透支等问题时，光靠ORM的乐观锁不太够。SQLAlchemy提供了with_for_update()方法，直接给选中的行加数据库级锁：

account = ( session.query(Account) .filter(Account.id == user_id) .with_for_update() .first() ) account.balance -= amount session.commit()

with_for_update()对应的SQL是SELECT ... FOR UPDATE。在事务提交或回滚前，其他事务无法更新这一行，从而避免并发覆盖。注意锁是数据库行为，不是ORM行为，所以无论你是用ORM执行还是裸SQL执行，锁机制都是一样生效的。

还有一点：with_for_update()必须在事务内使用，不能在没有commit的事务里放任Session自生自灭，否则锁会一直攥着不放，最终拖垮数据库连接。

6.4 批量插入与批量更新：逐行commit是性能灾难

一次循环里插入几百条记录，每插入一条就commit一次，这种做法我在很多初版代码里见过。提交事务是有代价的，涉及磁盘刷新、日志写入等操作，逐行提交等于把固定开销重复了几百次。

正确姿势是用add_all()或bulk_insert_mappings()：

session.bulk_insert_mappings( Order, [ {"order_no": "SO1001", "channel": "web", "status": 0}, {"order_no": "SO1002", "channel": "app", "status": 0}, # 几百条 ], ) session.commit()

bulk_insert_mappings()直接跳过ORM的对象关系管理，性能提升明显，但代价是你插入的dict不会变成ORM对象，后续如果还要对这批数据做业务处理，就得重新查询。取舍在于：你是要极致的插入速度，还是要ORM对象化的便利。

7. 迁移、回滚与项目落地：ORM的最后一道坎

单机开发时代，模型定义好直接自动建表没有毛病。但项目一上线，生产环境的数据库表结构变更就需要一套可靠的迁移机制。SQLAlchemy的create_all()只能新建缺失的表，不能改已存在表的字段结构。于是辅助工具就成了必需品。

7.1 Alembic：模型和数据库之间的版本控制

Alembic是SQLAlchemy官方推荐的迁移工具，它的工作方式和Git类似——每次数据库结构变更生成一个迁移脚本，脚本可以被版本化管理，也可以被回滚。

基本流程：

alembic init alembic

初始化后，修改配置文件里的sqlalchemy.url为你的数据库地址，再编辑env.py把目标模型的Base.metadata传给迁移环境。接着生成第一个迁移脚本：

alembic revision --autogenerate -m "create orders table" alembic upgrade head

--autogenerate会对比当前模型的元数据和目标数据库的真实表结构，自动生成差异脚本。但它不是万能的：重命名列、修改约束这类改变无法自动识别，需要你手工在生成的迁移脚本里修改。所以不要把autogenerate的结果直接交付，一定要人工审查一遍再执行。

7.2 迁移脚本的审查与回滚实践

有一次我在某个项目里用autogenerate生成迁移脚本，准备给用户表新增一个唯一索引。生成结果看着没问题：

op.create_index("idx_user_email", "users", ["email"], unique=True)

但我没注意到生产环境里已经存在重复的email数据，导致迁移在upgrade阶段直接报唯一约束冲突。最后只能手工清理脏数据后重新执行升级。

经验教训：迁移上线前，一定要在预发布环境先跑一遍，并且准备好对应的downgrade脚本。好的迁移脚本必须能向下回滚，不能只写upgrade不管downgrade。因为上线过程中出了任何问题，最快的自救方式是alembic downgrade -1。

7.3 多环境配置与敏感信息管理

alembic.ini里直接写数据库连接串虽然方便，但存在泄露风险，而且多环境（开发、测试、生产）切换非常麻烦。我的做法是：把连接串从配置文件中剥离，改为读取环境变量，通过os.getenv("DATABASE_URL")动态组装。

在env.py里这样改：

from os import getenv config.set_main_option("sqlalchemy.url", getenv("DATABASE_URL", "sqlite:///local_dev.db"))

这样同一个项目在不同环境部署时，只需要设置不同的环境变量即可，数据库连接信息不会打进代码仓库，安全性也改善不少。

7.4 上线前最后检查清单

每次准备把数据库相关的改动发布到生产前，我习惯做一轮快速自查：

• 新加的模型字段是否定义了合理的默认值？生产环境已有历史数据不会为NULL？
• 迁移脚本是否执行过downgrade验证？
• 查询是否有潜在N+1？用日志或SQLAlchemy提供的统计工具确认过？
• 大批量操作是否考虑过分批处理，避免长事务锁库？
• 索引是否针对频繁过滤的WHERE条件和排序字段建立了？

这几条看着简单，但每一条背后都是一次线上故障的教训。数据库这块出问题往往不是瞬间崩掉，而是性能一点点劣化、慢慢拖垮整个应用，等发现的时候，排查面已经很大了。

8. 前面没提到的补充技巧：配置连接池与日志监控

前面已经把ORM的主干内容过完了，这里再补充三个大家在实际运维中绕不开的细节，能帮你避开不少后期麻烦。

8.1 连接池参数：pool_size与max_overflow怎么定

SQLAlchemy的Engine默认维护一个连接池，pool_size=5表示连接池维持5个连接，max_overflow=10表示最多额外创建10个连接，即总连接上限是15。

遇到高并发场景，默认参数很容易成为瓶颈。调参时不能光看活动请求数，还得结合数据库的最大连接数限制：

engine = create_engine( db_url, pool_size=20, max_overflow=20, pool_pre_ping=True, )

pool_pre_ping=True值得重点关注。它会在每次从连接池取连接前先发送一个轻量级检测（比如SELECT 1），防止因为数据库重启、网络闪断导致的"死连接"被继续使用。我经历过不止一次：MySQL因为参数调整重启，应用没有感知，重启后所有请求都卡在"获取数据库连接失败"上，直到手动重启应用才恢复。加了pool_pre_ping之后，这个问题基本绝迹。

8.2 慢查询日志与ORM的甜蜜配合

如果你部署的应用后面挂着MySQL或PostgreSQL，建议打开数据库层的慢查询日志。比如MySQL：

slow_query_log = ON long_query_time = 1

一秒钟以上的查询会被记录下来。然后结合前面提到的SQLAlchemy引擎日志，你就能分析出"这条慢SQL是ORM从哪段代码生成的"。印象最深的案例是：某个统计接口每次请求要跑5秒，数据库慢日志里抓出两条巨型SQL，追踪到业务代码后发现是两个列表接口各自独立查询，循环里反复调用，完全没有批次加载。优化后接口响应降到300ms——性能问题往往不是数据库不行，而是应用层产生了不必要的高频查询。

8.3 让ORM和大数据场景共存：流式处理与分页

如果你要处理几万甚至几十万条记录，直接.all()会把全部结果加载进内存，内存消耗不说，构建这些ORM对象本身就耗时。推荐分页处理或者流式处理：

# 分页处理示例 page_size = 1000 page = 0 while True: batch = ( session.query(Order) .filter(Order.status == 0) .offset(page * page_size) .limit(page_size) .all() ) if not batch: break process_batch(batch) page += 1

每次只处理一页数据，内存占用稳定。这里同样建议配合yield_per(page_size)方法控制ORM内部预取行为，避免一次抓取过多行。

补充提醒：使用offset分页在数据量大时会有性能问题，因为数据库要扫描并跳过offset行。此时如果你按自增ID排序，可以记录上一页最大ID作为下一页的起点，走WHERE id > last_id ORDER BY id LIMIT ?，性能通常好上一个数量级。

写在最后的一点个人体会

SQLAlchemy ORM真正常被低估的地方，不是它能省多少SQL，而是它逼着你去理顺数据模型之间的关系、事务的边界、加载策略的取舍。我见过一个项目，接手的人根本不理解Session要关闭，最后连接池被打爆；也见过一个团队，ORM用得熟练到可以轻松改写复杂查询，性能调优起来得心应手。差别在于有没有系统性地把底层机制搞明白。

如果你现在刚开始接触，我的建议是：先别追求把所有高级API都背下来，而是用最简单的一对多关系，把查询、关联加载、事务提交回滚完整走一遍，把SQLAlchemy生成的SQL打印出来读一读——理解了SQL，才能真正理解ORM。踩过几次坑、看过几次执行日志，这套工具的脾气你基本就摸清了。后续再遇到多对多、级联、锁、迁移这些进阶话题，参考这篇指南里提到的思路，应该能少走不少弯路。