SqlAlchemy 2.0 中文文档（五十四）

ApacheCN_飞龙

发布于 2024-08-01 11:31:09

1030

发布于 2024-08-01 11:31:09

文章被收录于专栏：信数据得永生

原文：docs.sqlalchemy.org/en/20/contents.html

ORM 配置

原文：docs.sqlalchemy.org/en/20/faq/ormconfiguration.html

如何映射没有主键的表？
如何配置一个与 Python 保留字或类似的列？
如何在给定映射类的情况下获取所有列、关系、映射属性等的列表？
我收到关于“在属性 Y 下隐式组合列 X”的警告或错误
我正在使用声明式并使用 and_() 或 or_() 设置 primaryjoin/secondaryjoin，但我收到了关于外键的错误消息。
为什么推荐在 LIMIT 中使用 ORDER BY（特别是在 subqueryload() 中）？

如何映射没有主键的表？

为了映射到特定表，SQLAlchemy ORM 需要至少有一个列被标记为主键列；当然，多列，即复合主键，也是完全可行的。这些列不需要实际被数据库知道为主键列，尽管最好是这样。只需要这些列行为象主键一样，例如，作为行的唯一且非空的标识符。

大多数 ORM 都要求对象有某种形式的主键定义，因为内存中的对象必须对应于数据库表中的唯一可识别行；至少，这允许对象可以被定位用于仅影响该对象行而不影响其他行的 UPDATE 和 DELETE 语句。然而，主键的重要性远不止于此。在 SQLAlchemy 中，所有 ORM 映射的对象始终使用称为身份映射的模式与它们的特定数据库行唯一链接在一起，这是 SQLAlchemy 使用的工作单元系统的核心模式，也是最常见的（和不那么常见的） ORM 使用模式的关键。

注意

需要注意的是，我们只讨论 SQLAlchemy ORM；一个基于 Core 构建并且只处理 Table 对象、select() 构造等的应用程序 不需要 在任何方式上存在或关联表上有任何主键（尽管再次强调，在 SQL 中，所有表应该真的有某种主键，以免您实际上需要更新或删除特定行）。

在几乎所有情况下，表确实有所谓的候选键，它是一列或一系列列，可以唯一标识一行。如果一张表真的没有这个，而且有实际完全重复的行，那么该表就不符合第一范式，也不能被映射。否则，组成最佳候选键的任何列都可以直接应用到映射器上：

class SomeClass(Base):
    __table__ = some_table_with_no_pk
    __mapper_args__ = {
        "primary_key": [some_table_with_no_pk.c.uid, some_table_with_no_pk.c.bar]
    }

当使用完全声明的表元数据时，最好在这些列上使用 primary_key=True 标志：

class SomeClass(Base):
    __tablename__ = "some_table_with_no_pk"

    uid = Column(Integer, primary_key=True)
    bar = Column(String, primary_key=True)

关系数据库中的所有表都应该有主键。即使是多对多关联表 - 主键将是两个关联列的组合：

CREATE  TABLE  my_association  (
  user_id  INTEGER  REFERENCES  user(id),
  account_id  INTEGER  REFERENCES  account(id),
  PRIMARY  KEY  (user_id,  account_id)
)

如何配置一个 Python 保留字或类似的 Column？

基于列的属性可以在映射中被赋予任何所需的名称。请参阅明确命名声明式映射的列。

如何在给定一个映射类的情况下获取所有列、关系、映射属性等列表？

所有这些信息都可以从 Mapper 对象中获得。

要获取特定映射类的 Mapper，请对其调用 inspect() 函数：

from sqlalchemy import inspect

mapper = inspect(MyClass)

从那里，关于类的所有信息都可以通过属性访问，例如：

Mapper.attrs - 所有映射属性的命名空间。属性本身是 MapperProperty 的实例，如果适用的话，它们包含可以导致映射的 SQL 表达式或列的其他属性。
Mapper.column_attrs - 限于列和 SQL 表达式属性的映射属性命名空间。您可能想直接使用 Mapper.columns 来获取 Column 对象。
Mapper.relationships - 所有RelationshipProperty属性的命名空间。
Mapper.all_orm_descriptors - 所有映射属性的命名空间，以及使用hybrid_property、AssociationProxy等系统定义的用户定义属性。
Mapper.columns - Column对象和与映射相关的其他命名 SQL 表达式的命名空间。
Mapper.mapped_table - 该映射器所映射到的Table或其他可选择的对象。
Mapper.local_table - 此映射器“本地”的Table；这在将映射器映射到组合可选择项的情况下与Mapper.mapped_table不同。

我收到关于“隐式将列 X 组合到属性 Y 下”的警告或错误

此条件指的是当映射包含两列，这两列由于名称而被映射到同一属性名下，但没有表明这是有意的。映射的类需要为每个要存储独立值的属性明确指定名称；当两列具有相同的名称并且没有消歧时，它们就属于同一属性，其效果是将一列的值复制到另一列，根据哪一列首先分配给属性。

这种行为通常是可取的，在继承映射中通过外键关系将两列链接在一起时是允许的，而不会发出警告。当出现警告或异常时，可以通过将列分配给名称不同的属性来解决问题，或者如果希望将它们组合在一起，则使用column_property()使其明确。

给出如下示例：

from sqlalchemy import Integer, Column, ForeignKey
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)

class B(A):
    __tablename__ = "b"

    id = Column(Integer, primary_key=True)
    a_id = Column(Integer, ForeignKey("a.id"))

自 SQLAlchemy 版本 0.9.5 起，将检测到上述条件，并将警告说A和B的id列正在组合为同名属性id，这是一个严重的问题，因为这意味着B对象的主键将始终与其A的主键相同。

解决此问题的映射如下：

class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)

class B(A):
    __tablename__ = "b"

    b_id = Column("id", Integer, primary_key=True)
    a_id = Column(Integer, ForeignKey("a.id"))

假设我们确实希望A.id和B.id互为镜像，尽管B.a_id是A.id相关的地方。我们可以使用column_property()将它们组合在一起：

class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)

class B(A):
    __tablename__ = "b"

    # probably not what you want, but this is a demonstration
    id = column_property(Column(Integer, primary_key=True), A.id)
    a_id = Column(Integer, ForeignKey("a.id"))

我正在使用声明式语法，并使用`and_()`或`or_()`设置`primaryjoin/secondaryjoin`，但是我收到了关于外键的错误消息。

你这样做了吗？：

class MyClass(Base):
    # ....

    foo = relationship(
        "Dest", primaryjoin=and_("MyClass.id==Dest.foo_id", "MyClass.foo==Dest.bar")
    )

那是两个字符串表达式的and_()，而 SQLAlchemy 不能对其应用任何映射。声明式允许将relationship()参数指定为字符串，这些字符串将使用eval()转换为表达式对象。但这不会发生在and_()表达式内部 - 这是声明式仅对作为字符串传递给primaryjoin或其他参数的整体应用的特殊操作：

class MyClass(Base):
    # ....

    foo = relationship(
        "Dest", primaryjoin="and_(MyClass.id==Dest.foo_id, MyClass.foo==Dest.bar)"
    )

或者，如果您需要的对象已经可用，请跳过字符串：

class MyClass(Base):
    # ....

    foo = relationship(
        Dest, primaryjoin=and_(MyClass.id == Dest.foo_id, MyClass.foo == Dest.bar)
    )

相同的想法也适用于所有其他参数，比如foreign_keys：

# wrong !
foo = relationship(Dest, foreign_keys=["Dest.foo_id", "Dest.bar_id"])

# correct !
foo = relationship(Dest, foreign_keys="[Dest.foo_id, Dest.bar_id]")

# also correct !
foo = relationship(Dest, foreign_keys=[Dest.foo_id, Dest.bar_id])

# if you're using columns from the class that you're inside of, just use the column objects !
class MyClass(Base):
    foo_id = Column(...)
    bar_id = Column(...)
    # ...

    foo = relationship(Dest, foreign_keys=[foo_id, bar_id])

为什么推荐使用`ORDER BY`与`LIMIT`（特别是与`subqueryload()`一起）？

当 SELECT 语句返回行时未使用 ORDER BY 时，关系数据库可以以任意顺序返回匹配的行。虽然这种排序很常见，对应于表中行的自然顺序，但并不是所有数据库和所有查询都是如此。这样做的结果是，任何使用LIMIT或OFFSET限制行，或者仅选择结果的第一行，而放弃其余部分的查询，在返回结果行时不是确定性的，假设有多个行匹配查询的条件。

尽管我们可能不会注意到这一点，因为对于通常以其自然顺序返回行的数据库上的简单查询，它更多地成为问题，如果我们还使用subqueryload()来加载相关集合，并且我们可能无法按预期加载集合。

SQLAlchemy 通过发出单独的查询来实现subqueryload()，其结果与第一个查询的结果匹配。我们会看到像这样发出两个查询：

>>> session.scalars(select(User).options(subqueryload(User.addresses))).all()
-- the "main" query
SELECT  users.id  AS  users_id
FROM  users
-- the "load" query issued by subqueryload
SELECT  addresses.id  AS  addresses_id,
  addresses.user_id  AS  addresses_user_id,
  anon_1.users_id  AS  anon_1_users_id
FROM  (SELECT  users.id  AS  users_id  FROM  users)  AS  anon_1
JOIN  addresses  ON  anon_1.users_id  =  addresses.user_id
ORDER  BY  anon_1.users_id

第二个查询将第一个查询嵌入为行的来源。当内部查询使用OFFSET和/或LIMIT而没有排序时，这两个查询可能不会看到相同的结果：

>>> user = session.scalars(
...     select(User).options(subqueryload(User.addresses)).limit(1)
... ).first()
-- the "main" query
SELECT  users.id  AS  users_id
FROM  users
  LIMIT  1
-- the "load" query issued by subqueryload
SELECT  addresses.id  AS  addresses_id,
  addresses.user_id  AS  addresses_user_id,
  anon_1.users_id  AS  anon_1_users_id
FROM  (SELECT  users.id  AS  users_id  FROM  users  LIMIT  1)  AS  anon_1
JOIN  addresses  ON  anon_1.users_id  =  addresses.user_id
ORDER  BY  anon_1.users_id

根据数据库的具体情况，我们可能会得到如下两个查询的结果：

-- query #1
+--------+
|users_id|
+--------+
|       1|
+--------+

-- query #2
+------------+-----------------+---------------+
|addresses_id|addresses_user_id|anon_1_users_id|
+------------+-----------------+---------------+
|           3|                2|              2|
+------------+-----------------+---------------+
|           4|                2|              2|
+------------+-----------------+---------------+

在上面的例子中，我们为user.id为 2 的用户接收到了两行addresses，而对于 id 为 1 的用户却没有。我们浪费了两行，并且未能实际加载集合。这是一个隐匿的错误，因为不查看 SQL 和结果，ORM 将不会显示任何问题；如果我们访问已有的User的addresses，它会对集合进行惰性加载，我们将看不到任何实际错误发生。

解决此问题的方法是始终指定确定性排序顺序，以便主查询始终返回相同的行集。这通常意味着您应该在表的唯一列上进行 Select.order_by() 排序。主键是一个不错的选择：

session.scalars(
    select(User).options(subqueryload(User.addresses)).order_by(User.id).limit(1)
).first()

注意，joinedload() 这种预加载策略不会遇到相同的问题，因为只会发出一个查询，所以加载查询不会与主查询不同。同样，selectinload() 这种预加载策略也不会有此问题，因为它将其集合加载直接链接到刚刚加载的主键值。

另请参阅

子查询预加载 ## 我如何映射一个没有主键的表？

SQLAlchemy ORM 为了映射到特定表，需要至少有一个列被指定为主键列；多列，即复合主键，当然也是完全可行的。这些列不需要实际上被数据库知道为主键列，尽管它们是主键列是个好主意。只需要这些列表现出主键的行为即可，例如作为行的唯一标识符和不可为空的标识符。

大多数 ORM 要求对象定义某种主键，因为内存中的对象必须对应于数据库表中的唯一可识别行；至少，这允许对象可以成为 UPDATE 和 DELETE 语句的目标，这些语句将仅影响该对象的行，而不会影响其他行。但是，主键的重要性远不止于此。在 SQLAlchemy 中，所有 ORM 映射的对象始终通过称为标识映射的模式与其特定数据库行唯一链接到一个 Session 中，该模式是 SQLAlchemy 使用的工作单元系统的核心，并且也是最常见（以及不那么常见）的 ORM 使用模式的关键。

注意

需要注意的是，我们只谈论 SQLAlchemy ORM；一个建立在 Core 之上、仅处理Table对象、select()构造等的应用程序，不需要在任何方式上要求主键存在于或与表相关联（尽管在 SQL 中，所有表实际上都应该具有某种主键，否则你可能需要实际更新或删除特定行）。

几乎在所有情况下，表都具有所谓的候选键，这是一列或一系列列，唯一标识一行。如果表确实没有这个，且具有实际完全重复的行，则该表不符合第一范式，无法进行映射。否则，组成最佳候选键的任何列都可以直接应用于映射器：

class SomeClass(Base):
    __table__ = some_table_with_no_pk
    __mapper_args__ = {
        "primary_key": [some_table_with_no_pk.c.uid, some_table_with_no_pk.c.bar]
    }

当使用完全声明的表元数据时，最好在这些列上使用primary_key=True标志：

