SqlAlchemy 2.0 中文文档（二）（4）-阿里云开发者社区

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

ORM 实体子查询/CTEs

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

以下是将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 查询指南

公共表达式（CTEs）

使用CTE构造在 SQLAlchemy 中的使用方式与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() - CTE 的示例，包括如何使用 RECURSIVE 以及面向 DML 的 CTE 的示例

ORM 实体子查询/CTEs

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

下面是将 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

LATERAL 关联

LATERAL 关联是 SQL 关联的一个特殊子类别，允许可选择的单元引用同一 FROM 子句内的另一个可选择单元。这是一个极其特殊的用例，虽然它是 SQL 标准的一部分，但目前只知道最近的 PostgreSQL 版本支持它。

通常，如果 SELECT 语句在其 FROM 子句中引用 table1 JOIN (SELECT ...) AS subquery，右侧的子查询可能无法引用左侧的“table1”表达式；关联只能引用完全包含此 SELECT 的另一个 SELECT 的表。LATERAL 关键字允许我们改变这种行为，并允许来自右侧 JOIN 的关联。

SQLAlchemy 支持使用 Select.lateral() 方法来实现此功能，该方法创建一个称为 Lateral 的对象。Lateral 与 Subquery 和 Alias 属于同一家族，但在将构造添加到包含 SELECT 的 FROM 子句时还包括关联行为。以下示例说明了使用 LATERAL 的 SQL 查询，选择了在前一节中讨论过的“用户账户/电子邮件地址计数”数据：

>>> subq = (
...     select(
...         func.count(address_table.c.id).label("address_count"),
...         address_table.c.email_address,
...         address_table.c.user_id,
...     )
...     .where(user_table.c.id == address_table.c.user_id)
...     .lateral()
... )
>>> stmt = (
...     select(user_table.c.name, subq.c.address_count, subq.c.email_address)
...     .join_from(user_table, subq)
...     .order_by(user_table.c.id, subq.c.email_address)
... )
>>> print(stmt)
SELECT  user_account.name,  anon_1.address_count,  anon_1.email_address
FROM  user_account
JOIN  LATERAL  (SELECT  count(address.id)  AS  address_count,
address.email_address  AS  email_address,  address.user_id  AS  user_id
FROM  address
WHERE  user_account.id  =  address.user_id)  AS  anon_1
ON  user_account.id  =  anon_1.user_id
ORDER  BY  user_account.id,  anon_1.email_address

在上面的例子中，JOIN 的右侧是一个与左侧连接的 user_account 表的子查询。

使用 Select.lateral() 时，Select.correlate() 和 Select.correlate_except() 方法的行为也适用于 Lateral 构造。

另请参阅

Lateral

Select.lateral() ### LATERAL 关联

横向关联是 SQL 关联的一个特殊子类别，它允许一个可选择的单元在单个 FROM 子句内引用另一个可选择的单元。这是一个非常特殊的用例，虽然是 SQL 标准的一部分，但只有最近版本的 PostgreSQL 已知支持。

通常，如果一个 SELECT 语句在其 FROM 子句中引用了table1 JOIN (SELECT ...) AS subquery，则右侧的子查询可能不会引用左侧的“table1”表达式；关联可能仅引用完全包含此 SELECT 的另一个 SELECT 的表。LATERAL 关键字允许我们改变这种行为，允许从右侧 JOIN 进行关联。

SQLAlchemy 通过Select.lateral()方法支持此功能，该方法创建一个称为横向关联的对象。 横向关联与子查询和别名属于同一系列，但是当将构造添加到包围 SELECT 的 FROM 子句时，还包括关联行为。以下示例说明了使用 LATERAL 的 SQL 查询，选择了前一节中讨论的“用户帐户/电子邮件地址计数”数据：

>>> subq = (
...     select(
...         func.count(address_table.c.id).label("address_count"),
...         address_table.c.email_address,
...         address_table.c.user_id,
...     )
...     .where(user_table.c.id == address_table.c.user_id)
...     .lateral()
... )
>>> stmt = (
...     select(user_table.c.name, subq.c.address_count, subq.c.email_address)
...     .join_from(user_table, subq)
...     .order_by(user_table.c.id, subq.c.email_address)
... )
>>> print(stmt)
SELECT  user_account.name,  anon_1.address_count,  anon_1.email_address
FROM  user_account
JOIN  LATERAL  (SELECT  count(address.id)  AS  address_count,
address.email_address  AS  email_address,  address.user_id  AS  user_id
FROM  address
WHERE  user_account.id  =  address.user_id)  AS  anon_1
ON  user_account.id  =  anon_1.user_id
ORDER  BY  user_account.id,  anon_1.email_address

