AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

SqlAlchemy 2.0 中文文档（二）（1）

2024-07-05 63
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： SqlAlchemy 2.0 中文文档（二）


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

使用 SELECT 语句

原文：docs.sqlalchemy.org/en/20/tutorial/data_select.html

对于 Core 和 ORM，select() 函数生成一个用于所有 SELECT 查询的 Select 构造。传递给 Core 中的 Connection.execute() 方法和 ORM 中的 Session.execute() 方法，在当前事务中发出 SELECT 语句并通过返回的 Result 对象获取结果行。

ORM 读者 - 这里的内容同样适用于 Core 和 ORM 使用，并提到了基本 ORM 变体用例。然而，还有更多的 ORM 特定功能可用；这些在 ORM 查询指南中有文档记录。

select() SQL 表达式构造

select() 构造以与 insert() 相同的方式构建语句，使用 生成式 方法，其中每个方法都会将更多的状态添加到对象上。与其他 SQL 构造一样，它可以在原地字符串化：

>>> from sqlalchemy import select
>>> stmt = select(user_table).where(user_table.c.name == "spongebob")
>>> print(stmt)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  :name_1

与所有其他语句级别的 SQL 构造相同，要实际运行语句，我们将其传递给执行方法。由于 SELECT 语句返回行，我们始终可以迭代结果对象以获取 Row 对象返回：

>>> with engine.connect() as conn:
...     for row in conn.execute(stmt):
...         print(row)
BEGIN  (implicit)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?
[...]  ('spongebob',)
(1, 'spongebob', 'Spongebob Squarepants')
ROLLBACK

当使用 ORM 时，特别是对 ORM 实体组成的 select() 结构执行时，我们将希望使用 Session.execute() 方法在 Session 上执行它；通过这种方法，我们继续从结果中获取 Row 对象，但是这些行现在可以包括完整的实体，例如 User 类的实例，作为每行中的单独元素：

>>> stmt = select(User).where(User.name == "spongebob")
>>> with Session(engine) as session:
...     for row in session.execute(stmt):
...         print(row)
BEGIN  (implicit)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?
[...]  ('spongebob',)
(User(id=1, name='spongebob', fullname='Spongebob Squarepants'),)
ROLLBACK

以下各节将更详细地讨论 SELECT 构造。

设置 COLUMNS 和 FROM 子句

select() 函数接受表示任意数量Column和/或Table表达式的位置元素，以及一系列兼容对象，这些对象将解析为要从中选择的 SQL 表达式列表，这些表达式将作为结果集中的列返回。这些元素在更简单的情况下还用于创建 FROM 子句，该子句是从传递的列和类似表达式中推断出来的：

>>> print(select(user_table))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account

使用核心方法从单独列进行 SELECT 操作时，可以直接访问Table.c访问器中的Column对象；FROM 子句将被推断为由这些列所代表的所有Table和其他FromClause对象的集合：

>>> print(select(user_table.c.name, user_table.c.fullname))
SELECT  user_account.name,  user_account.fullname
FROM  user_account

或者，当使用任何FromClause的FromClause.c集合，例如Table时，可以通过使用字符串名称的元组指定多个列进行select()操作：

>>> print(select(user_table.c["name", "fullname"]))
SELECT  user_account.name,  user_account.fullname
FROM  user_account

2.0 版本中新增：为FromClause.c集合添加了元组访问器功能

选择 ORM 实体和列

ORM 实体，例如我们的User类以及其上的列映射属性，例如User.name，也参与 SQL 表达语言系统，表示表和列。下面举例说明了从User实体中进行 SELECT 操作的示例，最终呈现的方式与直接使用user_table相同：

>>> print(select(User))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account

当使用 ORM Session.execute()方法执行类似上述语句时，当我们从完整实体（例如User）选择时，与user_table相反，有一个重要的区别，即实体本身作为每行的单个元素返回。也就是说，当我们从上述语句中获取行时，因为在要获取的内容列表中只有User实体，所以我们会收到仅包含一个元素的Row对象，其中包含User类的实例：

>>> row = session.execute(select(User)).first()
BEGIN...
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
[...]  ()
>>> row
(User(id=1, name='spongebob', fullname='Spongebob Squarepants'),)

上述Row只有一个元素，代表User实体：

>>> row[0]
User(id=1, name='spongebob', fullname='Spongebob Squarepants')

实现与上述相同结果的一种高度推荐的便利方法是使用Session.scalars()方法直接执行语句；此方法将返回一个ScalarResult对象，该对象一次性返回每行的第一个“列”，在本例中是User类的实例：

>>> user = session.scalars(select(User)).first()
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
[...]  ()
>>> user
User(id=1, name='spongebob', fullname='Spongebob Squarepants')