class SomeClass(Base):
    __tablename__ = "some_table_with_no_pk"

    uid = Column(Integer, primary_key=True)
    bar = Column(String, primary_key=True)

关系数据库中的所有表都应该有主键。即使是多对多的关联表 - 主键也将是两个关联列的组合：

CREATE  TABLE  my_association  (
  user_id  INTEGER  REFERENCES  user(id),
  account_id  INTEGER  REFERENCES  account(id),
  PRIMARY  KEY  (user_id,  account_id)
)

如何配置一个是 Python 保留字或类似的列？

在映射中，基于列的属性可以赋予任何所需的名称。参见显式命名声明式映射的列。

如何获取给定映射类的所有列、关系、映射属性等列表？

所有这些信息都可以从Mapper对象中获取。

要获取特定映射类的Mapper，请在其上调用inspect()函数：

from sqlalchemy import inspect

mapper = inspect(MyClass)

从那里，可以通过诸如以下属性之类的属性访问有关类的所有信息：

Mapper.attrs - 所有映射属性的命名空间。这些属性本身是MapperProperty的实例，其中包含了可导致映射的 SQL 表达式或列的其他属性（如果适用）。
Mapper.column_attrs - 仅限于列和 SQL 表达式属性的映射属性命名空间。你可能想使用Mapper.columns直接获取 Column对象。
Mapper.relationships - 所有 RelationshipProperty 属性的命名空间。
Mapper.all_orm_descriptors - 所有映射属性的命名空间，以及使用诸如 hybrid_property、AssociationProxy 等系统定义的用户定义属性等。
Mapper.columns - 与映射相关联的 Column 对象和其他命名 SQL 表达式的命名空间。
Mapper.mapped_table - 此映射器映射到的 Table 或其他可选择的对象。
Mapper.local_table - 此映射器“本地”的 Table；在映射器使用继承映射到组合选择时，这与 Mapper.mapped_table 不同。

我收到了一个关于“隐式组合列 X 在属性 Y 下”的警告或错误

这种情况指的是映射包含两个列，这两个列由于它们的名称而被映射到同一属性名称下，但没有迹象表明这是有意的。映射类需要为每个要存储独立值的属性指定明确的名称；当两个列具有相同的名称并且没有消歧义时，它们就会落入同一个属性下，效果是从一个列中的值被复制到另一个列中，取决于哪个列首先分配给属性。

这种行为通常是可取的，在继承映射内部通过外键关系链接两个列时，无需警告即可允许。当出现警告或异常时，可以通过将列分配给不同命名的属性来解决问题，或者如果希望将它们组合在一起，则可以使用column_property()来明确表示这一点。

给出如下示例：

from sqlalchemy import Integer, Column, ForeignKey
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)

class B(A):
    __tablename__ = "b"

    id = Column(Integer, primary_key=True)
    a_id = Column(Integer, ForeignKey("a.id"))

截至 SQLAlchemy 版本 0.9.5，检测到上述条件，并将警告A和B的id列正在合并到同名属性id下，上面是一个严重问题，因为这意味着B对象的主键将始终反映其A的主键。

解决此问题的映射如下：

class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)

class B(A):
    __tablename__ = "b"

    b_id = Column("id", Integer, primary_key=True)
    a_id = Column(Integer, ForeignKey("a.id"))

假设我们确实希望A.id和B.id彼此镜像，尽管B.a_id是A.id相关的地方。我们可以使用column_property()将它们合并在一起：

class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)

class B(A):
    __tablename__ = "b"

    # probably not what you want, but this is a demonstration
    id = column_property(Column(Integer, primary_key=True), A.id)
    a_id = Column(Integer, ForeignKey("a.id"))

我正在使用声明式并使用`and_()`或`or_()`设置 primaryjoin/secondaryjoin，并且收到有关外键的错误消息。

您是这样做的吗？:

class MyClass(Base):
    # ....

    foo = relationship(
        "Dest", primaryjoin=and_("MyClass.id==Dest.foo_id", "MyClass.foo==Dest.bar")
    )

这是两个字符串表达式的and_()，SQLAlchemy 无法对其应用任何映射。声明式允许将relationship()参数指定为字符串，并使用eval()将其转换为表达式对象。但这不会发生在and_()表达式内部 - 它是声明式仅适用于作为字符串传递给 primaryjoin 或其他参数的整体的特殊操作：

class MyClass(Base):
    # ....

    foo = relationship(
        "Dest", primaryjoin="and_(MyClass.id==Dest.foo_id, MyClass.foo==Dest.bar)"
    )

或者如果您需要的对象已经可用，请跳过字符串：

class MyClass(Base):
    # ....

    foo = relationship(
        Dest, primaryjoin=and_(MyClass.id == Dest.foo_id, MyClass.foo == Dest.bar)
    )

相同的想法适用于所有其他参数，例如foreign_keys：

# wrong !
foo = relationship(Dest, foreign_keys=["Dest.foo_id", "Dest.bar_id"])

# correct !
foo = relationship(Dest, foreign_keys="[Dest.foo_id, Dest.bar_id]")

# also correct !
foo = relationship(Dest, foreign_keys=[Dest.foo_id, Dest.bar_id])

# if you're using columns from the class that you're inside of, just use the column objects !
class MyClass(Base):
    foo_id = Column(...)
    bar_id = Column(...)
    # ...

    foo = relationship(Dest, foreign_keys=[foo_id, bar_id])

为什么推荐在`LIMIT`中使用`ORDER BY`（特别是在`subqueryload()`中）？

当没有为返回行的 SELECT 语句使用 ORDER BY 时，关系数据库可以以任意的顺序返回匹配的行。虽然这种排序往往对应于表内行的自然顺序，但并非所有数据库和所有查询都是如此。这样做的结果是，任何使用LIMIT或OFFSET限制行数的查询，或者仅选择结果的第一行，丢弃其余行的查询，在返回哪个结果行时不是确定性的，假设查询的条件有多个匹配行。

尽管我们可能在通常按照它们的自然顺序返回行的数据库上的简单查询中没有注意到这一点，但如果我们还使用subqueryload()来加载相关集合，这就更成为一个问题，我们可能不会按预期加载集合。

SQLAlchemy 通过发出单独的查询来实现subqueryload()，其结果与第一个查询的结果匹配。我们看到像这样发出的两个查询：

>>> session.scalars(select(User).options(subqueryload(User.addresses))).all()
-- the "main" query
SELECT  users.id  AS  users_id
FROM  users
-- the "load" query issued by subqueryload
SELECT  addresses.id  AS  addresses_id,
  addresses.user_id  AS  addresses_user_id,
  anon_1.users_id  AS  anon_1_users_id
FROM  (SELECT  users.id  AS  users_id  FROM  users)  AS  anon_1
JOIN  addresses  ON  anon_1.users_id  =  addresses.user_id
ORDER  BY  anon_1.users_id

第二个查询将第一个查询嵌入为行的源。当内部查询使用OFFSET和/或LIMIT而没有排序时，这两个查询可能不会看到相同的结果：

>>> user = session.scalars(
...     select(User).options(subqueryload(User.addresses)).limit(1)
... ).first()
-- the "main" query
SELECT  users.id  AS  users_id
FROM  users
  LIMIT  1
-- the "load" query issued by subqueryload
SELECT  addresses.id  AS  addresses_id,
  addresses.user_id  AS  addresses_user_id,
  anon_1.users_id  AS  anon_1_users_id
FROM  (SELECT  users.id  AS  users_id  FROM  users  LIMIT  1)  AS  anon_1
JOIN  addresses  ON  anon_1.users_id  =  addresses.user_id
ORDER  BY  anon_1.users_id

根据数据库的具体情况，我们可能会得到以下两个查询的结果：

-- query #1
+--------+
|users_id|
+--------+
|       1|
+--------+

-- query #2
+------------+-----------------+---------------+
|addresses_id|addresses_user_id|anon_1_users_id|
+------------+-----------------+---------------+
|           3|                2|              2|
+------------+-----------------+---------------+
|           4|                2|              2|
+------------+-----------------+---------------+

如上所述，我们对于user.id为 2 的两个addresses行，却没有对于 1 的。我们浪费了两行，并且未能实际加载集合。这是一个潜在的错误，因为如果不查看 SQL 和结果，ORM 将不会显示任何问题；如果我们访问我们拥有的User的addresses，它将对集合进行惰性加载，并且我们将看不到任何实际出错的情况。

解决这个问题的方法是始终指定确定性的排序顺序，以便主查询始终返回相同的行集合。这通常意味着你应该在表上的一个唯一列上使用Select.order_by()。主键是一个不错的选择：

session.scalars(
    select(User).options(subqueryload(User.addresses)).order_by(User.id).limit(1)
).first()

请注意，joinedload() 急加载策略不会遭受相同的问题，因为只发出一次查询，因此加载查询不能与主查询不同。类似地，selectinload() 急加载策略也不会有此问题，因为它将其集合加载直接链接到刚刚加载的主键值。

另请参阅

子查询的急加载

性能

原文：docs.sqlalchemy.org/en/20/faq/performance.html

为什么升级到 1.4 和/或 2.x 后我的应用程序变慢了？
- 第一步 - 打开 SQL 日志记录并确认缓存是否正常工作
- 第二步 - 确定哪些构造阻止启用缓存
- 第三步 - 为给定的对象启用缓存和/或寻找替代方案
如何对基于 SQLAlchemy 的应用程序进行性能分析？
- 查询性能分析
- 代码性能分析
- 执行速度慢
- 结果获取慢 - 核心
- 结果获取慢 - ORM
我正在使用 ORM 插入 400,000 行，速度非常慢！

为什么升级到 1.4 和/或 2.x 后我的应用程序变慢了？

SQLAlchemy 自 1.4 版本起包含一个 SQL 编译缓存功能，它将允许核心和 ORM SQL 构造缓存它们的字符串形式，以及用于从语句中获取结果的其他结构信息，当下次使用另一个结构上等同的构造时，可以跳过相对昂贵的字符串编译过程。这个系统依赖于为所有 SQL 构造实现的功能，包括诸如 Column、select() 和 TypeEngine 等对象，以生成完全代表它们状态的缓存键，至于它们对 SQL 编译过程产生的影响程度。

缓存系统使得 SQLAlchemy 1.4 及以上版本在将 SQL 构造重复转换为字符串所花费的时间方面比 SQLAlchemy 1.3 更高效。然而，这仅在为使用的方言和 SQL 构造启用缓存时才有效；如果没有启用缓存，则字符串编译通常类似于 SQLAlchemy 1.3，但在某些情况下速度略有下降。

然而，有一种情况是，如果 SQLAlchemy 的新缓存系统已被禁用（出于以下原因），则 ORM 的性能实际上可能明显低于 1.3 或其他之前的版本，这是由于 ORM 惰性加载器和对象刷新查询中缺乏缓存，而在 1.3 版本和更早版本中使用了现在已经过时的 BakedQuery 系统。如果应用程序在切换到 1.4 时看到了显著（30% 或更高）的性能下降（以操作完成所需的时间为度量），这可能是问题的根本原因，下面有缓解措施。

参见

SQL 编译缓存 - 缓存系统概述

对象将不生成缓存键，性能影响 - 关于不启用缓存的元素生成的警告的额外信息。

第一步 - 打开 SQL 记录并确认缓存是否起作用

在这里，我们想要使用引擎日志记录中描述的技术，查找带有[no key]指示器的语句，甚至是带有[dialect does not support caching]的语句。对于首次调用语句时成功参与缓存系统的 SQL 语句，我们会看到[generated in Xs]指示器，随后对于绝大多数后续语句会看到[cached since Xs ago]指示器。如果对于特定的 SELECT 语句主要存在[no key]，或者如果由于[dialect does not support caching]完全禁用了缓存，这可能是导致性能严重下降的原因。

参见

使用日志估算缓存性能

第二步 - 确定哪些构造物阻止了缓存的启用

假设语句没有被缓存，在应用程序的日志中会及早发出警告（仅适用于 SQLAlchemy 1.4.28 及以上版本），指示不参与缓存的方言、TypeEngine 对象和 SQL 构造物。

对于诸如扩展TypeDecorator和UserDefinedType的用户定义数据类型，警告将如下所示：

sqlalchemy.ext.SAWarning: MyType will not produce a cache key because the
``cache_ok`` attribute is not set to True. This can have significant
performance implications including some performance degradations in
comparison to prior SQLAlchemy versions. Set this attribute to True if this
type object's state is safe to use in a cache key, or False to disable this
warning.

对于自定义和第三方 SQL 元素，例如那些使用自定义 SQL 构造和编译扩展中描述的技术构建的元素，这些警告会看起来像：

sqlalchemy.exc.SAWarning: Class MyClass will not make use of SQL
compilation caching as it does not set the 'inherit_cache' attribute to
``True``. This can have significant performance implications including some
performance degradations in comparison to prior SQLAlchemy versions. Set
this attribute to True if this object can make use of the cache key
generated by the superclass. Alternatively, this attribute may be set to
False which will disable this warning.

对于使用Dialect类层次结构的自定义和第三方方言，警告将如下所示：

sqlalchemy.exc.SAWarning: Dialect database:driver will not make use of SQL
compilation caching as it does not set the 'supports_statement_cache'
attribute to ``True``. This can have significant performance implications
including some performance degradations in comparison to prior SQLAlchemy
versions. Dialect maintainers should seek to set this attribute to True
after appropriate development and testing for SQLAlchemy 1.4 caching
support. Alternatively, this attribute may be set to False which will
disable this warning.

