【Python百日进阶-Web开发-Peewee】Day259 - Peewee 公用表达式_wee weeyeezcte

8.17 检索行元组/字典/命名元组

有时您不需要创建模型实例的开销，而只是想迭代行数据而不需要提供的所有 API Model。为此，请使用：

• dicts()
• namedtuples()
• tuples()
• objects()– 接受使用行元组调用的任意构造函数。

stats = (Stat
         .select(Stat.url, fn.Count(Stat.url))
         .group_by(Stat.url)
         .tuples())

# iterate over a list of 2-tuples containing the url and count
for stat_url, stat_count in stats:
    print(stat_url, stat_count)
1
2
3
4
5
6
7
8

同样，您可以使用以下命令将游标中的行作为字典返回 dicts()：

stats = (Stat
         .select(Stat.url, fn.Count(Stat.url).alias('ct'))
         .group_by(Stat.url)
         .dicts())

# iterate over a list of 2-tuples containing the url and count
for stat in stats:
    print(stat['url'], stat['ct'])
1
2
3
4
5
6
7
8

8.18 Returning Clause

PostgresqlDatabase支持关于, 和查询的RETURNING子句。指定子句允许您遍历查询访问的行。UPDATEINSERTDELETERETURNING

默认情况下，执行不同查询时的返回值是：

• INSERT- 自动增加新插入行的主键值。当不使用自增主键时，Postgres 将返回新行的主键，但 SQLite 和 MySQL 不会。
• UPDATE- 修改的行数
• DELETE- 删除的行数
  当使用返回子句时，执行查询时的返回值将是一个可迭代的游标对象。

Postgresql 允许通过该RETURNING子句从查询插入或修改的行中返回数据。

例如，假设您有一个Update停用所有注册已过期的用户帐户。停用它们后，您希望向每个用户发送一封电子邮件，让他们知道他们的帐户已停用。您可以在带有子句的单个查询中执行此操作，而不是编写两个查询 aSELECT和 an ：UPDATEUPDATERETURNING

query = (User
         .update(is_active=False)
         .where(User.registration_expired == True)
         .returning(User))

# Send an email to every user that was deactivated.
for deactivate_user in query.execute():
    send_deactivation_email(deactivated_user.email)
1
2
3
4
5
6
7
8

该RETURNING子句也可用于Insert和 Delete。与 一起使用时INSERT，将返回新创建的行。与 一起使用时DELETE，将返回已删除的行。

该RETURNING子句的唯一限制是它只能包含查询FROM子句中列出的表中的列。要从特定表中选择所有列，您只需传入Model 类即可。

作为另一个示例，让我们添加一个用户并将其创建日期设置为服务器生成的当前时间戳。我们将在单个查询中创建和检索新用户的 ID、电子邮件和创建时间戳：

query = (User
         .insert(email='foo@bar.com', created=fn.now())
         .returning(User))  # Shorthand for all columns on User.

# When using RETURNING, execute() returns a cursor.
cursor = query.execute()

# Get the user object we just inserted and log the data:
user = cursor[0]
logger.info('Created user %s (id=%s) at %s', user.email, user.id, user.created)
1
2
3
4
5
6
7
8
9
10

默认情况下，游标将返回Model实例，但您可以指定不同的行类型：

data = [{'name': 'charlie'}, {'name': 'huey'}, {'name': 'mickey'}]
query = (User
         .insert_many(data)
         .returning(User.id, User.username)
         .dicts())

for new_user in query.execute():
    print('Added user "%s", id=%s' % (new_user['username'], new_user['id']))
1
2
3
4
5
6
7
8

与Select查询一样，您可以指定各种结果行类型。

8.19 公用表表达式

Peewee 支持在所有类型的查询中包含公用表表达式 (CTE)。CTE 可用于：

• 分解出一个公共子查询。
• 按 CTE 结果集中派生的列进行分组或过滤。
• 编写递归查询。
  要将Select查询声明为 CTE，请使用 cte()将查询包装在CTE 对象中的方法。要指示 aCTE应作为查询的一部分包含，请使用该Query.with_cte()方法，传递 CTE 对象列表。

8.19.1 简单示例

例如，假设我们有一些由键和浮点值组成的数据点。让我们定义我们的模型并填充一些测试数据：

class Sample(Model):
    key = TextField()
    value = FloatField()

data = (
    ('a', (1.25, 1.5, 1.75)),
    ('b', (2.1, 2.3, 2.5, 2.7, 2.9)),
    ('c', (3.5, 3.5)))

# Populate data.
for key, values in data:
    Sample.insert_many([(key, value) for value in values],
                       fields=[Sample.key, Sample.value]).execute()
1
2
3
4
5
6
7
8
9
10
11
12
13

让我们使用 CTE 来计算，对于每个不同的键，哪些值高于该键的平均值。