或者，我们可以使用类绑定的属性选择 ORM 实体的各个列作为结果行中的单独元素；当这些属性传递给诸如select()之类的构造时，它们会解析为每个属性代表的Column或其他 SQL 表达式：

>>> print(select(User.name, User.fullname))
SELECT  user_account.name,  user_account.fullname
FROM  user_account

当我们使用Session.execute()调用此语句时，我们现在会收到每个值具有单独元素的行，每个元素对应一个单独的列或其他 SQL 表达式：

>>> row = session.execute(select(User.name, User.fullname)).first()
SELECT  user_account.name,  user_account.fullname
FROM  user_account
[...]  ()
>>> row
('spongebob', 'Spongebob Squarepants')

可以混合使用这些方法，如下所示，我们选择User实体的name属性作为行的第一个元素，并将其与完整的Address实体组合为第二个元素：

>>> session.execute(
...     select(User.name, Address).where(User.id == Address.user_id).order_by(Address.id)
... ).all()
SELECT  user_account.name,  address.id,  address.email_address,  address.user_id
FROM  user_account,  address
WHERE  user_account.id  =  address.user_id  ORDER  BY  address.id
[...]  ()
[('spongebob', Address(id=1, email_address='spongebob@sqlalchemy.org')),
('sandy', Address(id=2, email_address='sandy@sqlalchemy.org')),
('sandy', Address(id=3, email_address='sandy@squirrelpower.org'))]

在选择 ORM 实体和列以及将行转换为常见方法方面的方法进一步讨论。

另请参阅

选择 ORM 实体和列 - 在 ORM 查询指南中

从带标签的 SQL 表达式中进行选择

ColumnElement.label()方法以及可用于 ORM 属性的同名方法提供列或表达式的 SQL 标签，允许它在结果集中具有特定名称。当通过名称引用结果行中的任意 SQL 表达式时，这可能会有所帮助：

>>> from sqlalchemy import func, cast
>>> stmt = select(
...     ("Username: " + user_table.c.name).label("username"),
... ).order_by(user_table.c.name)
>>> with engine.connect() as conn:
...     for row in conn.execute(stmt):
...         print(f"{row.username}")
BEGIN  (implicit)
SELECT  ?  ||  user_account.name  AS  username
FROM  user_account  ORDER  BY  user_account.name
[...]  ('Username: ',)
Username: patrick
Username: sandy
Username: spongebob
ROLLBACK

另请参阅

按标签排序或分组 - 我们创建的标签名称也可以在Select的 ORDER BY 或 GROUP BY 子句中引用。

使用文本列表达式进行选择

当我们使用select()函数构造一个Select对象时，通常会向其中传递一系列使用 table metadata 定义的Table和Column对象，或者在使用 ORM 时，我们可能会发送代表表列的 ORM 映射属性。然而，有时也需要在语句中制造任意 SQL 块，比如常量字符串表达式，或者一些直接编写的任意 SQL。

在 Working with Transactions and the DBAPI 中介绍的text()构造实际上可以直接嵌入到Select构造中，如下所示，我们制造了一个硬编码的字符串字面量'some phrase'并将其嵌入到 SELECT 语句中：

>>> from sqlalchemy import text
>>> stmt = select(text("'some phrase'"), user_table.c.name).order_by(user_table.c.name)
>>> with engine.connect() as conn:
...     print(conn.execute(stmt).all())
BEGIN  (implicit)
SELECT  'some phrase',  user_account.name
FROM  user_account  ORDER  BY  user_account.name
[generated  in  ...]  ()
[('some phrase', 'patrick'), ('some phrase', 'sandy'), ('some phrase', 'spongebob')]
ROLLBACK

虽然text()构造可用于大多数位置来插入文字 SQL 短语，但我们实际上更多地处理的是每个代表单个列表达式的文本单元。在这种常见情况下，我们可以使用literal_column()构造来获得更多的功能。此对象类似于text()，但它不是表示任意形式的任意 SQL，而是明确表示一个“列”，然后可以在子查询和其他表达式中进行标记和引用：

>>> from sqlalchemy import literal_column
>>> stmt = select(literal_column("'some phrase'").label("p"), user_table.c.name).order_by(
...     user_table.c.name
... )
>>> with engine.connect() as conn:
...     for row in conn.execute(stmt):
...         print(f"{row.p}, {row.name}")
BEGIN  (implicit)
SELECT  'some phrase'  AS  p,  user_account.name
FROM  user_account  ORDER  BY  user_account.name
[generated  in  ...]  ()
some phrase, patrick
some phrase, sandy
some phrase, spongebob
ROLLBACK