第三步 - 为给定对象启用缓存和/或寻求替代方案

缓存不足的缓解步骤包括：

将 ExternalType.cache_ok 设置为 True，用于所有继承自 TypeDecorator、UserDefinedType 的自定义类型，以及这些类型的子类，如 PickleType。仅当自定义类型不包含任何影响其渲染 SQL 的额外状态属性时才设置这个属性：

class MyCustomType(TypeDecorator):
    cache_ok = True
    impl = String

如果使用的类型来自第三方库，请与该库的维护者联系，以便进行调整和发布。

另请参阅

ExternalType.cache_ok - 启用自定义数据类型缓存的要求背景。

确保第三方方言将 Dialect.supports_statement_cache 设置为 True。这表明第三方方言的维护者确保其方言与 SQLAlchemy 1.4 或更高版本兼容，并且其方言不包含可能干扰缓存的编译特性。由于存在一些常见的编译模式可能会干扰缓存，因此方言维护者务必仔细检查和测试，调整任何与缓存不兼容的传统模式。

另请参阅

第三方方言的缓存 - 第三方方言参与 SQL 语句缓存的背景和示例。

自定义 SQL 类，包括使用自定义 SQL 构造和编译扩展创建的所有 DQL / DML 构造，以及对象的临时子类，如 Column 或 Table。对于简单子类，可以将 HasCacheKey.inherit_cache 属性设置为 True，该属性不包含影响 SQL 编译的子类特定状态信息。

另请参阅

为自定义构造启用缓存支持 - 应用 HasCacheKey.inherit_cache 属性的指南。

另请参阅

SQL 编译缓存 - 缓存系统概述

对象不会生成缓存键，性能影响 - 背景是在为特定结构和/或方言未启用缓存时发出警告的情况。## 如何分析一个使用 SQLAlchemy 的应用程序？

寻找性能问题通常涉及两种策略。一种是查询性能分析，另一种是代码性能分析。

查询性能分析

有时仅仅记录普通的 SQL（通过 python 的 logging 模块启用，或者通过create_engine()上的echo=True参数启用）就能让你了解到事情花费了多长时间。例如，如果在 SQL 操作之后记录一些内容，你会在日志中看到类似这样的内容：

17:37:48,325 INFO  [sqlalchemy.engine.base.Engine.0x...048c] SELECT ...
17:37:48,326 INFO  [sqlalchemy.engine.base.Engine.0x...048c] {<params>}
17:37:48,660 DEBUG [myapp.somemessage]

如果你在操作之后记录了myapp.somemessage，你就知道完成 SQL 部分花费了 334ms。

记录 SQL 还会说明是否发出了数十/数百个查询，这些查询可以更好地组织成更少的查询。当使用 SQLAlchemy ORM 时，“eager loading”特性提供了部分（contains_eager()）或完全（joinedload()，subqueryload()）自动化此活动，但是没有 ORM 的“eager loading”通常意味着使用连接，以便结果可以在一个结果集中加载而不是随着更多深度的添加而增加查询的数量（即 r + r*r2 + r*r2*r3 …）

对于更长期的查询性能分析，或者实现应用程序端的“慢查询”监视器，可以使用事件来拦截游标执行，使用以下类似的配方：

from sqlalchemy import event
from sqlalchemy.engine import Engine
import time
import logging

logging.basicConfig()
logger = logging.getLogger("myapp.sqltime")
logger.setLevel(logging.DEBUG)

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("query_start_time", []).append(time.time())
    logger.debug("Start Query: %s", statement)

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info["query_start_time"].pop(-1)
    logger.debug("Query Complete!")
    logger.debug("Total Time: %f", total)

在上述例子中，我们使用ConnectionEvents.before_cursor_execute()和ConnectionEvents.after_cursor_execute()事件来在执行语句时建立拦截点。我们使用info字典在连接上附加一个计时器；我们在这里使用堆栈，因为偶尔会出现游标执行事件嵌套的情况。

代码性能分析

如果日志显示单个查询花费的时间太长，你需要分解在数据库内处理查询、通过网络发送结果、被 DBAPI 处理以及最终由 SQLAlchemy 的结果集和/或 ORM 层接收的时间。每个阶段都可能存在自己的瓶颈，具体取决于具体情况。

为此，您需要使用Python Profiling Module。以下是一个将分析集成到上下文管理器中的简单示例：

import cProfile
import io
import pstats
import contextlib

@contextlib.contextmanager
def profiled():
    pr = cProfile.Profile()
    pr.enable()
    yield
    pr.disable()
    s = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
    ps.print_stats()
    # uncomment this to see who's calling what
    # ps.print_callers()
    print(s.getvalue())

要对代码段进行分析：

with profiled():
    session.scalars(select(FooClass).where(FooClass.somevalue == 8)).all()

分析的输出可以用来了解时间花在哪里。分析输出的一部分如下所示：

13726 function calls (13042 primitive calls) in 0.014 seconds

Ordered by: cumulative time

ncalls  tottime  percall  cumtime  percall filename:lineno(function)
222/21    0.001    0.000    0.011    0.001 lib/sqlalchemy/orm/loading.py:26(instances)
220/20    0.002    0.000    0.010    0.001 lib/sqlalchemy/orm/loading.py:327(_instance)
220/20    0.000    0.000    0.010    0.000 lib/sqlalchemy/orm/loading.py:284(populate_state)
   20    0.000    0.000    0.010    0.000 lib/sqlalchemy/orm/strategies.py:987(load_collection_from_subq)
   20    0.000    0.000    0.009    0.000 lib/sqlalchemy/orm/strategies.py:935(get)
    1    0.000    0.000    0.009    0.009 lib/sqlalchemy/orm/strategies.py:940(_load)
   21    0.000    0.000    0.008    0.000 lib/sqlalchemy/orm/strategies.py:942(<genexpr>)
    2    0.000    0.000    0.004    0.002 lib/sqlalchemy/orm/query.py:2400(__iter__)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/orm/query.py:2414(_execute_and_instances)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/engine/base.py:659(execute)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/sql/elements.py:321(_execute_on_connection)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/engine/base.py:788(_execute_clauseelement)

...

在上面的例子中，我们可以看到 instances() SQLAlchemy 函数被调用了 222 次（递归调用，外部调用 21 次），总共花费了 .011 秒来执行所有调用。

执行慢

这些调用的具体情况可以告诉我们时间花在哪里。例如，如果您看到时间花在 cursor.execute() 内部，例如针对 DBAPI：

2    0.102    0.102    0.204    0.102 {method 'execute' of 'sqlite3.Cursor' objects}

这将表示数据库需要很长时间才能开始返回结果，这意味着您的查询应该进行优化，可以通过添加索引或重构查询和/或底层架构来实现。对于这项任务，应该使用数据库后端提供的 EXPLAIN、SHOW PLAN 等系统来分析查询计划。

结果获取慢 - 核心

另一方面，如果你看到与获取行相关的许多调用，或者 fetchall() 的调用时间非常长，这可能意味着查询返回的行数超出了预期，或者获取行本身很慢。ORM 本身通常使用 fetchall() 来获取行（如果使用 Query.yield_per() 选项，则使用 fetchmany()）。

通过在 DBAPI 级别使用 fetchall()，会导致调用非常慢，这可能表示行数过多：

2    0.300    0.600    0.300    0.600 {method 'fetchall' of 'sqlite3.Cursor' objects}

即使最终结果似乎没有很多行，但意外地大量行数可能是笛卡尔积的结果 - 当多组行未适当地连接在一起时。如果在复杂查询中使用了错误的Column 对象，从而引入了意外的额外 FROM 子句，那么使用 SQLAlchemy Core 或 ORM 查询往往很容易产生这种行为。

另一方面，在 DBAPI 级别使用 fetchall() 快速，但当 SQLAlchemy 的CursorResult 被要求执行 fetchall() 时却很慢，可能表示数据类型处理慢，比如 Unicode 转换等：

# the DBAPI cursor is fast...
2    0.020    0.040    0.020    0.040 {method 'fetchall' of 'sqlite3.Cursor' objects}

...

# but SQLAlchemy's result proxy is slow, this is type-level processing
2    0.100    0.200    0.100    0.200 lib/sqlalchemy/engine/result.py:778(fetchall)

在某些情况下，后端可能正在进行不需要的类型级处理。更具体地说，看到类型 API 中的慢调用是更好的指标 - 下面是使用此类类型时的情况：

from sqlalchemy import TypeDecorator
import time

class Foo(TypeDecorator):
    impl = String

    def process_result_value(self, value, thing):
        # intentionally add slowness for illustration purposes
        time.sleep(0.001)
        return value

这个有意慢操作的分析输出可以看起来像这样：

200    0.001    0.000    0.237    0.001 lib/sqlalchemy/sql/type_api.py:911(process)
200    0.001    0.000    0.236    0.001 test.py:28(process_result_value)
200    0.235    0.001    0.235    0.001 {time.sleep}

也就是说，我们在 type_api 系统中看到了许多昂贵的调用，而实际耗时的是 time.sleep() 调用。

请确保查阅方言文档以了解关于这个级别已知的性能调优建议的说明，特别是对于像 Oracle 这样的数据库。在这种情况下可能有一些关于确保数字精度或字符串处理的系统不需要在所有情况下都需要的系统。

在行提取性能受影响的更低级别可能还有更多的点；例如，如果花费的时间似乎集中在像 socket.receive() 这样的调用上，这可能表明除了实际的网络连接之外，一切都很快，而且花费了太多时间在数据在网络上传输上。

结果获取速度慢 - ORM

要检测 ORM 提取行的慢速（这是性能关注的最常见领域），调用如 populate_state() 和 _instance() 将说明单个 ORM 对象的填充情况：

# the ORM calls _instance for each ORM-loaded row it sees, and
# populate_state for each ORM-loaded row that results in the population
# of an object's attributes
220/20    0.001    0.000    0.010    0.000 lib/sqlalchemy/orm/loading.py:327(_instance)
220/20    0.000    0.000    0.009    0.000 lib/sqlalchemy/orm/loading.py:284(populate_state)

ORM 将行转换为 ORM 映射对象的速度慢是这个操作的复杂性与 cPython 的开销的产物。减轻这种情况的常见策略包括：

获取单个列而不是完整的实体，即：

select(User.id, User.name)

而不是：

select(User)

使用Bundle对象来组织基于列的结果：

u_b = Bundle("user", User.id, User.name)
a_b = Bundle("address", Address.id, Address.email)

for user, address in session.execute(select(u_b, a_b).join(User.addresses)):
    ...

使用结果缓存 - 参见 Dogpile Caching 了解深入示例。

考虑使用像 PyPy 这样的更快的解释器。

分析结果可能有点令人生畏，但经过一些实践后，它们就会变得非常容易阅读。

另请参阅

性能 - 一套具有捆绑式分析功能的性能演示。

我正在使用 ORM 插入 40 万行，但速度真的很慢！

ORM 插入的性质已经发生了变化，因为大多数包含的驱动程序在 SQLAlchemy 2.0 中都使用了 RETURNING 和 insertmanyvalues 支持。请参阅除了 MySQL 外的所有后端现在都已实现优化的 ORM 批量插入一节了解详情。

总的来说，SQLAlchemy 内置的驱动程序，除了 MySQL 外，现在应该提供非常快的 ORM 批量插入性能。

第三方驱动程序也可以选择使用一些小的代码更改来使用新的批量基础架构，假设他们的后端支持所需的语法。SQLAlchemy 开发人员鼓励第三方方言的用户发布关于这些驱动程序的问题，以便他们可以联系 SQLAlchemy 开发人员寻求帮助。

为什么我升级到 1.4 和/或 2.x 后我的应用程序变慢了？

截至版本 1.4，SQLAlchemy 包含一个 SQL 编译缓存设施，它允许核心和 ORM SQL 构造缓存它们的字符串形式，以及用于从语句中获取结果的其他结构信息，从而在下次使用另一个结构上等价的构造时跳过相对昂贵的字符串编译过程。该系统依赖于为所有 SQL 构造实现的功能，包括诸如Column、select()和TypeEngine对象等，以生成完全代表它们状态的缓存键，在影响 SQL 编译过程的程度上。

缓存系统使得 SQLAlchemy 1.4 及以上版本在将 SQL 构造反复转换为字符串方面比 SQLAlchemy 1.3 更高效。但是，这仅在启用了使用的方言和 SQL 构造的缓存时才有效；如果没有启用，字符串编译通常类似于 SQLAlchemy 1.3，某些情况下速度略有降低。

但是，有一种情况，即如果禁用了 SQLAlchemy 的新缓存系统（由于以下原因），则 ORM 的性能实际上可能显着低于 1.3 或其他先前版本，原因是在 1.3 和以前的版本中，ORM 惰性加载器和对象刷新查询中没有缓存，而是使用了现在已经过时的BakedQuery系统。如果应用程序在切换到 1.4 后性能显着下降（测量操作完成所需的时间为 30%或更高），则这可能是问题的主要原因，以下是缓解措施。

另请参阅

SQL 编译缓存 - 缓存系统概述

对象不会产生缓存键，性能影响 - 关于未启用缓存的元素生成警告的附加信息。

第一步 - 打开 SQL 日志并确认缓存是否正常工作

