SQLAlchemy性能反模式及其修复方法

ORM（对象关系映射）让应用程序开发人员的工作更加轻松。SQLAlchemy 提供了一个成熟的 ORM，它带来了模型关系、强大的查询构建范式、便捷的序列化等诸多优势。然而，正是由于这种易用性，人们很容易忽略其背后的工作原理。在使用 SQLAlchemy 时，一些看似微小的选择都可能对性能产生重大影响。

在本文中，我将探讨开发人员在使用 SQLAlchemy 时遇到的一些主要性能问题。

只需要计数却检索整个结果集

有时开发人员只需要统计结果的数量，但不会使用数据库计数，而是获取所有结果，然后使用 Python 中的 len 函数进行计数。

count = len(User.query.filter_by(acct_active=True).all())

改用 SQLAlchemy 的 count 方法会在服务器端执行计数，从而大大减少发送到客户端的数据量。此外，在前面的示例中调用 all() 方法还会导致模型对象的实例化，如果数据行数足够多，实例化操作的开销会迅速增加。

除非需要计数超过所需数量，否则请选择使用计数方法。

count = User.query.filter_by(acct_active=True).count()

仅需几列数据时检索整个模型

很多情况下，发出查询时只需要少数几列数据。SQLAlchemy 可以只获取我们感兴趣的列，而不是返回整个模型实例。这不仅减少了发送的数据量，还避免了实例化整个对象的需要。使用列数据元组而不是模型可以显著提高速度。

result = User.query.all()
for user in result:
    print(user.name, user.email)

建议使用 with_entities 方法仅选择所需内容。

result = User.query.with_entities(User.name, User.email).all()
for (username, email) in result:
    print(username, email)

在循环中一次更新一个对象

避免使用循环逐个更新集合。虽然数据库执行单个更新可能很快，但应用程序和数据库服务器之间的往返时间会迅速累积。一般来说，应尽可能减少查询次数。

for user in users_to_update:
  user.acct_active = True
  db.session.add(user)

请改用批量更新方法。

query = User.query.filter(user.id.in_([user.id for user in users_to_update]))
query.update({"acct_active": True}, synchronize_session=False)

触发级联删除

ORM 允许轻松配置模型之间的关系，但其中一些微妙的行为可能会令人意外。大多数数据库通过外键和各种级联选项来维护关系完整性。SQLAlchemy 允许您定义带有外键和级联选项的模型，但 ORM 有其自身的级联逻辑，这可能会抢占数据库的先机。

请考虑以下模型。

class Artist(Base):
    __tablename__ = "artist"

    id = Column(Integer, primary_key=True)
    songs = relationship("Song", cascade="all, delete")

class Song(Base):
    __tablename__ = "song"

    id = Column(Integer, primary_key=True)
    artist_id = Column(Integer, ForeignKey("artist.id", ondelete="CASCADE"))

删除艺术家信息会导致 ORM 对 Song 表发出删除查询，从而阻止因外键而导致的删除操作。在关系复杂且记录数量庞大的情况下，这种行为可能会成为性能瓶颈。

启用 `passive_deletes` 选项可确保数据库管理关联关系。但请务必确认您的数据库支持此功能。例如，SQLite 默认情况下不管理外键。

songs = relationship("Song", cascade="all, delete", passive_deletes=True)

应该使用预加载时却依赖延迟加载。

延迟加载是 SQLAlchemy 处理关系的默认方式。这意味着，以上一个例子为例，加载艺术家信息并不会同时加载其歌曲信息。这通常是个好主意，但如果某些关系始终需要加载，那么单独查询可能会造成资源浪费。

如果允许以延迟方式加载关系，像 Marshmallow 这样的流行序列化框架可能会触发一系列查询。

有几种方法可以控制这种行为。最简单的方法是通过关系函数本身。

songs = relationship("Song", lazy="joined", cascade="all, delete")

这将导致所有针对艺术家的查询都添加左连接，因此歌曲集将立即可用。虽然返回给客户端的数据量更多，但往返次数可能会大大减少。

SQLAlchemy 为无法采用这种一刀切方法的情况提供了更精细的控制。joinedload() 函数可用于针对每个查询切换连接加载。

from sqlalchemy.orm import joinedload

artists = Artist.query.options(joinedload(Artist.songs))
print(artists.songs) # Does not incur a roundtrip to load

使用 ORM 进行批量记录导入

当导入数千条记录时，构建完整模型实例的开销会成为主要的瓶颈。例如，想象一下，从一个文件中加载数千条歌曲记录，而每首歌曲都首先被转换成一个字典。

for song in songs:
    db.session.add(Song(**song))

相反，绕过 ORM，仅使用 SQLAlchemy 核心的参数绑定功能。

batch = []
insert_stmt = Song.__table__.insert()
for song in songs:
    if len(batch) > 1000:
       db.session.execute(insert_stmt, batch)
       batch.clear()
    batch.append(song)
if batch:
    db.session.execute(insert_stmt, batch)

请注意，此方法自然会跳过您可能依赖的任何客户端 ORM 逻辑，例如基于 Python 的列默认值。虽然此方法比将对象作为完整模型实例加载要快，但您的数据库可能具有更快的批量加载方法。例如，PostgreSQL 的 COPY 命令在加载大量记录方面可能具有最佳性能。

过早调用 commit 或 flush 函数

很多情况下，您需要将子记录与其父记录关联起来，反之亦然。一种显而易见的方法是刷新会话，以便为相关记录分配一个 ID。

artist = Artist(name="Bob Dylan")
song = Song(title="Mr. Tambourine Man")

db.session.add(artist)
db.session.flush()

song.artist_id = artist.id

通常情况下，每次请求多次提交或刷新数据是没有必要的，也不建议这样做。数据库刷新操作会强制数据库服务器执行磁盘写入，在大多数情况下，客户端会被阻塞，直到服务器确认数据已写入。

SQLAlchemy 可以在后台跟踪关系并管理键。

artist = Artist(name="Bob Dylan")
song = Song(title="Mr. Tambourine Man")

artist.songs.append(song)

总结

我希望这份常见陷阱清单能帮助您避免这些问题，并确保您的应用程序流畅运行。一如既往，在诊断性能问题时，测量是关键。大多数数据库都提供性能诊断工具，可以帮助您精确定位问题，例如 PostgreSQL 的 pg_stat_statements 模块。