请注意，在使用 text() 或 literal_column() 时，我们正在编写一个语法上的 SQL 表达式，而不是一个字面值。因此，我们必须包括所需的任何引号或语法，以便我们想要看到的 SQL 被呈现出来。## WHERE 子句

SQLAlchemy 允许我们通过使用标准 Python 运算符结合 Column 和类似对象来组合 SQL 表达式，例如 name = 'squidward' 或 user_id > 10。对于布尔表达式，大多数 Python 运算符（如 ==、!=、<、>= 等）生成新的 SQL 表达式对象，而不是纯粹的布尔 True/False 值：

>>> print(user_table.c.name == "squidward")
user_account.name = :name_1
>>> print(address_table.c.user_id > 10)
address.user_id > :user_id_1

我们可以使用这样的表达式来生成 WHERE 子句，方法是将生成的对象传递给 Select.where() 方法：

>>> print(select(user_table).where(user_table.c.name == "squidward"))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  :name_1

要生成由 AND 连接的多个表达式，可以多次调用 Select.where() 方法：

>>> print(
...     select(address_table.c.email_address)
...     .where(user_table.c.name == "squidward")
...     .where(address_table.c.user_id == user_table.c.id)
... )
SELECT  address.email_address
FROM  address,  user_account
WHERE  user_account.name  =  :name_1  AND  address.user_id  =  user_account.id

对于具有相同效果的多个表达式，单次调用 Select.where() 也可以接受多个表达式：

>>> print(
...     select(address_table.c.email_address).where(
...         user_table.c.name == "squidward",
...         address_table.c.user_id == user_table.c.id,
...     )
... )
SELECT  address.email_address
FROM  address,  user_account
WHERE  user_account.name  =  :name_1  AND  address.user_id  =  user_account.id

“AND” 和 “OR” 连接词可以直接使用 and_() 和 or_() 函数，下面是在 ORM 实体方面的示例：

>>> from sqlalchemy import and_, or_
>>> print(
...     select(Address.email_address).where(
...         and_(
...             or_(User.name == "squidward", User.name == "sandy"),
...             Address.user_id == User.id,
...         )
...     )
... )
SELECT  address.email_address
FROM  address,  user_account
WHERE  (user_account.name  =  :name_1  OR  user_account.name  =  :name_2)
AND  address.user_id  =  user_account.id

对于针对单个实体的简单“相等性”比较，还有一种称为 Select.filter_by() 的流行方法，它接受与列键或 ORM 属性名称匹配的关键字参数。它将针对最左边的 FROM 子句或最后一个连接的实体进行过滤：

>>> print(select(User).filter_by(name="spongebob", fullname="Spongebob Squarepants"))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  :name_1  AND  user_account.fullname  =  :fullname_1

另请参阅

运算符参考 - SQLAlchemy 中大多数 SQL 运算符函数的描述## 明确的 FROM 子句和 JOINs

正如前面提到的，FROM 子句通常是基于我们在列子句中设置的表达式以及 Select 的其他元素而 推断 的。

如果我们在 COLUMNS 子句中设置了一个特定 Table 的单个列，它也会将该 Table 放在 FROM 子句中：

>>> print(select(user_table.c.name))
SELECT  user_account.name
FROM  user_account

如果我们从两个表中取列，那么我们得到一个用逗号分隔的 FROM 子句：

>>> print(select(user_table.c.name, address_table.c.email_address))
SELECT  user_account.name,  address.email_address
FROM  user_account,  address

为了将这两个表 JOIN 在一起，我们通常在 Select 上使用两种方法之一。第一种是 Select.join_from() 方法，它允许我们明确指示 JOIN 的左侧和右侧：

>>> print(
...     select(user_table.c.name, address_table.c.email_address).join_from(
...         user_table, address_table
...     )
... )
SELECT  user_account.name,  address.email_address
FROM  user_account  JOIN  address  ON  user_account.id  =  address.user_id

另一个是 Select.join() 方法，它表示 JOIN 的右侧，左侧被推断：

>>> print(select(user_table.c.name, address_table.c.email_address).join(address_table))
SELECT  user_account.name,  address.email_address
FROM  user_account  JOIN  address  ON  user_account.id  =  address.user_id

如果 FROM 子句没有按照我们想要的方式进行推断，我们还可以选择将元素明确添加到 FROM 子句中。我们使用 Select.select_from() 方法来实现这一点，如下所示，我们将 user_table 设为 FROM 子句中的第一个元素，然后使用 Select.join() 将 address_table 设为第二个元素：

>>> print(select(address_table.c.email_address).select_from(user_table).join(address_table))
SELECT  address.email_address
FROM  user_account  JOIN  address  ON  user_account.id  =  address.user_id