在这里，我们希望使用引擎日志记录中描述的技术，寻找带有[no key]指示器或甚至[dialect does not support caching]的语句。对于成功参与缓存系统的 SQL 语句，我们将看到首次调用语句时指示为[generated in Xs]，对于绝大多数后续语句，将看到[cached since Xs ago]。如果特别是对于 SELECT 语句，或者如果由于[dialect does not support caching]而完全禁用缓存，则这可能是性能显著下降的原因。

另请参阅

使用日志估算缓存性能

第二步 - 确定是哪些构造物阻止了缓存的启用

假设语句未被缓存，应该会在应用程序日志的早期（仅适用于 SQLAlchemy 1.4.28 及以上版本）中发出警告，指示不参与缓存的方言、TypeEngine 对象和 SQL 构造。

对于像那些扩展 TypeDecorator 和 UserDefinedType 的用户定义数据类型，警告将如下所示：

sqlalchemy.ext.SAWarning: MyType will not produce a cache key because the
``cache_ok`` attribute is not set to True. This can have significant
performance implications including some performance degradations in
comparison to prior SQLAlchemy versions. Set this attribute to True if this
type object's state is safe to use in a cache key, or False to disable this
warning.

对于自定义和第三方 SQL 元素，比如那些使用自定义 SQL 构造和编译扩展中描述的技术构建的元素，这些警告将如下所示：

sqlalchemy.exc.SAWarning: Class MyClass will not make use of SQL
compilation caching as it does not set the 'inherit_cache' attribute to
``True``. This can have significant performance implications including some
performance degradations in comparison to prior SQLAlchemy versions. Set
this attribute to True if this object can make use of the cache key
generated by the superclass. Alternatively, this attribute may be set to
False which will disable this warning.

对于使用 Dialect 类层次结构的自定义和第三方方言，警告将如下所示：

sqlalchemy.exc.SAWarning: Dialect database:driver will not make use of SQL
compilation caching as it does not set the 'supports_statement_cache'
attribute to ``True``. This can have significant performance implications
including some performance degradations in comparison to prior SQLAlchemy
versions. Dialect maintainers should seek to set this attribute to True
after appropriate development and testing for SQLAlchemy 1.4 caching
support. Alternatively, this attribute may be set to False which will
disable this warning.

第三步 - 为给定对象启用缓存和/或寻求替代方案

缓解缓存不足的步骤包括：

检查并设置 ExternalType.cache_ok 为 True，用于所有继承自 TypeDecorator、UserDefinedType 的自定义类型，以及这些类型的子类，如 PickleType。只有在自定义类型不包含影响其呈现 SQL 的其他状态属性时才设置此项：

class MyCustomType(TypeDecorator):
    cache_ok = True
    impl = String

如果使用的类型来自第三方库，请与该库的维护者联系，以便进行调整和发布。

另请参阅

ExternalType.cache_ok - 关于启用自定义数据类型缓存的要求背景信息。

确保第三方方言将 Dialect.supports_statement_cache 设置为 True。这表示第三方方言的维护者已确保其方言与 SQLAlchemy 1.4 或更高版本兼容，并且他们的方言不包含可能妨碍缓存的任何编译特性。由于有一些常见的编译模式实际上可能会干扰缓存，因此方言维护者必须仔细检查和测试此内容，并针对任何无法与缓存一起使用的旧模式进行调整。

另请参见

第三方方言的缓存 - 第三方方言参与 SQL 语句缓存的背景和示例。

自定义 SQL 类，包括使用自定义 SQL 构造和编译扩展可能创建的所有 DQL / DML 构造，以及对象的临时子类，如Column或Table。HasCacheKey.inherit_cache 属性可以设置为 True，用于简单的子类，这些子类不包含影响 SQL 编译的子类特定状态信息。

另请参见

为自定义构造启用缓存支持 - 应用HasCacheKey.inherit_cache 属性的指南。

另请参见

SQL 编译缓存 - 缓存系统概述

对象不会生成缓存键，性能影响 - 在未为特定构造和/或方言启用缓存时发出警告的背景信息。

步骤一 - 打开 SQL 日志记录并确认缓存是否起作用

在这里，我们想要使用引擎日志记录中描述的技术，查找具有 [no key] 指示器或甚至 [dialect does not support caching] 的语句。当首次调用语句时，我们将看到参与缓存系统的 SQL 语句指示 [generated in Xs]，然后对于绝大多数后续语句，指示为 [cached since Xs ago]。如果 [no key] 特别是对于 SELECT 语句普遍存在，或者如果由于 [dialect does not support caching] 而完全禁用缓存，这可能是导致性能严重下降的原因。

另请参见

使用日志估算缓存性能

第二步 - 确定哪些构造阻止了缓存的启用

假设语句未被缓存，则应在应用程序的日志中尽早发出警告（仅适用于 SQLAlchemy 1.4.28 及以上版本），指示不参与缓存的方言、TypeEngine 对象和 SQL 构造。

对于用户定义的数据类型，比如那些扩展了TypeDecorator和UserDefinedType的数据类型，警告信息如下：

sqlalchemy.ext.SAWarning: MyType will not produce a cache key because the
``cache_ok`` attribute is not set to True. This can have significant
performance implications including some performance degradations in
comparison to prior SQLAlchemy versions. Set this attribute to True if this
type object's state is safe to use in a cache key, or False to disable this
warning.

对于自定义和第三方 SQL 元素，例如使用自定义 SQL 构造和编译扩展中描述的技术构造的元素，这些警告信息如下：

sqlalchemy.exc.SAWarning: Class MyClass will not make use of SQL
compilation caching as it does not set the 'inherit_cache' attribute to
``True``. This can have significant performance implications including some
performance degradations in comparison to prior SQLAlchemy versions. Set
this attribute to True if this object can make use of the cache key
generated by the superclass. Alternatively, this attribute may be set to
False which will disable this warning.

对于使用 Dialect 类层次结构的自定义和第三方方言，警告信息如下：

sqlalchemy.exc.SAWarning: Dialect database:driver will not make use of SQL
compilation caching as it does not set the 'supports_statement_cache'
attribute to ``True``. This can have significant performance implications
including some performance degradations in comparison to prior SQLAlchemy
versions. Dialect maintainers should seek to set this attribute to True
after appropriate development and testing for SQLAlchemy 1.4 caching
support. Alternatively, this attribute may be set to False which will
disable this warning.

第三步 - 为给定的对象启用缓存和/或寻找替代方案

缓存缺失的缓解步骤包括：

查看并设置 ExternalType.cache_ok，对于所有扩展自 TypeDecorator、UserDefinedType 的自定义类型，以及这些类型的子类，例如 PickleType。仅在自定义类型不包含影响其呈现 SQL 的其他状态属性时才设置此选项：

class MyCustomType(TypeDecorator):
    cache_ok = True
    impl = String

如果使用的类型来自第三方库，请咨询该库的维护者，以便进行调整并发布。

另请参阅

ExternalType.cache_ok - 启用自定义数据类型缓存的要求背景。

确保第三方方言设置Dialect.supports_statement_cache为 True。这表示第三方方言的维护者已确保他们的方言与 SQLAlchemy 1.4 或更高版本兼容，并且他们的方言不包含任何可能干扰缓存的编译特性。由于有一些常见的编译模式实际上可能会干扰缓存，因此方言维护者需要仔细检查和测试，并调整任何不适用于缓存的旧模式。

另请参阅

第三方方言的缓存 - 第三方方言参与 SQL 语句缓存的背景和示例。

自定义 SQL 类，包括使用自定义 SQL 构造和编译扩展创建的所有 DQL / DML 构造，以及对象的临时子类，例如 Column 或 Table。对于不包含影响 SQL 编译的子类特定状态信息的简单子类，可以将 HasCacheKey.inherit_cache 属性设置为 True。

另请参阅

为自定义结构启用缓存支持 - 应用HasCacheKey.inherit_cache属性的指南。

另请参阅

SQL 编译缓存 - 缓存系统概述

对象不会生成缓存密钥，性能影响 - 当为特定构造和/或方言禁用缓存时发出警告的背景信息。

如何对由 SQLAlchemy 驱动的应用进行性能分析？

寻找性能问题通常涉及两种策略。一种是查询分析，另一种是代码分析。

查询分析

有时候，仅仅通过普通的 SQL 记录（通过 python 的 logging 模块启用，或者通过create_engine()上的 echo=True 参数启用）就可以了解操作花费的时间。例如，如果在 SQL 操作之后记录了一些内容，你会在日志中看到像这样的信息：

17:37:48,325 INFO  [sqlalchemy.engine.base.Engine.0x...048c] SELECT ...
17:37:48,326 INFO  [sqlalchemy.engine.base.Engine.0x...048c] {<params>}
17:37:48,660 DEBUG [myapp.somemessage]

如果你在操作之后记录了 myapp.somemessage，你就知道完成 SQL 部分花费了 334ms。

记录 SQL 也会说明是否发出了数十/数百个查询，这些查询可以更好地组织成更少的查询。在使用 SQLAlchemy ORM 时，“急加载”功能提供了部分 (contains_eager()) 或完全 (joinedload(), subqueryload()) 自动化此活动，但是没有 ORM 的“急加载”通常意味着使用连接，以便跨多个表加载结果集，而不是随着深度的增加而增加查询次数（即 r + r*r2 + r*r2*r3 …）

对于更长期的查询分析，或者实现应用程序端的“慢查询”监视器，可以使用事件拦截游标执行，使用以下类似的方法：

from sqlalchemy import event
from sqlalchemy.engine import Engine
import time
import logging

logging.basicConfig()
logger = logging.getLogger("myapp.sqltime")
logger.setLevel(logging.DEBUG)

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("query_start_time", []).append(time.time())
    logger.debug("Start Query: %s", statement)

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info["query_start_time"].pop(-1)
    logger.debug("Query Complete!")
    logger.debug("Total Time: %f", total)

在上面的例子中，我们使用 ConnectionEvents.before_cursor_execute() 和 ConnectionEvents.after_cursor_execute() 事件来建立在执行语句时的拦截点。我们使用 info 字典在连接上附加一个计时器；我们在这里使用堆栈，以处理游标执行事件可能嵌套的偶发情况。

代码分析

如果日志显示单个查询花费的时间过长，您需要了解在数据库内部处理查询、通过网络发送结果、由 DBAPI 处理以及最终由 SQLAlchemy 的结果集和/或 ORM 层接收的时间分别花费了多少。每个阶段都可能出现自己的瓶颈，具体取决于具体情况。

为此，您需要使用 Python Profiling Module。以下是一个将分析集成到上下文管理器中的简单示例：

import cProfile
import io
import pstats
import contextlib

@contextlib.contextmanager
def profiled():
    pr = cProfile.Profile()
    pr.enable()
    yield
    pr.disable()
    s = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
    ps.print_stats()
    # uncomment this to see who's calling what
    # ps.print_callers()
    print(s.getvalue())

要对代码段进行分析：

with profiled():
    session.scalars(select(FooClass).where(FooClass.somevalue == 8)).all()

分析的输出可以用来了解时间花费在哪里。分析输出的一部分看起来像这样：

13726 function calls (13042 primitive calls) in 0.014 seconds

Ordered by: cumulative time

ncalls  tottime  percall  cumtime  percall filename:lineno(function)
222/21    0.001    0.000    0.011    0.001 lib/sqlalchemy/orm/loading.py:26(instances)
220/20    0.002    0.000    0.010    0.001 lib/sqlalchemy/orm/loading.py:327(_instance)
220/20    0.000    0.000    0.010    0.000 lib/sqlalchemy/orm/loading.py:284(populate_state)
   20    0.000    0.000    0.010    0.000 lib/sqlalchemy/orm/strategies.py:987(load_collection_from_subq)
   20    0.000    0.000    0.009    0.000 lib/sqlalchemy/orm/strategies.py:935(get)
    1    0.000    0.000    0.009    0.009 lib/sqlalchemy/orm/strategies.py:940(_load)
   21    0.000    0.000    0.008    0.000 lib/sqlalchemy/orm/strategies.py:942(<genexpr>)
    2    0.000    0.000    0.004    0.002 lib/sqlalchemy/orm/query.py:2400(__iter__)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/orm/query.py:2414(_execute_and_instances)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/engine/base.py:659(execute)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/sql/elements.py:321(_execute_on_connection)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/engine/base.py:788(_execute_clauseelement)

...

在上面的例子中，我们可以看到 instances() SQLAlchemy 函数被调用了 222 次（递归调用，并且从外部调用了 21 次），所有调用总共花费了 .011 秒。

执行速度慢

这些调用的具体信息可以告诉我们时间花费在哪里。例如，如果您看到时间花费在 cursor.execute() 内部，例如针对 DBAPI：

2    0.102    0.102    0.204    0.102 {method 'execute' of 'sqlite3.Cursor' objects}

这表明数据库启动返回结果花费了很长时间，这意味着你的查询应该进行优化，可以通过添加索引或重构查询和/或底层模式来实现。对于这项任务，有必要分析查询计划，使用诸如 EXPLAIN、SHOW PLAN 等数据库后端提供的系统。

结果获取缓慢 - Core

另一方面，如果您看到与获取行相关的成千上万次调用，或者对 fetchall() 的调用非常长，这可能意味着您的查询返回的行数比预期的多，或者获取行本身很慢。ORM 本身通常使用 fetchall() 来获取行（或者如果使用了 Query.yield_per() 选项，则使用 fetchmany()）。

非常大量的行数将通过在 DBAPI 级别非常慢的调用 fetchall() 来表示：