上述，JOIN 的右侧是一个子查询，它与 JOIN 左侧的user_account表相关联。

使用Select.lateral()时，Select.correlate()和Select.correlate_except()方法的行为也适用于横向关联构造。

另请参见

横向关联

Select.lateral()

UNION、UNION ALL 和其他集合操作

在 SQL 中，SELECT 语句可以使用 UNION 或 UNION ALL SQL 操作合并在一起，该操作生成由一个或多个语句一起生成的所有行的集合。还可以执行其他集合操作，如 INTERSECT [ALL]和 EXCEPT [ALL]。

SQLAlchemy 的Select构造支持使用诸如union()、intersect()和except_()之类的函数进行这种性质的组合，以及“all”对应项union_all()、intersect_all()和except_all()。这些函数都接受任意数量的子可选择项，通常是Select构造，但也可以是现有的组合。

由这些函数生成的构造是CompoundSelect，其使用方式与Select构造相同，只是它的方法较少。例如，由union_all()产生的CompoundSelect可以直接通过Connection.execute()调用：

>>> from sqlalchemy import union_all
>>> stmt1 = select(user_table).where(user_table.c.name == "sandy")
>>> stmt2 = select(user_table).where(user_table.c.name == "spongebob")
>>> u = union_all(stmt1, stmt2)
>>> with engine.connect() as conn:
...     result = conn.execute(u)
...     print(result.all())
BEGIN  (implicit)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?
UNION  ALL  SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?
[generated  in  ...]  ('sandy',  'spongebob')
[(2, 'sandy', 'Sandy Cheeks'), (1, 'spongebob', 'Spongebob Squarepants')]
ROLLBACK

要将CompoundSelect用作子查询，就像Select一样，它提供了一个SelectBase.subquery()方法，它将生成一个带有FromClause.c集合的Subquery对象，可以在封闭的select()中引用：

>>> u_subq = u.subquery()
>>> stmt = (
...     select(u_subq.c.name, address_table.c.email_address)
...     .join_from(address_table, u_subq)
...     .order_by(u_subq.c.name, address_table.c.email_address)
... )
>>> with engine.connect() as conn:
...     result = conn.execute(stmt)
...     print(result.all())
BEGIN  (implicit)
SELECT  anon_1.name,  address.email_address
FROM  address  JOIN
  (SELECT  user_account.id  AS  id,  user_account.name  AS  name,  user_account.fullname  AS  fullname
  FROM  user_account
  WHERE  user_account.name  =  ?
UNION  ALL
  SELECT  user_account.id  AS  id,  user_account.name  AS  name,  user_account.fullname  AS  fullname
  FROM  user_account
  WHERE  user_account.name  =  ?)
AS  anon_1  ON  anon_1.id  =  address.user_id
ORDER  BY  anon_1.name,  address.email_address
[generated  in  ...]  ('sandy',  'spongebob')
[('sandy', 'sandy@sqlalchemy.org'), ('sandy', 'sandy@squirrelpower.org'), ('spongebob', 'spongebob@sqlalchemy.org')]
ROLLBACK

从联合中选择 ORM 实体

前面的示例说明了如何构造一个 UNION，给定两个Table对象，然后返回数据库行。如果我们想要使用 UNION 或其他集合操作来选择行，然后将其作为 ORM 对象接收，有两种方法可以使用。在这两种情况下，我们首先构造一个表示我们想要执行的 SELECT / UNION / 等语句的select()或CompoundSelect对象；这个语句应该针对目标 ORM 实体或它们的底层映射的Table对象组成：

>>> stmt1 = select(User).where(User.name == "sandy")
>>> stmt2 = select(User).where(User.name == "spongebob")
>>> u = union_all(stmt1, stmt2)

对于一个简单的 SELECT，带有 UNION，它尚未嵌套在子查询内部，通常可以通过使用Select.from_statement()方法在 ORM 对象获取上下文中使用。通过这种方法，UNION 语句代表整个查询；在使用Select.from_statement()之后，不能添加额外的条件：

>>> orm_stmt = select(User).from_statement(u)
>>> with Session(engine) as session:
...     for obj in session.execute(orm_stmt).scalars():
...         print(obj)
BEGIN  (implicit)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?  UNION  ALL  SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?
[generated  in  ...]  ('sandy',  'spongebob')
User(id=2, name='sandy', fullname='Sandy Cheeks')
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
ROLLBACK