我们可能想要使用 Select.select_from() 的另一个示例是，如果我们的 columns 子句没有足够的信息提供 FROM 子句。例如，要从常见的 SQL 表达式 count(*) 中选择，我们使用名为 sqlalchemy.sql.expression.func 的 SQLAlchemy 元素来生成 SQL count() 函数：

>>> from sqlalchemy import func
>>> print(select(func.count("*")).select_from(user_table))
SELECT  count(:count_2)  AS  count_1
FROM  user_account

另请参阅

在连接中设置最左侧的 FROM 子句 - 在 ORM 查询指南 - 包含有关 Select.select_from() 和 Select.join() 互动的附加示例和注释。

设置 ON 子句

前面 JOIN 的示例说明了 Select 结构可以在两个表之间进行 JOIN，并自动生成 ON 子句。这在这些示例中发生，因为 user_table 和 address_table Table 对象包含单个 ForeignKeyConstraint 定义，用于形成此 ON 子句。

如果连接的左右目标没有这样的约束，或者存在多个约束，则需要直接指定 ON 子句。 Select.join() 和 Select.join_from() 都接受用于 ON 子句的额外参数，其使用与我们在 WHERE 子句 中看到的 SQL 表达式机制相同：

>>> print(
...     select(address_table.c.email_address)
...     .select_from(user_table)
...     .join(address_table, user_table.c.id == address_table.c.user_id)
... )
SELECT  address.email_address
FROM  user_account  JOIN  address  ON  user_account.id  =  address.user_id

ORM 提示 - 在使用 ORM 实体时，当使用 relationship() 构造时，还有另一种生成 ON 子句的方式，就像在 声明映射类 中的前一节设置的映射一样。这是一个单独的主题，详细介绍在 使用关系连接。

OUTER 和 FULL join

Select.join() 和 Select.join_from() 方法都接受关键字参数 Select.join.isouter 和 Select.join.full，分别会渲染 LEFT OUTER JOIN 和 FULL OUTER JOIN：

>>> print(select(user_table).join(address_table, isouter=True))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account  LEFT  OUTER  JOIN  address  ON  user_account.id  =  address.user_id
>>> print(select(user_table).join(address_table, full=True))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account  FULL  OUTER  JOIN  address  ON  user_account.id  =  address.user_id

还有一个方法 Select.outerjoin()，它等同于使用 .join(..., isouter=True)。

提示

SQL 还有一个“RIGHT OUTER JOIN”。SQLAlchemy 不会直接呈现这个；相反，反转表的顺序并使用“LEFT OUTER JOIN”。 ## ORDER BY、GROUP BY、HAVING

SELECT SQL 语句包括一个称为 ORDER BY 的子句，用于以给定顺序返回所选行。

GROUP BY 子句的构造方式类似于 ORDER BY 子句，其目的是将所选行细分为特定的分组，从而可以对这些分组调用聚合函数。HAVING 子句通常与 GROUP BY 一起使用，其形式与 WHERE 子句类似，只是它应用于分组内使用的聚合函数。

ORDER BY

ORDER BY 子句是根据 SQL 表达式构造的，通常基于Column或类似对象。Select.order_by()方法按位置接受一个或多个这些表达式：

>>> print(select(user_table).order_by(user_table.c.name))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account  ORDER  BY  user_account.name

升序/降序可以从ColumnElement.asc()和ColumnElement.desc()修饰符中获得，这些修饰符也存在于 ORM 绑定的属性中：

>>> print(select(User).order_by(User.fullname.desc()))
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account  ORDER  BY  user_account.fullname  DESC

上述语句将按照user_account.fullname列按降序排序的行。### 带有 GROUP BY / HAVING 的聚合函数

在 SQL 中，聚合函数允许跨多行的列表达式聚合在一起，以产生单个结果。示例包括计数、计算平均值，以及在一组值中定位最大值或最小值。

SQLAlchemy 以一种开放式的方式提供 SQL 函数，使用一个名为func的命名空间。这是一个特殊的构造对象，当给定特定 SQL 函数的名称时，它将创建Function的新实例，该函数可以有任何名称，以及零个或多个要传递给函数的参数，就像在所有其他情况下一样，都是 SQL 表达式构造。例如，要针对user_account.id列渲染 SQL COUNT()函数，我们调用count()名称：

>>> from sqlalchemy import func
>>> count_fn = func.count(user_table.c.id)
>>> print(count_fn)
count(user_account.id)

SQL 函数在本教程的后面部分详细描述，链接在使用 SQL 函数。