2    0.300    0.600    0.300    0.600 {method 'fetchall' of 'sqlite3.Cursor' objects}

即使最终结果似乎没有很多行，但出现意外地大量行的情况可能是笛卡尔积的结果 - 当多组行未经适当连接地组合在一起时。如果在复杂查询中使用了错误的 Column 对象，导致引入意外的额外 FROM 子句，那么用 SQLAlchemy Core 或 ORM 查询往往很容易产生这种行为。

另一方面，在 DBAPI 级别快速调用 fetchall()，但当 SQLAlchemy 的 CursorResult 要求执行 fetchall() 时出现缓慢，可能表示数据类型的处理速度较慢，例如 unicode 转换等：

# the DBAPI cursor is fast...
2    0.020    0.040    0.020    0.040 {method 'fetchall' of 'sqlite3.Cursor' objects}

...

# but SQLAlchemy's result proxy is slow, this is type-level processing
2    0.100    0.200    0.100    0.200 lib/sqlalchemy/engine/result.py:778(fetchall)

在某些情况下，后端可能正在进行不需要的类型级处理。更具体地说，看到类型 API 内的调用很慢更好，下面是我们使用这样一个类型时的情况：

from sqlalchemy import TypeDecorator
import time

class Foo(TypeDecorator):
    impl = String

    def process_result_value(self, value, thing):
        # intentionally add slowness for illustration purposes
        time.sleep(0.001)
        return value

这个有意慢操作的分析结果可以看起来像这样：

200    0.001    0.000    0.237    0.001 lib/sqlalchemy/sql/type_api.py:911(process)
200    0.001    0.000    0.236    0.001 test.py:28(process_result_value)
200    0.235    0.001    0.235    0.001 {time.sleep}

也就是说，我们在 type_api 系统内看到了很多昂贵的调用，而实际耗时的事情是 time.sleep() 调用。

确保检查 Dialect documentation 以了解此级别已知的性能调整建议，特别是对于像 Oracle 这样的数据库。可能有关于确保数值精度或字符串处理的系统可能并不在所有情况下都需要。

还可能存在更低级别的点导致行获取性能下降；例如，如果时间主要花在像 socket.receive() 这样的调用上，这可能表明除了网络连接本身外，其他所有东西都很快，而且花费了太多时间在网络上传输数据。

结果获取缓慢 - ORM

要检测 ORM 获取行的速度慢（这是性能关注的最常见领域），像 populate_state() 和 _instance() 这样的调用将说明单个 ORM 对象的加载情况：

# the ORM calls _instance for each ORM-loaded row it sees, and
# populate_state for each ORM-loaded row that results in the population
# of an object's attributes
220/20    0.001    0.000    0.010    0.000 lib/sqlalchemy/orm/loading.py:327(_instance)
220/20    0.000    0.000    0.009    0.000 lib/sqlalchemy/orm/loading.py:284(populate_state)

ORM 将行转换为 ORM 映射对象的速度慢是由于此操作的复杂性与 cPython 的开销相结合造成的。缓解这种情况的常见策略包括：

获取单个列而不是完整的实体，即：

select(User.id, User.name)

而不是：

select(User)

使用 Bundle 对象组织基于列的结果：

u_b = Bundle("user", User.id, User.name)
a_b = Bundle("address", Address.id, Address.email)

for user, address in session.execute(select(u_b, a_b).join(User.addresses)):
    ...

使用结果缓存 - 有关此的详细示例，请参见 Dogpile Caching。

考虑使用像 PyPy 这样的更快的解释器。

一次性分析的输出可能有点令人生畏，但经过一些练习后，它们会变得非常容易阅读。

另请参阅

性能 - 一套具有捆绑分析功能的性能演示。

查询分析

有时，仅仅记录 SQL（通过 Python 的 logging 模块启用或通过 create_engine() 的 echo=True 参数启用）可以让人了解到事情花费的时间。例如，如果在 SQL 操作之后记录了某些内容，则在日志中会看到类似于以下内容：

17:37:48,325 INFO  [sqlalchemy.engine.base.Engine.0x...048c] SELECT ...
17:37:48,326 INFO  [sqlalchemy.engine.base.Engine.0x...048c] {<params>}
17:37:48,660 DEBUG [myapp.somemessage]

如果在操作之后记录了 myapp.somemessage，则知道完成 SQL 部分花费了 334ms。

记录 SQL 还会说明是否发出了数十个/数百个查询，这些查询可以更好地组织为更少的查询。在使用 SQLAlchemy ORM 时，“急加载”功能提供了部分（contains_eager()）或完全（joinedload()、subqueryload()）自动化此活动，但在没有 ORM 的“急加载”时，通常意味着使用连接以便在一个结果集中加载多个表的结果，而不是随着深度的增加而增加查询次数（即 r + r*r2 + r*r2*r3 …）

为了更长期地分析查询，或者实现应用程序端的“慢查询”监视器，可以使用事件拦截游标执行，使用以下类似的方法：

from sqlalchemy import event
from sqlalchemy.engine import Engine
import time
import logging

logging.basicConfig()
logger = logging.getLogger("myapp.sqltime")
logger.setLevel(logging.DEBUG)

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("query_start_time", []).append(time.time())
    logger.debug("Start Query: %s", statement)

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info["query_start_time"].pop(-1)
    logger.debug("Query Complete!")
    logger.debug("Total Time: %f", total)

以上，我们使用ConnectionEvents.before_cursor_execute()和ConnectionEvents.after_cursor_execute()事件在语句执行时建立拦截点。我们在连接上附加一个计时器，使用info字典；在这里我们使用堆栈，偶尔情况下游标执行事件可能是嵌套的。

代码性能分析

如果日志显示个别查询花费了太长时间，您需要详细了解在数据库内部处理查询、通过网络发送结果、由 DBAPI 处理以及最终由 SQLAlchemy 的结果集和/或 ORM 层接收的时间。每个阶段都可能存在自己的瓶颈，具体取决于特定情况。

为此，您需要使用Python 性能分析模块。以下是一个将性能分析嵌入到上下文管理器中的简单示例：

import cProfile
import io
import pstats
import contextlib

@contextlib.contextmanager
def profiled():
    pr = cProfile.Profile()
    pr.enable()
    yield
    pr.disable()
    s = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
    ps.print_stats()
    # uncomment this to see who's calling what
    # ps.print_callers()
    print(s.getvalue())

对代码段进行性能分析：

with profiled():
    session.scalars(select(FooClass).where(FooClass.somevalue == 8)).all()

性能分析的输出可以让我们了解时间消耗在哪里。性能分析的一部分输出如下所示：

13726 function calls (13042 primitive calls) in 0.014 seconds

Ordered by: cumulative time

ncalls  tottime  percall  cumtime  percall filename:lineno(function)
222/21    0.001    0.000    0.011    0.001 lib/sqlalchemy/orm/loading.py:26(instances)
220/20    0.002    0.000    0.010    0.001 lib/sqlalchemy/orm/loading.py:327(_instance)
220/20    0.000    0.000    0.010    0.000 lib/sqlalchemy/orm/loading.py:284(populate_state)
   20    0.000    0.000    0.010    0.000 lib/sqlalchemy/orm/strategies.py:987(load_collection_from_subq)
   20    0.000    0.000    0.009    0.000 lib/sqlalchemy/orm/strategies.py:935(get)
    1    0.000    0.000    0.009    0.009 lib/sqlalchemy/orm/strategies.py:940(_load)
   21    0.000    0.000    0.008    0.000 lib/sqlalchemy/orm/strategies.py:942(<genexpr>)
    2    0.000    0.000    0.004    0.002 lib/sqlalchemy/orm/query.py:2400(__iter__)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/orm/query.py:2414(_execute_and_instances)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/engine/base.py:659(execute)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/sql/elements.py:321(_execute_on_connection)
    2    0.000    0.000    0.002    0.001 lib/sqlalchemy/engine/base.py:788(_execute_clauseelement)

...

以上，我们可以看到instances() SQLAlchemy 函数被调用了 222 次（递归调用，从外部调用了 21 次），所有调用总共花了 0.011 秒。

执行速度慢

这些调用的具体细节可以告诉我们时间都花在哪里。例如，如果您看到在cursor.execute()内花费了时间，例如针对 DBAPI：

2    0.102    0.102    0.204    0.102 {method 'execute' of 'sqlite3.Cursor' objects}

这将表明数据库花费了很长时间才开始返回结果，这意味着您的查询应该被优化，可以通过添加索引或重组查询和/或底层架构来完成此任务。对查询计划的分析是有必要的，可以使用像 EXPLAIN、SHOW PLAN 等数据库后端提供的系统。

结果获取速度慢 - 核心

另一方面，如果你看到与获取行有关的成千上万次调用，或者对fetchall()的非常长时间的调用，这可能意味着你的查询返回的行数超出了预期，或者获取行本身的速度很慢。ORM 本身通常使用fetchall()来获取行（如果使用了Query.yield_per()选项，则使用fetchmany()）。

非常慢的fetchall()调用会被 DBAPI 级别上指示出异乎寻常的大量行：

2    0.300    0.600    0.300    0.600 {method 'fetchall' of 'sqlite3.Cursor' objects}

如果行数意外地很大，即使最终结果似乎没有很多行，也可能是笛卡尔积的结果 - 当多组行组合在一起而没有适当地连接表时。如果在复杂查询中使用了错误的Column对象，拉入意外的 FROM 子句，很容易在 SQLAlchemy Core 或 ORM 查询中产生这种行为。

另一方面，在 DBAPI 级别快速调用fetchall()，但当要求 SQLAlchemy 的CursorResult执行fetchall()时变慢，可能表明在处理数据类型（如 unicode 转换等）时存在缓慢：

# the DBAPI cursor is fast...
2    0.020    0.040    0.020    0.040 {method 'fetchall' of 'sqlite3.Cursor' objects}

...

# but SQLAlchemy's result proxy is slow, this is type-level processing
2    0.100    0.200    0.100    0.200 lib/sqlalchemy/engine/result.py:778(fetchall)

在某些情况下，后端可能正在进行不必要的类型级处理。更具体地说，看到类型 API 中的调用很慢更好，下面是当我们使用这样的类型时的情况：

from sqlalchemy import TypeDecorator
import time

class Foo(TypeDecorator):
    impl = String

    def process_result_value(self, value, thing):
        # intentionally add slowness for illustration purposes
        time.sleep(0.001)
        return value

这个故意缓慢操作的分析输出看起来像这样：

200    0.001    0.000    0.237    0.001 lib/sqlalchemy/sql/type_api.py:911(process)
200    0.001    0.000    0.236    0.001 test.py:28(process_result_value)
200    0.235    0.001    0.235    0.001 {time.sleep}

也就是说，我们在type_api系统中看到许多昂贵的调用，而实际耗时的是time.sleep()调用。

确保查看 Dialect 文档以获取关于已知性能调优建议的说明，特别是对于像 Oracle 这样的数据库。可能存在确保数字精度或字符串处理的系统，在某些情况下可能不需要。

还可能存在更多低级别的点导致行提取性能下降；例如，如果花费的时间似乎集中在像socket.receive()这样的调用上，那可能表明除了实际的网络连接外，一切都很快，而花费太多时间在数据在网络上传输上。

结果提取缓慢 - ORM

要检测 ORM 提取行的缓慢（这是性能关注的最常见领域），像populate_state()和_instance()这样的调用将说明单个 ORM 对象的填充：

# the ORM calls _instance for each ORM-loaded row it sees, and
# populate_state for each ORM-loaded row that results in the population
# of an object's attributes
220/20    0.001    0.000    0.010    0.000 lib/sqlalchemy/orm/loading.py:327(_instance)
220/20    0.000    0.000    0.009    0.000 lib/sqlalchemy/orm/loading.py:284(populate_state)

ORM 在将行转换为 ORM 映射对象时的缓慢是这个操作的复杂性与 cPython 的开销相结合的产物。减轻这种情况的常见策略包括：

获取单个列而不是完整实体，也就是：

select(User.id, User.name)

而不是：

select(User)

使用Bundle对象来组织基于列的结果：

u_b = Bundle("user", User.id, User.name)
a_b = Bundle("address", Address.id, Address.email)

for user, address in session.execute(select(u_b, a_b).join(User.addresses)):
    ...

使用结果缓存 - 参见 Dogpile Caching 以获取关于此的深入示例。

考虑使用像 PyPy 这样的更快解释器。

分析的输出可能有点令人生畏，但经过一些练习后，它们非常容易阅读。

另请参阅

性能 - 一套具有捆绑分析功能的性能演示。

我正在使用 ORM 插入 40 万行，速度真的很慢！

ORM 插入的性质已经改变，因为大多数包含的驱动程序在 SQLAlchemy 2.0 中都使用了带有 insertmanyvalues 支持的 RETURNING。详情请参见优化的 ORM 批量插入现在已经为除 MySQL 外的所有后端实现部分。

总的来说，除了 MySQL 外，SQLAlchemy 内置的驱动程序现在应该提供非常快速的 ORM 批量插入性能。

第三方驱动程序也可以通过一些小的代码更改选择使用新的批量基础设施，假设它们的后端支持必要的语法。SQLAlchemy 开发人员鼓励第三方方言的用户发布与这些驱动程序相关的问题，以便他们可以联系 SQLAlchemy 开发人员寻求帮助。

会话 / 查询

原文：docs.sqlalchemy.org/en/20/faq/sessions.html

