SQLAlchemy 2.1.0b3:ORM 加载更快,正式版前值得提前试跑

2026-06-30 39 预计阅读时间: 1 分钟
来源: oschina.net AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:7 分钟

SQLAlchemy 2.1 系列的第三个测试版已经发布。这个版本带来了一组新功能、性能改进和错误修复,并预计会成为 2.1.0 正式版发布前的最后一个测试版。对多数后端团队来说,最值得关注的是 ORM 加载路径的性能变化:ORM 加载器改用基于位置的访问,获取 ORM 结果行时可以以普通元组形式处理,绕开 Row 接口带来的额外开销。

这次性能改进改在了哪里

SQLAlchemy ORM 查询结果的加载过程并不只是“数据库返回行,然后塞进对象”这么简单。ORM 需要读取每一列、匹配映射关系、构造或复用实体对象、处理 identity map,还要兼容各种结果行接口。

2.1.0b3 的发布亮点提到,ORM 加载器现在使用基于位置的访问,因此 ORM 结果行获取可以走普通元组,而不需要通过 Row 接口。这类优化通常不会改变你的查询写法,却会影响高频路径:

  • 大批量读取实体列表
  • API 列表页、后台导出任务
  • 带 eager loading 的复杂查询
  • 单请求内多次小查询叠加的业务接口

这不是一个“把慢 SQL 变快”的魔法。数据库执行计划、索引、网络往返仍然是主要成本。但当应用侧 ORM 装配成本已经明显时,这类内部路径优化会更有价值。

Beta 版本适合怎么评估

2.1.0b3 是测试版,不建议直接推到关键生产路径。更合理的做法是:拿一两个真实查询场景,在隔离环境里跑兼容性和性能对比。

可以重点检查三类代码:

  • 依赖 ORM 实体加载的接口,例如 session.scalars(select(User)).all()
  • 混合实体和列的查询,例如 select(User, Address.email)
  • 对 SQLAlchemy Row 行为有显式依赖的代码,例如访问 _mapping、按字段名取值、序列解包

发布摘要强调的是 ORM 加载器内部使用普通元组获取结果行,不代表所有 Core 查询都变成元组,也不代表应用代码应该依赖内部表示。应用层仍应使用公开 API:实体对象、scalars()mappings()execute() 等。

可以这样实践:用一段脚本试跑 ORM 加载路径

下面示例使用 SQLite 内存库,构造一批用户记录,然后对 ORM 实体加载做一个粗略计时。它不是严谨基准测试,但适合放进你的项目里改成真实模型和真实查询,用于比较当前版本与 2.1.0b3 的行为。

运行前可以把 SQLALCHEMY_VERSION 改成你要测试的版本;如果 beta 包尚未在你的镜像源可用,请换成官方 PyPI 或你的内部源。

python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install 'SQLAlchemy==2.1.0b3'
python orm_load_check.py

创建 orm_load_check.py

from __future__ import annotations

import time
from sqlalchemy import Integer, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False)
    email: Mapped[str] = mapped_column(String(100), nullable=False)


def seed(session: Session, total: int = 50_000) -> None:
    session.add_all(
        User(name=f"user-{i}", email=f"user-{i}@example.com")
        for i in range(total)
    )
    session.commit()


def time_query(session: Session, loops: int = 5) -> None:
    stmt = select(User).order_by(User.id)

    # Warm up: avoid counting first-use mapper and SQLite cache effects too heavily.
    session.scalars(stmt.limit(100)).all()

    for i in range(loops):
        start = time.perf_counter()
        users = session.scalars(stmt).all()
        elapsed = time.perf_counter() - start
        print(f"run={i + 1} rows={len(users)} elapsed={elapsed:.4f}s")


def main() -> None:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        seed(session)
        time_query(session)


if __name__ == "__main__":
    main()

如果要在项目中做更接近真实情况的验证,可以把脚本改成连接测试数据库,并替换成你的实际查询:

# 假设:你的项目已经有 engine、SessionLocal 和 ORM 模型。
from sqlalchemy import select
from your_app.db import SessionLocal
from your_app.models import Order

stmt = (
    select(Order)
    .where(Order.status == "paid")
    .order_by(Order.created_at.desc())
    .limit(10_000)
)

with SessionLocal() as session:
    orders = session.scalars(stmt).all()
    print(len(orders), orders[0].id if orders else None)

这里的重点不是追求单次数字好看,而是确认三件事:结果数量是否一致、对象字段是否正确、耗时是否在多次运行中稳定改善或至少没有退化。

升级前要盯住的边界

2.1.0b3 既然是 beta,就应该按“候选验证对象”处理,而不是按“默认升级对象”处理。建议团队做一个短清单:

  • 锁定依赖版本,避免 CI 和本地环境拿到不同 SQLAlchemy 版本
  • 跑完整 ORM 测试,尤其是关系加载、分页、导出、批处理任务
  • 检查是否有代码直接假设 Row 的内部结构
  • 对高流量查询做 A/B 基准,不只看平均值,也看 P95/P99
  • 关注数据库驱动差异,例如 PostgreSQL、MySQL、SQLite 在结果返回路径上的表现可能不同

对于库作者和框架维护者,2.1.0b3 很适合提前做兼容性扫描。对于业务系统,较稳妥的节奏是先在 CI 和预发环境试跑,再等 2.1.0 正式版发布后安排生产升级窗口。ORM 内部加载路径变快是一件好事,但真正的升级质量来自测试覆盖、真实查询样本和可回滚的发布流程。


相关推荐