在 SQL 中使用聚合函数时，GROUP BY 子句是必不可少的，因为它允许将行分成组，其中聚合函数将分别应用于每个组。在 SELECT  语句的 COLUMNS 子句中请求非聚合列时，SQL 要求这些列都受到 GROUP BY  子句的约束，直接或间接地基于主键关联。然后，HAVING 子句类似于 WHERE 子句，不同之处在于它根据聚合值而不是直接行内容来过滤行。

SQLAlchemy 提供了这两个子句的功能，使用 Select.group_by() 和 Select.having() 方法。下面我们示例选择用户名称字段以及地址计数，对于那些拥有多个地址的用户：

>>> with engine.connect() as conn:
...     result = conn.execute(
...         select(User.name, func.count(Address.id).label("count"))
...         .join(Address)
...         .group_by(User.name)
...         .having(func.count(Address.id) > 1)
...     )
...     print(result.all())
BEGIN  (implicit)
SELECT  user_account.name,  count(address.id)  AS  count
FROM  user_account  JOIN  address  ON  user_account.id  =  address.user_id  GROUP  BY  user_account.name
HAVING  count(address.id)  >  ?
[...]  (1,)
[('sandy', 2)]
ROLLBACK 
```### 按标签排序或分组
特别重要的一项技术，在某些数据库后端特别是，是有能力按照已在列子句中已经声明的表达式进行 ORDER BY 或 GROUP BY，而无需在 ORDER BY 或 GROUP BY 子句中重新声明表达式，而是使用 COLUMNS 子句中的列名或标记名。可以通过将名称的字符串文本传递给 `Select.order_by()` 或 `Select.group_by()` 方法来使用这种形式。传递的文本**不会直接渲染**；而是在列子句中给定的表达式名称，并在上下文中呈现为该表达式名称，如果找不到匹配项，则会引发错误。这种形式也可以使用一元修饰符 `asc()` 和 `desc()`：
```py
>>> from sqlalchemy import func, desc
>>> stmt = (
...     select(Address.user_id, func.count(Address.id).label("num_addresses"))
...     .group_by("user_id")
...     .order_by("user_id", desc("num_addresses"))
... )
>>> print(stmt)
SELECT  address.user_id,  count(address.id)  AS  num_addresses
FROM  address  GROUP  BY  address.user_id  ORDER  BY  address.user_id,  num_addresses  DESC 
```## 使用别名
现在我们正在从多个表中进行选择并使用连接，我们很快就会遇到需要在语句的 FROM 子句中多次引用同一张表的情况。我们使用 SQL **别名** 来实现这一点，这是一种为表或子查询提供替代名称的语法，可以在语句中引用它。
在 SQLAlchemy 表达语言中，这些“名称”代替了 `FromClause` 对象，被称为 `Alias` 构造，在 Core 中使用 `FromClause.alias()` 方法构造。一个 `Alias` 构造就像一个 `Table` 构造一样，它也有一个在 `Alias.c` 集合中的 `Column` 对象的命名空间。例如下面的 SELECT 语句返回所有唯一的用户名对：
```py
>>> user_alias_1 = user_table.alias()
>>> user_alias_2 = user_table.alias()
>>> print(
...     select(user_alias_1.c.name, user_alias_2.c.name).join_from(
...         user_alias_1, user_alias_2, user_alias_1.c.id > user_alias_2.c.id
...     )
... )
SELECT  user_account_1.name,  user_account_2.name  AS  name_1
FROM  user_account  AS  user_account_1
JOIN  user_account  AS  user_account_2  ON  user_account_1.id  >  user_account_2.id

ORM 实体别名

ORM 中与 FromClause.alias() 方法对应的方法是 ORM aliased() 函数，可应用于实体，如 User 和 Address。这将在内部生成一个 Alias 对象，针对原始映射的 Table 对象，同时保持 ORM 功能。下面的 SELECT 从 User 实体中选择包含两个特定电子邮件地址的所有对象：

>>> from sqlalchemy.orm import aliased
>>> address_alias_1 = aliased(Address)
>>> address_alias_2 = aliased(Address)
>>> print(
...     select(User)
...     .join_from(User, address_alias_1)
...     .where(address_alias_1.email_address == "patrick@aol.com")
...     .join_from(User, address_alias_2)
...     .where(address_alias_2.email_address == "patrick@gmail.com")
... )
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
JOIN  address  AS  address_1  ON  user_account.id  =  address_1.user_id
JOIN  address  AS  address_2  ON  user_account.id  =  address_2.user_id
WHERE  address_1.email_address  =  :email_address_1
AND  address_2.email_address  =  :email_address_2

Tip

如在 设置 ON 子句 中提到的，ORM 提供了使用 relationship() 结构进行连接的另一种方式。上述使用别名的示例是使用 relationship() 在 使用关系在别名目标之间进行连接 中演示的。## 子查询和 CTE