我正在使用 Session 重新加载数据，但它没有看到我在其他地方提交的更改
“此会话的事务已由于刷新期间的先前异常而回滚。”（或类似消息）
- 但为什么 flush() 坚持发出 ROLLBACK？
- 但为什么一个自动调用 ROLLBACK 不够？为什么我必须再次 ROLLBACK？
如何创建一个始终向每个查询添加特定过滤器的查询？
我的查询返回的对象数与 query.count() 告诉我的不一致 - 为什么？
我已经创建了一个对 Outer Join 的映射，虽然查询返回行，但没有返回对象。为什么？
我使用 joinedload() 或 lazy=False 创建 JOIN/OUTER JOIN，但当我尝试添加 WHERE、ORDER BY、LIMIT 等条件时，SQLAlchemy 没有构造正确的查询（这取决于 (OUTER) JOIN）
查询没有 __len__()，为什么？
如何在 ORM 查询中使用文本 SQL？
调用 Session.delete(myobject) 后，我的对象未从父集合中移除！
加载对象时为什么不调用我的 __init__()？
如何在 SA 的 ORM 中使用 ON DELETE CASCADE？
我将实例的“foo_id”属性设置为“7”，但“foo”属性仍然为 None - 应该加载 id 为 #7 的 Foo 吗？
如何遍历与给定对象相关的所有对象？
有没有一种方法可以自动只获取唯一关键字（或其他类型的对象），而不需要查询关键字并获取包含该关键字的行的引用？
为什么 post_update 除了第一个 UPDATE 之外还会发出 UPDATE？

我重新加载了我的会话中的数据，但它没有看到我在其他地方提交的更改

这种行为的主要问题在于，会话表现得好像事务处于可串行化隔离状态一样，即使实际上并非如此（通常也不是）。从实际角度来看，这意味着会话在事务范围内已经读取的数据不会发生任何更改。

如果术语“隔离级别”不熟悉，那么您首先需要阅读此链接：

隔离级别

简而言之，可串行化隔离级通常意味着一旦在事务中选择了一系列行，每次重新发出该 SELECT 时都会获得相同的数据。如果您处于较低的隔离级别“可重复读”，您将看到新添加的行（不再看到已删除的行），但对于您已经加载的行，您不会看到任何更改。只有当您处于较低的隔离级别，例如“读取提交”，才有可能看到数据行更改其值。

有关在使用 SQLAlchemy ORM 时控制隔离级别的信息，请参阅设置事务隔离级别 / DBAPI AUTOCOMMIT。

为了极大地简化事情，Session本身是基于完全隔离的事务运行的，并且不会覆盖已经读取的任何映射属性，除非您告诉它这样做。尝试在进行中的事务中重新读取已加载的数据的用例是一个不常见的用例，在许多情况下没有任何效果，因此这被认为是例外而不是规范；为了在这种例外情况下工作，提供了几种方法允许在进行中的事务上下文中重新加载特定数据。

要理解我们在谈论Session时所说的“事务”是什么意思，您的Session只能在事务内部工作。有关概述，请参阅管理事务。

一旦我们弄清楚了我们的隔离级别是什么，并且我们认为我们的隔离级别设置得足够低，以便如果我们重新选择一行，我们应该能够在我们的 Session 中看到新数据，那么我们该如何看到它？

从最常见到最不常见的三种方式：

我们只需结束当前事务，并在下一次访问时启动新事务，通过调用 Session.commit()（请注意，如果 Session 处于较少使用的“自动提交”模式，则还会调用 Session.begin()）。绝大多数应用和用例不会出现无法“看到”其他事务中的数据的问题，因为它们遵循了这一模式，这是短事务最佳实践的核心。有关此问题的一些想法，请参阅何时构建 Session，何时提交它，何时关闭它？。
我们告诉我们的 Session 重新读取它已经读取过的行，要么在下次使用 Session.expire_all() 或 Session.expire() 查询它们时，要么立即在对象上使用 refresh。有关此事的详细信息，请参阅刷新 / 过期。
我们可以在设置了“填充现有”选项的情况下运行整个查询，这样它们读取行时就会覆盖已加载的对象。这是一个在填充现有中描述的执行选项。

但请记住，如果我们的隔离级别是可重复读或更高，则 ORM 无法看到行中的更改，除非我们开始新的事务。## “此会话的事务由于在 flush 期间发生的先前异常而回滚。”（或类似）

当 Session.flush() 引发异常、回滚事务，但后续对 Session 的命令未显式调用 Session.rollback() 或 Session.close() 时会出现此错误。

通常对应于在 Session.flush() 或 Session.commit() 上捕获异常并且不正确处理异常的应用程序。例如：

from sqlalchemy import create_engine, Column, Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base(create_engine("sqlite://"))

class Foo(Base):
    __tablename__ = "foo"
    id = Column(Integer, primary_key=True)

Base.metadata.create_all()

session = sessionmaker()()

# constraint violation
session.add_all([Foo(id=1), Foo(id=1)])

try:
    session.commit()
except:
    # ignore error
    pass

# continue using session without rolling back
session.commit()

使用 Session 应该符合类似于此结构：

try:
    # <use session>
    session.commit()
except:
    session.rollback()
    raise
finally:
    session.close()  # optional, depends on use case

除了 flushes 外，很多事情都可能导致 try/except 内部的失败。应用程序应确保对基于 ORM 的过程应用某种“框架”系统，以便连接和事务资源具有明确的边界，并且如果发生任何失败条件，则可以显式地回滚事务。

这并不意味着整个应用程序都应该有 try/except 块，这将不是一种可扩展的架构。相反，一种典型的方法是，当首次调用基于 ORM 的方法和函数时，从最顶层调用函数的过程将处于一个块中，该块在一系列操作成功完成时提交事务，并且在任何原因失败时，包括失败的 flushes 时回滚事务。也有使用函数装饰器或上下文管理器来实现类似结果的方法。采取的方法取决于正在编写的应用程序的类型。

关于如何组织使用 Session 的详细讨论，请参阅何时构造会话，何时提交它，何时关闭它？。

但为什么 flush() 一定要发出 ROLLBACK 呢？

如果 Session.flush() 能够部分完成而不回滚，那将是很好的，但是由于其当前能力有限，因此这超出了它的当前能力范围，因为其内部簿记必须被修改，以便可以随时停止，并且与已刷新到数据库的内容完全一致。虽然从理论上讲这是可能的，但是增强功能的有用性因为许多数据库操作在任何情况下都需要回滚而大大降低。特别是，Postgres 有一些操作，一旦失败，事务就不允许继续进行：

test=> create table foo(id integer primary key);
NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "foo_pkey" for table "foo"
CREATE TABLE
test=> begin;
BEGIN
test=> insert into foo values(1);
INSERT 0 1
test=> commit;
COMMIT
test=> begin;
BEGIN
test=> insert into foo values(1);
ERROR:  duplicate key value violates unique constraint "foo_pkey"
test=> insert into foo values(2);
ERROR:  current transaction is aborted, commands ignored until end of transaction block

SQLAlchemy 提供的解决这两个问题的方法是支持 SAVEPOINT，通过 Session.begin_nested()。使用 Session.begin_nested()，您可以在事务内部对可能失败的操作进行框架化，然后在保持封闭事务的同时“回滚”到失败之前的点。

但为什么一个自动调用 ROLLBACK 不够？为什么我必须再次 ROLLBACK？

由flush()引起的回滚并不是完整事务块的结束；虽然它结束了正在进行的数据库事务，但从Session的角度来看，仍然存在一个现在处于非活动状态的事务。

给定这样一个块：

sess = Session()  # begins a logical transaction
try:
    sess.flush()

    sess.commit()
except:
    sess.rollback()

在上面，当首次创建一个Session时，假设没有使用“自动提交模式”，在Session内建立了一个逻辑事务。这个事务是“逻辑”的，因为直到调用 SQL 语句时才会实际使用任何数据库资源，此时会启动连接级和 DBAPI 级事务。然而，无论数据库级事务是否是其状态的一部分，逻辑事务将保持不变，直到使用Session.commit()、Session.rollback()或Session.close()结束它。

当上面的flush()失败时，代码仍然位于由 try/commit/except/rollback 块框定的事务中。如果flush()完全回滚逻辑事务，那么当我们到达except:块时，Session将处于干净状态，准备在全新事务上发出新的 SQL，并且调用Session.rollback()将不按顺序进行。特别是，此时Session已经开始了一个新事务，而Session.rollback()将错误地对其进行操作。在这个正常情况下应该进行回滚的地方，而不是允许 SQL 操作在新事务上继续进行，Session会拒绝继续直到显式回滚实际发生。

换句话说，期望调用代码始终调用 Session.commit()、Session.rollback() 或 Session.close() 来对应当前事务块。flush() 保持 Session 在这个事务块内，以便上述代码的行为是可预测和一致的。

如何使一个查询始终对每个查询添加某个过滤器？

请查看 FilteredQuery 的配方。

我的查询返回的对象数与 query.count() 告诉我的不一样 - 为什么？

当 Query 对象被要求返回一个 ORM 映射对象列表时，将根据主键对对象进行去重。也就是说，如果我们例如使用了在使用 ORM 声明形式定义表元数据中描述的 User 映射，并且我们有一个如下所示的 SQL 查询：

q = session.query(User).outerjoin(User.addresses).filter(User.name == "jack")

在教程中使用的示例数据中，addresses 表中有两行数据，其中 name 为 'jack'、主键值为 5 的 users 行。如果我们要求上述查询的 Query.count()，我们将得到答案 2：

>>> q.count()
2

但是，如果我们运行 Query.all() 或遍历查询，我们将得到一个元素：

>>> q.all()
[User(id=5, name='jack', ...)]

这是因为当 Query 对象返回完整实体时，它们会被去重。如果我们改为请求单个列返回，则不会发生这种情况：

>>> session.query(User.id, User.name).outerjoin(User.addresses).filter(
...     User.name == "jack"
... ).all()
[(5, 'jack'), (5, 'jack')]

Query 将去重的主要原因有两个：

允许联接预加载正常工作 - 联接预加载通过使用与相关表的连接来查询行，然后将这些连接的行路由到主对象的集合中来工作。为了做到这一点，它必须获取主对象主键在每个子条目中重复的行。这种模式可以继续到更深层的子集合，以便为单个主对象（如User(id=5)）处理多行。去重允许我们按照查询的方式接收对象，例如所有名为’jack’的User()对象，对于我们来说是一个对象，其中User.addresses集合已经被急加载，就像在relationship()上通过lazy='joined'或通过joinedload()选项指示的那样。为了保持一致性，无论是否建立了联接加载，去重仍然适用，因为急加载背后的关键理念是这些选项永远不会影响结果。
消除关于身份映射的混淆 - 这显然是较不重要的原因。由于Session使用了身份映射，即使我们的 SQL 结果集中有两行主键为 5 的记录，Session 中只有一个User(id=5)对象，必须在其身份上保持唯一，即其主键/类组合。如果一个查询User()对象，多次在列表中获取相同对象实际上并没有太多意义。有序集合可能更好地表示Query 在返回完整对象时所寻求的内容。

Query去重问题仍然存在问题，主要是因为Query.count()方法不一致，并且当前状态是最近的版本中的连接急加载首先被“子查询急加载”策略取代，最近是“select IN 急加载”策略，这两种策略通常更适合于集合急加载。随着这一演变的继续，SQLAlchemy 可能会更改 Query的行为，这也可能涉及新的 API，以更直接地控制此行为，并且也可能更改连接的急加载的行为，以创建更一致的使用模式。

我已经针对外连接创建了映射，但是虽然查询返回行，但没有返回对象。为什么？

由外连接返回的行可能包含主键的部分 NULL，因为主键是两个表的组合。Query对象忽略不具有可接受主键的传入行。根据Mapper上的allow_partial_pks标志的设置，如果该值至少具有一个非 NULL 值，则接受主键，或者如果该值没有 NULL 值，则接受该值。请参见 Mapper上的allow_partial_pks。

我正在使用`joinedload()`或`lazy=False`来创建 JOIN/OUTER JOIN，当我尝试添加 WHERE、ORDER BY、LIMIT 等条件时，SQLAlchemy 没有构造正确的查询。（这依赖于（OUTER）JOIN）

由连接的急加载生成的连接仅用于完全加载相关集合，并设计为不影响查询的主要结果。由于它们是匿名别名，因此不能直接引用。

关于这种行为的详细信息，请参见急加载的禅意。

查询没有`len()`，为什么？

应用于对象的 Python __len__() 魔术方法允许使用len()内置函数来确定集合的长度。直觉上，SQL 查询对象将__len__()链接到Query.count()方法，该方法发出 SELECT COUNT。不可能的原因是评估查询为列表将导致两次 SQL 调用而不是一次：

class Iterates:
    def __len__(self):
        print("LEN!")
        return 5

    def __iter__(self):
        print("ITER!")
        return iter([1, 2, 3, 4, 5])

list(Iterates())

输出：

ITER!
LEN!

如何在 ORM 查询中使用文本 SQL？

参见：

从文本语句获取 ORM 结果 - 使用 Query 进行即席文本块。
在会话中使用 SQL 表达式 - 直接使用 Session 进行文本 SQL 操作。

我调用 `Session.delete(myobject)`，但它没有从父集合中删除！

查看关于删除的注释 - 从集合和标量关系中删除对象以了解此行为的描述。

当加载对象时，为什么我的 `init()` 没有被调用？

查看跨加载保持非映射状态以了解此行为的描述。