要以更灵活的方式将 UNION 或其他与实体相关的构造用作实体相关组件，可以使用CompoundSelect构造，使用CompoundSelect.subquery()将其组织成子查询，然后使用aliased()函数将其链接到 ORM 对象。这与在 ORM 实体子查询/CTEs 中介绍的方式相同，首先创建我们所需实体的临时“映射”，然后从该新实体选择，就像它是任何其他映射类一样。在下面的示例中，我们能够添加额外的条件，例如在 UNION 之外的 ORDER BY，因为我们可以过滤或按子查询导出的列排序：

>>> user_alias = aliased(User, u.subquery())
>>> orm_stmt = select(user_alias).order_by(user_alias.id)
>>> with Session(engine) as session:
...     for obj in session.execute(orm_stmt).scalars():
...         print(obj)
BEGIN  (implicit)
SELECT  anon_1.id,  anon_1.name,  anon_1.fullname
FROM  (SELECT  user_account.id  AS  id,  user_account.name  AS  name,  user_account.fullname  AS  fullname
FROM  user_account
WHERE  user_account.name  =  ?  UNION  ALL  SELECT  user_account.id  AS  id,  user_account.name  AS  name,  user_account.fullname  AS  fullname
FROM  user_account
WHERE  user_account.name  =  ?)  AS  anon_1  ORDER  BY  anon_1.id
[generated  in  ...]  ('sandy',  'spongebob')
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
ROLLBACK

另请参阅

从 UNIONs 和其他集合操作中选择实体 - 在 ORM 查询指南中的 ORM 实体从联合中选择

前面的示例说明了如何在给定两个Table对象的情况下构造一个 UNION，然后返回数据库行。如果我们想要使用 UNION 或其他集合操作来选择行，然后将其作为 ORM 对象接收，有两种方法可以使用。在这两种情况下，我们首先构造一个select()或CompoundSelect对象，该对象表示我们要执行的 SELECT / UNION /等语句;此语句应针对目标 ORM 实体或其底层映射的Table对象组成:

>>> stmt1 = select(User).where(User.name == "sandy")
>>> stmt2 = select(User).where(User.name == "spongebob")
>>> u = union_all(stmt1, stmt2)

对于不在子查询内部的简单 SELECT 与 UNION，通常可以使用Select.from_statement()方法在 ORM 对象获取上下文中使用。通过这种方法，UNION 语句表示整个查询;在使用Select.from_statement()之后，不能添加额外的条件:

>>> orm_stmt = select(User).from_statement(u)
>>> with Session(engine) as session:
...     for obj in session.execute(orm_stmt).scalars():
...         print(obj)
BEGIN  (implicit)
SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?  UNION  ALL  SELECT  user_account.id,  user_account.name,  user_account.fullname
FROM  user_account
WHERE  user_account.name  =  ?
[generated  in  ...]  ('sandy',  'spongebob')
User(id=2, name='sandy', fullname='Sandy Cheeks')
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
ROLLBACK

要以更灵活的方式将 UNION 或其他集相关构造用作实体相关组件，可以使用CompoundSelect构造将其组织到子查询中，然后使用CompoundSelect.subquery()将其链接到 ORM 对象，然后使用aliased()函数。这与 ORM 实体子查询/ CTEs 中介绍的方式相同，首先创建我们所需实体到子查询的临时“映射”，然后从该新实体中选择，就像它是任何其他映射类一样。在下面的示例中，我们能够添加额外的条件，例如在 UNION 本身之外进行 ORDER BY，因为我们可以通过子查询导出的列进行过滤或排序:

>>> user_alias = aliased(User, u.subquery())
>>> orm_stmt = select(user_alias).order_by(user_alias.id)
>>> with Session(engine) as session:
...     for obj in session.execute(orm_stmt).scalars():
...         print(obj)
BEGIN  (implicit)
SELECT  anon_1.id,  anon_1.name,  anon_1.fullname
FROM  (SELECT  user_account.id  AS  id,  user_account.name  AS  name,  user_account.fullname  AS  fullname
FROM  user_account
WHERE  user_account.name  =  ?  UNION  ALL  SELECT  user_account.id  AS  id,  user_account.name  AS  name,  user_account.fullname  AS  fullname
FROM  user_account
WHERE  user_account.name  =  ?)  AS  anon_1  ORDER  BY  anon_1.id
[generated  in  ...]  ('sandy',  'spongebob')
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
ROLLBACK

另请参阅