SQL 中的子查询是在括号内呈现并放置在封闭语句上下文中的 SELECT 语句，通常是 SELECT 语句，但不一定。

本节将介绍所谓的“非标量”子查询，通常放置在封闭 SELECT 的 FROM 子句中。我们还将介绍通用表达式（Common Table Expression，CTE），它与子查询的使用方式类似，但包含其他功能。

SQLAlchemy 使用 Subquery 对象表示子查询，使用 CTE 表示 CTE，通常分别从 Select.subquery() 和 Select.cte() 方法获取。这两个对象都可以作为较大的 select() 结构中的 FROM 元素使用。

我们可以构造一个 Subquery ，将从 address 表中选择行的聚合计数（聚合函数和 GROUP BY 在 具有 GROUP BY / HAVING 的聚合函数 中已介绍）：

>>> subq = (
...     select(func.count(address_table.c.id).label("count"), address_table.c.user_id)
...     .group_by(address_table.c.user_id)
...     .subquery()
... )

单独将子查询字符串化，而不将其嵌入到另一个Select或其他语句中，会生成不带任何封闭括号的普通 SELECT 语句：

>>> print(subq)
SELECT  count(address.id)  AS  count,  address.user_id
FROM  address  GROUP  BY  address.user_id

Subquery对象的行为类似于任何其他 FROM 对象，例如Table，特别是它包含一个Subquery.c列的命名空间，该命名空间选择它。 我们可以使用此命名空间来引用user_id列以及我们的自定义标记的count表达式：

>>> print(select(subq.c.user_id, subq.c.count))
SELECT  anon_1.user_id,  anon_1.count
FROM  (SELECT  count(address.id)  AS  count,  address.user_id  AS  user_id
FROM  address  GROUP  BY  address.user_id)  AS  anon_1

通过包含在subq对象中的一系列行的选择，我们可以将该对象应用于一个更大的Select，将数据连接到user_account表：

>>> stmt = select(user_table.c.name, user_table.c.fullname, subq.c.count).join_from(
...     user_table, subq
... )
>>> print(stmt)
SELECT  user_account.name,  user_account.fullname,  anon_1.count
FROM  user_account  JOIN  (SELECT  count(address.id)  AS  count,  address.user_id  AS  user_id
FROM  address  GROUP  BY  address.user_id)  AS  anon_1  ON  user_account.id  =  anon_1.user_id

为了从user_account连接到address，我们利用了Select.join_from()方法。 正如之前所说明的，此连接的 ON 子句再次基于外键约束推断。 即使 SQL 子查询本身没有任何约束，SQLAlchemy 也可以根据列上表示的约束来操作列，从而确定subq.c.user_id列派生自表达外键关系的address_table.c.user_id列，该列又表达了与user_table.c.id列的外键关系，然后用于生成 ON 子句。

公共表达式（CTEs）

在 SQLAlchemy 中使用CTE结构的用法与使用Subquery结构几乎相同。 通过将Select.subquery()方法的调用更改为使用Select.cte()而不是，我们可以像以前一样使用结果对象作为 FROM 元素，但是渲染的 SQL 是非常不同的常用表达式语法：

>>> subq = (
...     select(func.count(address_table.c.id).label("count"), address_table.c.user_id)
...     .group_by(address_table.c.user_id)
...     .cte()
... )
>>> stmt = select(user_table.c.name, user_table.c.fullname, subq.c.count).join_from(
...     user_table, subq
... )
>>> print(stmt)
WITH  anon_1  AS
(SELECT  count(address.id)  AS  count,  address.user_id  AS  user_id
FROM  address  GROUP  BY  address.user_id)
  SELECT  user_account.name,  user_account.fullname,  anon_1.count
FROM  user_account  JOIN  anon_1  ON  user_account.id  =  anon_1.user_id

CTE结构还具有以“递归”方式使用的能力，并且在更复杂的情况下可以由 INSERT、UPDATE 或 DELETE 语句的 RETURNING 子句组成。 CTE的文档字符串包含有关这些附加模式的详细信息。

在这两种情况下，子查询和 CTE 在 SQL 层面上都被命名为“匿名”名称。在 Python 代码中，我们根本不需要提供这些名称。当渲染时，Subquery 或 CTE 实例的对象标识作为对象的语法标识。可以通过将其作为 Select.subquery() 或 Select.cte() 方法的第一个参数来提供在 SQL 中呈现的名称。

参见

Select.subquery() - 关于子查询的进一步细节

Select.cte() - 包括如何使用 RECURSIVE 以及面向 DML 的 CTE 的示例