我如何在 SA 的 ORM 中使用 ON DELETE CASCADE？

SQLAlchemy 总是对当前加载在 Session 中的依赖行发出 UPDATE 或 DELETE 语句。对于未加载的行，默认情况下会发出 SELECT 语句来加载这些行并更新/删除它们；换句话说，它假定没有配置 ON DELETE CASCADE。要配置 SQLAlchemy 与 ON DELETE CASCADE 协作，请参见使用 ORM 关系的外键 ON DELETE cascade。

我将实例的“foo_id”属性设置为“7”，但“foo”属性仍然为`None` - 它不应该加载具有 id #7 的 Foo 吗？

ORM 的构建不支持根据外键属性变化驱动的关系的立即填充 - 相反，它被设计成反向工作 - 外键属性由 ORM 在幕后处理，最终用户自然设置对象关系。因此，设置 o.foo 的推荐方式是做到这一点 - 设置它！：

foo = session.get(Foo, 7)
o.foo = foo
Session.commit()

当然，对外键属性进行操作是完全合法的。但是，将外键属性设置为新值当前不会触发 relationship() 中涉及的“expire”事件。这意味着对于以下序列：

o = session.scalars(select(SomeClass).limit(1)).first()

# assume the existing o.foo_id value is None;
# accessing o.foo will reconcile this as ``None``, but will effectively
# "load" the value of None
assert o.foo is None

# now set foo_id to something.  o.foo will not be immediately affected
o.foo_id = 7

当首次访问时，o.foo 会加载其有效的数据库值为 None。将 o.foo_id = 7 设置为挂起更改的值为“7”，但尚未刷新 - 因此 o.foo 仍然为 None：

# attribute is already "loaded" as None, has not been
# reconciled with o.foo_id = 7 yet
assert o.foo is None

对于 o.foo 基于外键变化进行加载，通常在提交后自然实现，因为提交既刷新了新的外键值，又使所有状态过期：

session.commit()  # expires all attributes

foo_7 = session.get(Foo, 7)

# o.foo will lazyload again, this time getting the new object
assert o.foo is foo_7

更精简的操作是单独过期属性 - 这可以为任何持久的对象执行，使用Session.expire()：

o = session.scalars(select(SomeClass).limit(1)).first()
o.foo_id = 7
Session.expire(o, ["foo"])  # object must be persistent for this

foo_7 = session.get(Foo, 7)

assert o.foo is foo_7  # o.foo lazyloads on access

请注意，如果对象不是持久的但存在于Session中，则被称为待定。这意味着对象的行尚未插入到数据库中。对于这样的对象，设置foo_id在行被插入之前没有意义；否则还没有行：

new_obj = SomeClass()
new_obj.foo_id = 7

Session.add(new_obj)

# returns None but this is not a "lazyload", as the object is not
# persistent in the DB yet, and the None value is not part of the
# object's state
assert new_obj.foo is None

Session.flush()  # emits INSERT

assert new_obj.foo is foo_7  # now it loads

该配方ExpireRelationshipOnFKChange展示了一个使用 SQLAlchemy 事件的示例，以协调设置具有多对一关系的外键属性。

如何遍历所有与给定对象相关联的对象？

具有其他对象相关联的对象将对应于设置在映射器之间的relationship()构造。此代码片段将迭代所有对象，并校正循环：

from sqlalchemy import inspect

def walk(obj):
    deque = [obj]

    seen = set()

    while deque:
        obj = deque.pop(0)
        if obj in seen:
            continue
        else:
            seen.add(obj)
            yield obj
        insp = inspect(obj)
        for relationship in insp.mapper.relationships:
            related = getattr(obj, relationship.key)
            if relationship.uselist:
                deque.extend(related)
            elif related is not None:
                deque.append(related)

该函数可以演示如下：

Base = declarative_base()

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B", backref="a")

class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    c_id = Column(ForeignKey("c.id"))
    c = relationship("C", backref="bs")

class C(Base):
    __tablename__ = "c"
    id = Column(Integer, primary_key=True)

a1 = A(bs=[B(), B(c=C())])

for obj in walk(a1):
    print(obj)

输出：

<__main__.A object at 0x10303b190>
<__main__.B object at 0x103025210>
<__main__.B object at 0x10303b0d0>
<__main__.C object at 0x103025490>

是否有一种方法可以自动地仅获取唯一的关键词（或其他类型的对象），而不是对关键词进行查询并获取包含该关键词的行的引用？

当人们阅读文档中的多对多示例时，他们会遇到一个事实，即如果您两次创建相同的Keyword，它会被放入数据库两次。这有点不方便。

这个UniqueObject配方是为了解决这个问题而创建的。

为什么 post_update 除了第一个 UPDATE 外还发出 UPDATE？

post_update 功能，在指向自身的行/相互依赖的行文档中记录，涉及在对特定关系绑定的外键进行更改时发出 UPDATE 语句，除了通常会对目标行发出的 INSERT/UPDATE/DELETE 之外。虽然这个 UPDATE 语句的主要目的是与 INSERT 或 DELETE 配对，以便它可以在 INSERT 或 DELETE 操作后设置或取消设置一个外键引用，以断开与相互依赖的外键的循环，但它目前也被捆绑为在目标行本身被更新时发出的第二个 UPDATE。在这种情况下，post_update 发出的 UPDATE 通常是不必要的，并且通常会显得浪费。

然而，对尝试删除这种“UPDATE / UPDATE”行为的一些研究表明，不仅需要在 post_update 实现中进行重大更改，而且还需要在与 post_update 不相关的区域进行更改，以使其工作，因为在某些情况下需要对非 post_update 部分的操作顺序进行反转，这反过来又会影响其他情况，例如正确处理引用主键值的 UPDATE（参见#1063 以获取概念验证）。

答案是，“post_update”用于打破两个相互依赖的外键之间的循环，并且使得此循环打破仅限于目标表的 INSERT/DELETE 意味着其他地方的 UPDATE 语句的顺序需要变得自由化，导致其他边缘情况的破坏。## 我正在使用我的会话重新加载数据，但它没有看到我在其他地方提交的更改

关于这种行为的主要问题是，会话的行为就像事务处于可串行化隔离状态一样，即使事务并不是（通常情况下并不是）。从实际角度来看，这意味着会话不会更改已经在事务范围内读取的任何数据。

如果术语“隔离级别”不熟悉，那么您首先需要阅读此链接：

隔离级别

简而言之，可串行化隔离级别通常意味着一旦您在事务中选择一系列行，您每次重新发出该 SELECT 时都会得到相同的数据。如果您处于较低的隔离级别，例如“可重复读”，您将看到新添加的行（不再看到删除的行），但对于您已经加载的行，您不会看到任何更改。只有当您处于较低的隔离级别时，例如“读取已提交的”，才有可能看到数据行更改其值。

关于在使用 SQLAlchemy ORM 时控制隔离级别的信息，请参阅设置事务隔离级别 / DBAPI AUTOCOMMIT。

要极大地简化事情，Session 本身是在完全隔离的事务中运行的，并且不会覆盖任何已经读取的映射属性，除非你告诉它这样做。在进行中的事务中尝试重新读取已经加载的数据的用例是一个不常见的用例，在许多情况下没有效果，因此这被认为是例外而不是规范；为了在这个例外中工作，提供了几种方法，允许在进行中的事务的上下文中重新加载特定的数据。

当我们谈论Session时，理解我们所说的“事务”是什么意思，你的Session只能在事务内工作。关于此的概述请参阅管理事务。

一旦我们确定了我们的隔离级别，并且我们认为我们的隔离级别设置得足够低，以至于如果我们重新选择一行，我们应该在我们的Session中看到新数据，那我们如何看到它呢？

三种方式，从最常见到最不常见：

我们只需结束当前事务，并在下一次访问时通过调用Session.commit()（请注意，如果Session处于较少使用的“自动提交”模式，则还将调用Session.begin()）。绝大多数应用程序和用例不会出现无法在其他事务中“看到”数据的问题，因为它们遵循这种模式，这是短事务最佳实践的核心。有关此问题的一些想法，请参阅我何时构造一个会话，何时提交它，何时关闭它？。
我们告诉我们的Session重新读取已经读取的行，要么在下次查询它们时使用Session.expire_all()或Session.expire()，要么立即在对象上使用refresh。有关此操作的详细信息，请参阅刷新/过期。
我们可以在设置了“填充现有”选项的情况下运行整个查询，以确保在读取行时覆盖已加载的对象。这是一种在填充现有中描述的执行选项。

但请记住，如果我们的隔离级别是可重复读或更高级别，ORM 无法看到行中的更改，除非我们启动一个新的事务。

“此会话的事务由于刷新期间的先前异常已被回滚。”（或类似内容）

当Session.flush()引发异常，回滚事务，但在未显式调用Session.rollback()或Session.close()的情况下调用Session上的进一步命令时，就会发生这种错误。

这通常对应于一个应用程序在Session.flush()或Session.commit()上捕获异常，但未正确处理异常。例如：

from sqlalchemy import create_engine, Column, Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base(create_engine("sqlite://"))

class Foo(Base):
    __tablename__ = "foo"
    id = Column(Integer, primary_key=True)

Base.metadata.create_all()

session = sessionmaker()()

# constraint violation
session.add_all([Foo(id=1), Foo(id=1)])

try:
    session.commit()
except:
    # ignore error
    pass

# continue using session without rolling back
session.commit()

使用Session应该符合类似于这样的结构：

try:
    # <use session>
    session.commit()
except:
    session.rollback()
    raise
finally:
    session.close()  # optional, depends on use case

许多事情除了刷新之外，都可能导致 try/except 中的失败。应用程序应确保对 ORM 导向的进程应用某种“框架”系统，以便连接和事务资源具有明确定界，并且如果发生任何失败条件，则可以显式回滚事务。

这并不意味着整个应用程序中应该到处都是 try/except 块，这不是可扩展的架构。相反，一个典型的方法是，当首次调用 ORM 导向的方法和函数时，从最顶层调用函数的进程将在成功完成一系列操作时提交事务，并且如果操作因任何原因失败，包括失败的刷新，则回滚事务。还有使用函数装饰器或上下文管理器来实现类似结果的方法。采取的方法取决于正在编写的应用程序的类型。

有关如何组织使用Session的详细讨论，请参见何时构建会话，何时提交会话，何时关闭会话？。

但为什么 flush()坚持发出 ROLLBACK？

如果 Session.flush() 能部分完成然后不回滚，那将会很好，但是由于它当前的能力限制，这是不可能的，因为它的内部记录必须被修改，以便随时停止，并且与已刷新到数据库的内容完全一致。虽然这在理论上是可能的，但增强功能的有用性大大降低了，因为许多数据库操作在任何情况下都需要回滚。特别是 Postgres 有一些操作，一旦失败，事务就不允许继续：

test=> create table foo(id integer primary key);
NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "foo_pkey" for table "foo"
CREATE TABLE
test=> begin;
BEGIN
test=> insert into foo values(1);
INSERT 0 1
test=> commit;
COMMIT
test=> begin;
BEGIN
test=> insert into foo values(1);
ERROR:  duplicate key value violates unique constraint "foo_pkey"
test=> insert into foo values(2);
ERROR:  current transaction is aborted, commands ignored until end of transaction block

SQLAlchemy 提供的解决这两个问题的方法是通过支持 SAVEPOINT，通过 Session.begin_nested()。使用 Session.begin_nested()，您可以在事务中设置一个可能会失败的操作，然后在保持封闭事务的同时“回滚”到其失败之前的点。

但为什么一次自动调用 ROLLBACK 不够？为什么我还必须再次 ROLLBACK？

由 flush() 引起的回滚并不是完整事务块的结束；尽管它结束了正在进行的数据库事务，但从 Session 的角度来看，仍然存在一个处于非活动状态的事务。

鉴于这样的代码块：

sess = Session()  # begins a logical transaction
try:
    sess.flush()

    sess.commit()
except:
    sess.rollback()

在上面的例子中，当一个 Session 第一次被创建时，假设没有使用“自动提交模式”，则在 Session 内建立了一个逻辑事务。这个事务是“逻辑”的，因为它实际上并不使用任何数据库资源，直到调用 SQL 语句时，此时会启动一个连接级别和 DBAPI 级别的事务。然而，无论数据库级别的事务是否是其状态的一部分，逻辑事务都会保持不变，直到使用 Session.commit()、Session.rollback() 或 Session.close() 结束它。

当上面的flush()失败时，代码仍位于由 try/commit/except/rollback 块框定的事务内。如果flush()完全回滚逻辑事务，那么当我们到达except:块时，Session将处于干净状态，准备在全新的事务上发出新的 SQL，并且对Session.rollback()的调用将处于不正确的顺序。特别是，到这一点为止，Session已经开始了一个新的事务，而Session.rollback()将错误地对其进行操作。与其允许 SQL 操作在此处继续新事务，而正常用法规定要进行回滚的地方，则Session拒绝继续，直到显式回滚实际发生。

换句话说，预期调用代码将始终调用Session.commit()、Session.rollback()或Session.close()与当前事务块对应。 flush()保持Session在此事务块中，以便上述代码的行为可预测且一致。

但为什么`flush()`坚持要发出一个 ROLLBACK 呢？