# First we'll declare the query that will be used as a CTE. This query
# simply determines the average value for each key.
cte = (Sample
       .select(Sample.key, fn.AVG(Sample.value).alias('avg_value'))
       .group_by(Sample.key)
       .cte('key_avgs', columns=('key', 'avg_value')))

# Now we'll query the sample table, using our CTE to find rows whose value
# exceeds the average for the given key. We'll calculate how far above the
# average the given sample's value is, as well.
query = (Sample
         .select(Sample.key, Sample.value)
         .join(cte, on=(Sample.key == cte.c.key))
         .where(Sample.value > cte.c.avg_value)
         .order_by(Sample.value)
         .with_cte(cte))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

我们可以遍历查询返回的样本，以查看哪些样本的给定组的值高于平均值：

>>> for sample in query:
...     print(sample.key, sample.value)

# 'a', 1.75
# 'b', 2.7
# 'b', 2.9
1
2
3
4
5
6

8.19.2 复杂示例

对于更完整的示例，让我们考虑以下查询，该查询使用多个 CTE 仅在最高销售区域中查找每个产品的销售总额。我们的模型如下所示：

class Order(Model):
    region = TextField()
    amount = FloatField()
    product = TextField()
    quantity = IntegerField()
1
2
3
4
5

下面是用 SQL 编写查询的方式。这个例子可以在postgresql 文档中找到。

WITH regional_sales AS (
    SELECT region, SUM(amount) AS total_sales
    FROM orders
    GROUP BY region
  ), top_regions AS (
    SELECT region
    FROM regional_sales
    WHERE total_sales > (SELECT SUM(total_sales) / 10 FROM regional_sales)
  )
SELECT region,
       product,
       SUM(quantity) AS product_units,
       SUM(amount) AS product_sales
FROM orders
WHERE region IN (SELECT region FROM top_regions)
GROUP BY region, product;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

使用 Peewee，我们会这样写：

reg_sales = (Order
             .select(Order.region,
                     fn.SUM(Order.amount).alias('total_sales'))
             .group_by(Order.region)
             .cte('regional_sales'))

top_regions = (reg_sales
               .select(reg_sales.c.region)
               .where(reg_sales.c.total_sales > (
                   reg_sales.select(fn.SUM(reg_sales.c.total_sales) / 10)))
               .cte('top_regions'))

query = (Order
         .select(Order.region,
                 Order.product,
                 fn.SUM(Order.quantity).alias('product_units'),
                 fn.SUM(Order.amount).alias('product_sales'))
         .where(Order.region.in_(top_regions.select(top_regions.c.region)))
         .group_by(Order.region, Order.product)
         .with_cte(regional_sales, top_regions))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

8.19.3 递归 CTE

Peewee 支持递归 CTE。例如，当您有一个由父链接外键表示的树数据结构时，递归 CTE 可能很有用。例如，假设我们有一个在线书店的类别层次结构。我们希望生成一个表格，显示所有类别及其绝对深度，以及从根到类别的路径。

我们将假设以下模型定义，其中每个类别都有一个指向其直接父类别的外键：

class Category(Model):
    name = TextField()
    parent = ForeignKeyField('self', backref='children', null=True)
1
2
3

要列出所有类别及其深度和父级，我们可以使用递归 CTE：

# Define the base case of our recursive CTE. This will be categories that
# have a null parent foreign-key.
Base = Category.alias()
level = Value(1).alias('level')
path = Base.name.alias('path')
base_case = (Base
             .select(Base.id, Base.name, Base.parent, level, path)
             .where(Base.parent.is_null())
             .cte('base', recursive=True))

# Define the recursive terms.
RTerm = Category.alias()
rlevel = (base_case.c.level + 1).alias('level')
rpath = base_case.c.path.concat('->').concat(RTerm.name).alias('path')
recursive = (RTerm
             .select(RTerm.id, RTerm.name, RTerm.parent, rlevel, rpath)
             .join(base_case, on=(RTerm.parent == base_case.c.id)))

# The recursive CTE is created by taking the base case and UNION ALL with
# the recursive term.
cte = base_case.union_all(recursive)

# We will now query from the CTE to get the categories, their levels,  and
# their paths.
query = (cte
         .select_from(cte.c.name, cte.c.level, cte.c.path)
         .order_by(cte.c.path))

# We can now iterate over a list of all categories and print their names,
# absolute levels, and path from root -> category.
for category in query:
    print(category.name, category.level, category.path)

# Example output:
# root, 1, root
# p1, 2, root->p1
# c1-1, 3, root->p1->c1-1
# c1-2, 3, root->p1->c1-2
# p2, 2, root->p2
# c2-1, 3, root->p2->c2-1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

8.20 外键和连接

此部分已移至其自己的文档中：Relationships and Joins。