ORM 实体子查询/CTEs

在 ORM 中，aliased() 构造可用于将 ORM 实体（例如我们的 User 或 Address 类）与表示行来源的任何 FromClause 概念相关联。前一节 ORM 实体别名 演示了如何使用 aliased() 将映射类与其映射的 Table 的 Alias 相关联。在这里，我们演示了 aliased() 对一个 Subquery 以及对一个由 Select 构造生成的 CTE 执行相同操作，最终从相同的映射 Table 派生。

下面是将aliased() 应用到Subquery 构造的示例，以便从其行中提取 ORM 实体。结果显示了一系列User和Address对象，其中每个Address对象的数据最终来自于针对address表的子查询，而不是直接来自该表：

>>> subq = select(Address).where(~Address.email_address.like("%@aol.com")).subquery()
>>> address_subq = aliased(Address, subq)
>>> stmt = (
...     select(User, address_subq)
...     .join_from(User, address_subq)
...     .order_by(User.id, address_subq.id)
... )
>>> with Session(engine) as session:
...     for user, address in session.execute(stmt):
...         print(f"{user} {address}")
BEGIN  (implicit)
SELECT  user_account.id,  user_account.name,  user_account.fullname,
anon_1.id  AS  id_1,  anon_1.email_address,  anon_1.user_id
FROM  user_account  JOIN
(SELECT  address.id  AS  id,  address.email_address  AS  email_address,  address.user_id  AS  user_id
FROM  address
WHERE  address.email_address  NOT  LIKE  ?)  AS  anon_1  ON  user_account.id  =  anon_1.user_id
ORDER  BY  user_account.id,  anon_1.id
[...]  ('%@aol.com',)
User(id=1, name='spongebob', fullname='Spongebob Squarepants') Address(id=1, email_address='spongebob@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=2, email_address='sandy@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=3, email_address='sandy@squirrelpower.org')
ROLLBACK

下面是另一个例子，与之前的例子完全相同，只是它使用了CTE 构造：

>>> cte_obj = select(Address).where(~Address.email_address.like("%@aol.com")).cte()
>>> address_cte = aliased(Address, cte_obj)
>>> stmt = (
...     select(User, address_cte)
...     .join_from(User, address_cte)
...     .order_by(User.id, address_cte.id)
... )
>>> with Session(engine) as session:
...     for user, address in session.execute(stmt):
...         print(f"{user} {address}")
BEGIN  (implicit)
WITH  anon_1  AS
(SELECT  address.id  AS  id,  address.email_address  AS  email_address,  address.user_id  AS  user_id
FROM  address
WHERE  address.email_address  NOT  LIKE  ?)
SELECT  user_account.id,  user_account.name,  user_account.fullname,
anon_1.id  AS  id_1,  anon_1.email_address,  anon_1.user_id
FROM  user_account
JOIN  anon_1  ON  user_account.id  =  anon_1.user_id
ORDER  BY  user_account.id,  anon_1.id
[...]  ('%@aol.com',)
User(id=1, name='spongebob', fullname='Spongebob Squarepants') Address(id=1, email_address='spongebob@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=2, email_address='sandy@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=3, email_address='sandy@squirrelpower.org')
ROLLBACK

另请参阅

从子查询中选择实体 - 在 ORM 查询指南 ## 标量和相关子查询

标量子查询是一个返回零行或一行且一列的子查询。然后，该子查询在包含 SELECT 语句的 COLUMNS 或 WHERE 子句中使用，并且与常规子查询不同之处在于它不在 FROM 子句中使用。相关子查询 是指在包含 SELECT 语句中引用表的标量子查询。

SQLAlchemy 使用ScalarSelect 构造来表示标量子查询，该构造是ColumnElement 表达式层次结构的一部分，与常规子查询不同，常规子查询由Subquery 构造表示，该构造位于FromClause 层次结构中。

标量子查询通常与聚合函数一起使用，但不一定要这样，之前在带有 GROUP BY / HAVING 的聚合函数中介绍过。标量子查询通过显式使用Select.scalar_subquery() 方法来指示。下面是一个示例，其默认的字符串形式在单独字符串化时呈现为从两个表中选择的普通 SELECT 语句：

>>> subq = (
...     select(func.count(address_table.c.id))
...     .where(user_table.c.id == address_table.c.user_id)
...     .scalar_subquery()
... )
>>> print(subq)
(SELECT  count(address.id)  AS  count_1
FROM  address,  user_account
WHERE  user_account.id  =  address.user_id)

上述subq对象现在位于ColumnElement SQL 表达式层次结构中，因此它可以像任何其他列表达式一样使用：