如果Session.flush()可以部分完成然后不回滚，那将是很好的，但是由于其当前能力范围之外，因为其内部记账必须被修改，以便它可以随时停止，并且与已经刷新到数据库的内容完全一致。尽管理论上可能，但增强功能的实用性大大降低了，因为许多数据库操作无论如何都要求回滚。特别是，Postgres 有一些操作，一旦失败，就不允许事务继续：

test=> create table foo(id integer primary key);
NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "foo_pkey" for table "foo"
CREATE TABLE
test=> begin;
BEGIN
test=> insert into foo values(1);
INSERT 0 1
test=> commit;
COMMIT
test=> begin;
BEGIN
test=> insert into foo values(1);
ERROR:  duplicate key value violates unique constraint "foo_pkey"
test=> insert into foo values(2);
ERROR:  current transaction is aborted, commands ignored until end of transaction block

SQLAlchemy 提供解决这两个问题的方法是通过 Session.begin_nested() 支持 SAVEPOINT。使用 Session.begin_nested()，您可以在事务中执行一个可能会失败的操作，然后在维持封闭事务的同时“回滚”到失败之前的状态。

但为什么一次自动调用 ROLLBACK 不够？为什么我必须再次 ROLLBACK？

由 flush() 引起的回滚不是完整事务块的结束；虽然它结束了正在进行的数据库事务，在Session的视角下仍然存在一个现在处于不活动状态的事务。

给定一个如下的块：

sess = Session()  # begins a logical transaction
try:
    sess.flush()

    sess.commit()
except:
    sess.rollback()

在上述情况中，当首次创建一个Session时，假设没有使用“自动提交模式”，则在Session内建立了一个逻辑事务。该事务是“逻辑”的，因为它实际上不使用任何数据库资源，直到调用 SQL 语句，此时开始连接级和 DBAPI 级的事务。但是，无论数据库级事务是否是其状态的一部分，逻辑事务将保持不变，直到使用Session.commit()、Session.rollback()或Session.close()结束为止。

当上面的flush()失败时，代码仍然处于由 try/commit/except/rollback 块框定的事务中。如果flush()完全回滚了逻辑事务，这意味着当我们到达except:块时，Session将处于干净的状态，准备在一个全新的事务中发出新的 SQL，并且对Session.rollback()的调用将会处于顺序错误的状态。特别是，Session此时已经开始了一个新的事务，而Session.rollback()将在错误地执行。与其在这个地方允许 SQL 操作在新的事务中进行，而正常使用指示将要进行回滚的地方，则Session拒绝继续，直到显式回滚实际发生为止。

换句话说，期望调用代码始终调用Session.commit()、Session.rollback()或Session.close()与当前事务块相对应。flush()保持Session在这个事务块内，以便上述代码的行为是可预测且一致的。

如何创建一个始终向每个查询添加特定过滤器的查询？

参见FilteredQuery中的配方。

我的查询返回的对象数量与 query.count() 告诉我的数量不一样 - 为什么？

当Query对象被要求返回一个 ORM 映射对象列表时，将根据主键对对象进行去重。也就是说，如果我们例如使用了在使用 ORM 声明形式定义表元数据中描述的User映射，并且我们有一个如下的 SQL 查询：

q = session.query(User).outerjoin(User.addresses).filter(User.name == "jack")

在上面的例子中，教程中使用的样例数据在addresses表中有两行，对应于名为'jack'的users行，主键值为 5。如果我们对上述查询使用Query.count()，我们将得到答案2：

>>> q.count()
2

然而，如果我们运行Query.all()或者迭代查询，我们会得到一个元素：

>>> q.all()
[User(id=5, name='jack', ...)]

这是因为当Query对象返回完整实体时，它们会被去重。如果我们请求单个列返回，则不会发生这种情况：

>>> session.query(User.id, User.name).outerjoin(User.addresses).filter(
...     User.name == "jack"
... ).all()
[(5, 'jack'), (5, 'jack')]

Query会进行去重的两个主要原因有：

允许连接式贪婪加载正常工作 - 连接式贪婪加载通过使用与相关表的连接查询行，然后将这些连接查询行路由到导航对象的集合中来工作。为了做到这一点，它必须获取重复了主导对象主键的行，以便每个子条目。这种模式可以继续到更进一步的子集合，以便为单个主导对象，如User(id=5)，处理多行。去重允许我们按照查询时的方式接收对象，例如，所有User()对象其名称为'jack'，对我们来说是一个对象，并且User.addresses集合被贪婪加载，就像在relationship()上使用lazy='joined'或通过joinedload()选项指示的那样。为了保持一致性，去重仍然适用于是否已建立连接加载，因为贪婪加载的核心理念是这些选项从不影响结果。
消除关于身份映射的混淆 - 这显然是较不重要的原因。由于Session使用了一个身份映射，即使我们的 SQL 结果集有两行主键为 5 的记录，Session内也只有一个User(id=5)对象，必须以其身份唯一性进行维护，即其主键/类组合。如果查询User()对象，获取相同对象多次在列表中实际上没有太多意义。有序集合可能更能代表Query在返回完整对象时所寻求的内容。

Query 去重的问题仍然存在问题，主要原因是 Query.count() 方法不一致，当前状态是，在最近的发布中，联合急加载首先被“子查询急加载”策略所取代，更近期的是“选择 IN 急加载”策略，这两者通常更适用于集合急加载。随着这种演变的继续，SQLAlchemy 可能会改变 Query 的行为，这也可能涉及到新的 API，以更直接地控制这种行为，并且还可能改变联合急加载的行为，以创建更一致的使用模式。

我已经创建了一个针对 Outer Join 的映射，虽然查询返回了行，但没有返回对象。为什么？

外部连接返回的行可能会对主键的某部分包含 NULL，因为主键是两个表的组合。Query 对象忽略那些没有可接受主键的传入行。根据 Mapper 上 allow_partial_pks 标志的设置，如果值至少有一个非 NULL 值，则接受主键，或者如果值没有 NULL 值，则接受主键。请参阅 Mapper 上的 allow_partial_pks。

当我尝试添加 WHERE、ORDER BY、LIMIT 等条件（这依赖于（外部）JOIN）时，我使用 `joinedload()` 或 `lazy=False` 创建了一个 JOIN/OUTER JOIN，但 SQLAlchemy 在构造查询时出现了问题。

由联合急加载生成的连接仅用于完全加载相关集合，并且设计为不会影响查询的主要结果。由于它们是匿名别名，因此不能直接引用。

关于这种行为的详细信息，请参见 Joined Eager Loading 的禅意。

Query 没有 `len()`，为什么？

Python 中的 __len__() 魔法方法应用于对象，允许使用 len() 内置函数来确定集合的长度。很直观地，一个 SQL 查询对象会将 __len__() 关联到 Query.count() 方法，该方法会发出一个 SELECT COUNT。然而，不可能做到这一点的原因是因为将查询作为列表进行评估会导致两个 SQL 调用而不是一个：

class Iterates:
    def __len__(self):
        print("LEN!")
        return 5

    def __iter__(self):
        print("ITER!")
        return iter([1, 2, 3, 4, 5])

list(Iterates())

输出：

ITER!
LEN!

如何在 ORM 查询中使用 Textual SQL？

请参阅：

从文本语句获取 ORM 结果 - 使用 Query 进行自定义文本块。
使用 SQL 表达式与会话 - 直接使用文本 SQL 与 Session。

我调用`Session.delete(myobject)`但它没有从父集合中删除！

有关此行为的描述，请参阅关于删除的说明 - 从集合和标量关系引用的对象删除。

当我加载对象时，为什么我的`init()`没有被调用？

有关此行为的描述，请参阅跨加载保持非映射状态。

我如何在 SA 的 ORM 中使用 ON DELETE CASCADE？

SQLAlchemy 总是针对当前加载在 Session 中的依赖行发出 UPDATE 或 DELETE 语句。对于未加载的行，默认情况下会发出 SELECT 语句来加载这些行，并对其进行更新/删除；换句话说，它假定未配置 ON DELETE CASCADE。要配置 SQLAlchemy 以配合 ON DELETE CASCADE，请参阅使用 ORM 关系的外键 ON DELETE cascade。

我将我的实例的“foo_id”属性设置为“7”，但“foo”属性仍然为`None` - 它不应该加载 ID 为#7 的 Foo 吗？

ORM 并非以支持从外键属性更改驱动的关系的即时填充方式构建的 - 相反，它设计为以相反的方式工作 - 外键属性由 ORM 在幕后处理，最终用户自然设置对象关系。因此，设置o.foo的推荐方法就是这样 - 设置它！：

foo = session.get(Foo, 7)
o.foo = foo
Session.commit()

当然，操作外键属性是完全合法的。但是，目前设置外键属性为新值不会触发其中涉及的 relationship() 的“过期”事件。这意味着对于以下序列：

o = session.scalars(select(SomeClass).limit(1)).first()

# assume the existing o.foo_id value is None;
# accessing o.foo will reconcile this as ``None``, but will effectively
# "load" the value of None
assert o.foo is None

# now set foo_id to something.  o.foo will not be immediately affected
o.foo_id = 7

当首次访问时，o.foo加载为其有效的数据库值None。设置o.foo_id = 7将使值“7”作为挂起更改，但尚未刷新 - 因此o.foo仍然为None：

# attribute is already "loaded" as None, has not been
# reconciled with o.foo_id = 7 yet
assert o.foo is None

对于o.foo的加载，基于外键变异通常在提交后自然实现，这既刷新了新的外键值，也使所有状态失效：

session.commit()  # expires all attributes

foo_7 = session.get(Foo, 7)

# o.foo will lazyload again, this time getting the new object
assert o.foo is foo_7

一个更简单的操作是单独使属性过期 - 这可以针对任何 persistent 对象使用Session.expire():

o = session.scalars(select(SomeClass).limit(1)).first()
o.foo_id = 7
Session.expire(o, ["foo"])  # object must be persistent for this

foo_7 = session.get(Foo, 7)

assert o.foo is foo_7  # o.foo lazyloads on access

请注意，如果对象不是持久的但存在于Session中，则称为 pending。这意味着对象的行尚未 INSERT 到数据库中。对于这样的对象，设置foo_id在行被插入之前没有意义；否则还没有行：

new_obj = SomeClass()
new_obj.foo_id = 7

Session.add(new_obj)

# returns None but this is not a "lazyload", as the object is not
# persistent in the DB yet, and the None value is not part of the
# object's state
assert new_obj.foo is None

Session.flush()  # emits INSERT

assert new_obj.foo is foo_7  # now it loads

这个方案ExpireRelationshipOnFKChange提供了一个使用 SQLAlchemy 事件的示例，以便协调与多对一关系中的外键属性的设置。

如何遍历与给定对象相关的所有对象？

与之相关的其他对象的对象将与映射器之间设置的relationship()构造相对应。这段代码片段将迭代所有对象，纠正循环：

from sqlalchemy import inspect

def walk(obj):
    deque = [obj]

    seen = set()

    while deque:
        obj = deque.pop(0)
        if obj in seen:
            continue
        else:
            seen.add(obj)
            yield obj
        insp = inspect(obj)
        for relationship in insp.mapper.relationships:
            related = getattr(obj, relationship.key)
            if relationship.uselist:
                deque.extend(related)
            elif related is not None:
                deque.append(related)

函数可以如下所示演示：

Base = declarative_base()

class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    bs = relationship("B", backref="a")

class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    c_id = Column(ForeignKey("c.id"))
    c = relationship("C", backref="bs")

class C(Base):
    __tablename__ = "c"
    id = Column(Integer, primary_key=True)

a1 = A(bs=[B(), B(c=C())])

for obj in walk(a1):
    print(obj)

输出：

<__main__.A object at 0x10303b190>
<__main__.B object at 0x103025210>
<__main__.B object at 0x10303b0d0>
<__main__.C object at 0x103025490>

有没有一种方法可以自动地只有唯一的关键词（或其他类型的对象），而不必查询关键词并获取包含该关键词的行的引用呢？

当人们阅读文档中的多对多示例时，他们会发现如果您创建相同的Keyword两次，它会在数据库中出现两次。这有点不方便。

这个UniqueObject方案是为了解决这个问题而创建的。

为什么 post_update 除了第一个 UPDATE 之外还会发出 UPDATE？

该特性，详细说明请参见指向自身的行/相互依赖行，会在特定关系绑定的外键发生更改时发出 UPDATE 语句，除了会针对目标行通常发出的 INSERT/UPDATE/DELETE 之外。虽然此 UPDATE 语句的主要目的是与该行的 INSERT 或 DELETE 配对，以便它可以在后设置或前取消外键引用，以打破与相互依赖的外键的循环，但目前它也被捆绑为第二个 UPDATE，当目标行本身被 UPDATE 时发出。在这种情况下，post_update 发出的 UPDATE 通常是不必要的，并且通常会显得浪费。

然而，一些研究试图消除这种“UPDATE / UPDATE”行为的努力表明，不仅需要在 post_update 实现中进行重大更改，还需要在与 post_update 无关的领域进行一些变更，以使其生效，因为在某些情况下，非 post_update 方面的操作顺序需要被颠倒，这反过来可能会影响其他情况，比如正确处理引用主键值的 UPDATE（参见#1063以获取概念验证）。

答案是，“post_update”用于打破两个相互依赖的外键之间的循环，并且使得这种循环打破仅限于目标表的 INSERT/DELETE 意味着其他地方 UPDATE 语句的排序需要被放宽，导致其他边缘情况的破坏。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2024-06-22，如有侵权请联系 cloudcommunity@tencent.com 删除

对象

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！