从 UNION 和其他集合操作中选择实体 - 在 ORM 查询指南中

EXISTS 子查询

SQL EXISTS 关键字是一个与标量子查询一起使用的运算符，根据 SELECT 语句是否返回行来返回布尔值 true 或 false。SQLAlchemy 包含一个名为Exists的ScalarSelect对象的变体，它将生成一个 EXISTS 子查询，并且最方便的方式是使用SelectBase.exists()方法生成。下面我们生成一个 EXISTS，以便我们可以返回user_account行，其中address有多于一个相关行。

>>> subq = (
...     select(func.count(address_table.c.id))
...     .where(user_table.c.id == address_table.c.user_id)
...     .group_by(address_table.c.user_id)
...     .having(func.count(address_table.c.id) > 1)
... ).exists()
>>> with engine.connect() as conn:
...     result = conn.execute(select(user_table.c.name).where(subq))
...     print(result.all())
BEGIN  (implicit)
SELECT  user_account.name
FROM  user_account
WHERE  EXISTS  (SELECT  count(address.id)  AS  count_1
FROM  address
WHERE  user_account.id  =  address.user_id  GROUP  BY  address.user_id
HAVING  count(address.id)  >  ?)
[...]  (1,)
[('sandy',)]
ROLLBACK

EXISTS 结构更常用于否定，例如 NOT EXISTS，因为它提供了一种 SQL 效率高的方式来定位一个相关表没有行的行。下面我们选择没有电子邮件地址的用户名称；注意在第二个 WHERE 子句中使用的二进制否定运算符（~）：

>>> subq = (
...     select(address_table.c.id).where(user_table.c.id == address_table.c.user_id)
... ).exists()
>>> with engine.connect() as conn:
...     result = conn.execute(select(user_table.c.name).where(~subq))
...     print(result.all())
BEGIN  (implicit)
SELECT  user_account.name
FROM  user_account
WHERE  NOT  (EXISTS  (SELECT  address.id
FROM  address
WHERE  user_account.id  =  address.user_id))
[...]  ()
[('patrick',)]
ROLLBACK

使用 SQL 函数

此部分较早前在带有 GROUP BY / HAVING 的聚合函数中首次介绍，func对象用作创建新的Function对象的工厂，在像select()这样的构造中使用时，会产生一个 SQL 函数显示，通常包含一个名称、一些括号（尽管不总是），以及可能的一些参数。典型 SQL 函数的示例包括：

count()函数，一个聚合函数，用于计算返回的行数：

>>> print(select(func.count()).select_from(user_table))
SELECT  count(*)  AS  count_1
FROM  user_account

lower()函数，一个字符串函数，用于将字符串转换为小写：

>>> print(select(func.lower("A String With Much UPPERCASE")))
SELECT  lower(:lower_2)  AS  lower_1

now()函数，提供当前日期和时间；由于这是一个常见的函数，SQLAlchemy 知道如何为每个后端呈现不同的结果，在 SQLite 中使用 CURRENT_TIMESTAMP 函数：

>>> stmt = select(func.now())
>>> with engine.connect() as conn:
...     result = conn.execute(stmt)
...     print(result.all())
BEGIN  (implicit)
SELECT  CURRENT_TIMESTAMP  AS  now_1
[...]  ()
[(datetime.datetime(...),)]
ROLLBACK

由于大多数数据库后端包含数十甚至数百个不同的 SQL 函数，func尝试在接受的内容上尽可能宽松。从此命名空间中访问的任何名称都会自动被视为一个 SQL 函数，以一种通用的方式呈现：

>>> print(select(func.some_crazy_function(user_table.c.name, 17)))
SELECT  some_crazy_function(user_account.name,  :some_crazy_function_2)  AS  some_crazy_function_1
FROM  user_account

同时，一组相对较小但极其常见的 SQL 函数，比如 count、now、max、concat 等，包含了它们自己的预打包版本，这些版本提供了正确的类型信息以及在某些情况下特定于后端的 SQL 生成。下面的示例对比了 PostgreSQL 方言和 Oracle 方言中 now 函数的 SQL 生成：

>>> from sqlalchemy.dialects import postgresql
>>> print(select(func.now()).compile(dialect=postgresql.dialect()))
SELECT  now()  AS  now_1
>>> from sqlalchemy.dialects import oracle
>>> print(select(func.now()).compile(dialect=oracle.dialect()))
SELECT  CURRENT_TIMESTAMP  AS  now_1  FROM  DUAL

函数具有返回类型