>>> print(subq == 5)
(SELECT  count(address.id)  AS  count_1
FROM  address,  user_account
WHERE  user_account.id  =  address.user_id)  =  :param_1

虽然标量子查询本身在自身字符串化时在其 FROM 子句中呈现了user_account和address，但是，当将其嵌入到处理user_account表的封闭select()构造中时，user_account表会自动相关联，这意味着它不会在子查询的 FROM 子句中呈现：

>>> stmt = select(user_table.c.name, subq.label("address_count"))
>>> print(stmt)
SELECT  user_account.name,  (SELECT  count(address.id)  AS  count_1
FROM  address
WHERE  user_account.id  =  address.user_id)  AS  address_count
FROM  user_account

简单的相关子查询通常会执行所需的正确操作。但是，在相关性不明确的情况下，SQLAlchemy 将通知我们需要更清晰：

>>> stmt = (
...     select(
...         user_table.c.name,
...         address_table.c.email_address,
...         subq.label("address_count"),
...     )
...     .join_from(user_table, address_table)
...     .order_by(user_table.c.id, address_table.c.id)
... )
>>> print(stmt)
Traceback (most recent call last):
...
InvalidRequestError: Select statement '<... Select object at ...>' returned
no FROM clauses due to auto-correlation; specify correlate(<tables>) to
control correlation manually.

要指定user_table是我们要关联的表，我们使用ScalarSelect.correlate()或ScalarSelect.correlate_except()方法来指定：

>>> subq = (
...     select(func.count(address_table.c.id))
...     .where(user_table.c.id == address_table.c.user_id)
...     .scalar_subquery()
...     .correlate(user_table)
... )

然后，该语句可以像处理其他列一样返回此列的数据：

>>> with engine.connect() as conn:
...     result = conn.execute(
...         select(
...             user_table.c.name,
...             address_table.c.email_address,
...             subq.label("address_count"),
...         )
...         .join_from(user_table, address_table)
...         .order_by(user_table.c.id, address_table.c.id)
...     )
...     print(result.all())
BEGIN  (implicit)
SELECT  user_account.name,  address.email_address,  (SELECT  count(address.id)  AS  count_1
FROM  address
WHERE  user_account.id  =  address.user_id)  AS  address_count
FROM  user_account  JOIN  address  ON  user_account.id  =  address.user_id  ORDER  BY  user_account.id,  address.id
[...]  ()
[('spongebob', 'spongebob@sqlalchemy.org', 1), ('sandy', 'sandy@sqlalchemy.org', 2),
 ('sandy', 'sandy@squirrelpower.org', 2)]
ROLLBACK


SqlAlchemy 2.0 中文文档（二）（2）https://developer.aliyun.com/article/1560324

文章标签：
Python
SQL
测试技术
布客飞龙
目录
相关文章
布客飞龙
|
5月前
|
SQL 存储 数据库
SqlAlchemy 2.0 中文文档（一）（4）
SqlAlchemy 2.0 中文文档（一）
布客飞龙
86 1
布客飞龙
|
5月前
|
SQL API 数据库
SqlAlchemy 2.0 中文文档（一）（5）
SqlAlchemy 2.0 中文文档（一）
布客飞龙
129 1
布客飞龙
|
5月前
|
SQL 存储 数据库
SqlAlchemy 2.0 中文文档（一）（3）
SqlAlchemy 2.0 中文文档（一）
布客飞龙
94 1
布客飞龙
|
5月前
|
SQL 缓存 关系型数据库
SqlAlchemy 2.0 中文文档（三）（2）
SqlAlchemy 2.0 中文文档（三）
布客飞龙
39 1
布客飞龙
|
5月前
|
SQL API 数据库
SqlAlchemy 2.0 中文文档（四）（1）
SqlAlchemy 2.0 中文文档（四）
布客飞龙
44 1
布客飞龙
|
5月前
|
SQL 存储 API
SqlAlchemy 2.0 中文文档（十）（5）
SqlAlchemy 2.0 中文文档（十）
布客飞龙
35 1
布客飞龙
|
5月前
|
SQL 数据库 Python
SqlAlchemy 2.0 中文文档（十）（3）
SqlAlchemy 2.0 中文文档（十）
布客飞龙
39 1
布客飞龙
|
5月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档（五）（3）
SqlAlchemy 2.0 中文文档（五）
布客飞龙
52 0
布客飞龙
|
5月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档（五）（1）
SqlAlchemy 2.0 中文文档（五）
布客飞龙
49 0
布客飞龙
|
5月前
|
SQL 关系型数据库 数据库
SqlAlchemy 2.0 中文文档（十）（2）
SqlAlchemy 2.0 中文文档（十）
布客飞龙
33 0