由于函数是列表达式，它们还有描述生成的 SQL 表达式的数据类型的 SQL 数据类型。我们在这里将这些类型称为“SQL 返回类型”，指的是在数据库端 SQL 表达式的上下文中函数返回的 SQL 值类型，而不是 Python 函数的“返回类型”。

任何 SQL 函数的 SQL 返回类型都可以访问，通常用于调试目的，方法是引用 Function.type 属性：

>>> func.now().type
DateTime()

这些 SQL 返回类型在使用函数表达式时非常重要，特别是在更大表达式的上下文中；也就是说，当表达式的数据类型是类似 Integer 或 Numeric 这样的类型时，数学运算符将更有效，为了让 JSON 访问器正常工作，需要使用类似 JSON 这样的类型。某些类别的函数返回整行而不是列值，在需要引用特定列的情况下；这些函数被称为表值函数。

当执行语句并获取行时，函数的 SQL 返回类型也可能很重要，对于 SQLAlchemy 需要应用结果集处理的情况来说尤其如此。SQLite 上的日期相关函数是一个典型例子，其中 SQLAlchemy 的 DateTime 和相关数据类型在接收到结果行时起到将字符串值转换为 Python datetime() 对象的作用。

要将特定类型应用于我们正在创建的函数，我们可以使用 Function.type_ 参数进行传递；类型参数可以是 TypeEngine 类或实例。在下面的示例中，我们将 JSON 类传递给生成 PostgreSQL json_object() 函数，注意 SQL 返回类型将是 JSON 类型：

>>> from sqlalchemy import JSON
>>> function_expr = func.json_object('{a, 1, b, "def", c, 3.5}', type_=JSON)

通过使用具有 JSON 数据类型的 JSON 函数，SQL 表达式对象具有与 JSON 相关的功能，例如访问元素：

>>> stmt = select(function_expr["def"])
>>> print(stmt)
SELECT  json_object(:json_object_1)[:json_object_2]  AS  anon_1

内置函数具有预配置的返回类型

对于像count、max和min这样的常见聚合函数，以及一些非常少数的日期函数，比如now和字符串函数，SQL 返回类型会根据使用情况进行适当设置。max函数和类似的聚合过滤函数将根据给定的参数设置 SQL 返回类型：

>>> m1 = func.max(Column("some_int", Integer))
>>> m1.type
Integer()
>>> m2 = func.max(Column("some_str", String))
>>> m2.type
String()

日期和时间函数通常对应于由 DateTime、Date 或 Time 描述的 SQL 表达式：

>>> func.now().type
DateTime()
>>> func.current_date().type
Date()

已知的字符串函数，如 concat，将知道 SQL 表达式的类型将是 String：

>>> func.concat("x", "y").type
String()

但是，对于绝大多数 SQL 函数，SQLAlchemy 并没有在其极少量的已知函数列表中明确地提供它们。例如，虽然通常使用 SQL 函数 func.lower() 和 func.upper() 来转换字符串的大小写没有问题，但 SQLAlchemy 实际上并不知道这些函数，因此它们具有“null”SQL 返回类型：

>>> func.upper("lowercase").type
NullType()

对于像upper和lower这样的简单函数，问题通常不是很重要，因为字符串值可能从数据库接收而不需要在 SQLAlchemy 端进行任何特殊类型处理，而且 SQLAlchemy 的类型强制规则通常也可以正确猜测意图；例如，Python 的+运算符将根据表达式两侧的内容正确解释为字符串连接运算符：

>>> print(select(func.upper("lowercase") + " suffix"))
SELECT  upper(:upper_1)  ||  :upper_2  AS  anon_1

总的来说，Function.type_参数可能是必要的情况是：

函数不是 SQLAlchemy 内置函数；这可以通过创建函数并观察Function.type属性来证明，即：

>>> func.count().type
Integer()

vs.：

>>> func.json_object('{"a", "b"}').type
NullType()

需要函数感知表达式支持；这通常指的是与数据类型相关的特殊运算符，如JSON或ARRAY。
需要结果值处理，可能包括诸如DateTime、Boolean、Enum等类型，或者再次特殊数据类型，如JSON、ARRAY。

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

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

ORM 实体子查询/CTEs

公共表达式（CTEs）

ORM 实体子查询/CTEs

标量和关联子查询

LATERAL 关联

UNION、UNION ALL 和其他集合操作

从联合中选择 ORM 实体

EXISTS 子查询

使用 SQL 函数

函数具有返回类型

内置函数具有预配置的返回类型

热门文章

最新文章

相关课程

相关电子书