十、 关系和连接

在本文档中，我们将介绍 Peewee 如何处理模型之间的关系。

10.1 模型定义

我们将在示例中使用以下模型定义：

import datetime
from peewee import *db = SqliteDatabase(':memory:')class BaseModel(Model):class Meta:database = dbclass User(BaseModel):username = TextField()class Tweet(BaseModel):content = TextField()timestamp = DateTimeField(default=datetime.datetime.now)user = ForeignKeyField(User, backref='tweets')class Favorite(BaseModel):user = ForeignKeyField(User, backref='favorites')tweet = ForeignKeyField(Tweet, backref='favorites')

Peewee 用于ForeignKeyField定义模型之间的外键关系。每个外键字段都有一个隐含的反向引用，它使用提供的属性作为预过滤Select查询 公开。backref

10.1.1 创建测试数据

为了跟随示例，让我们用一些测试数据填充这个数据库：

def populate_test_data():db.create_tables([User, Tweet, Favorite])data = (('huey', ('meow', 'hiss', 'purr')),('mickey', ('woof', 'whine')),('zaizee', ()))for username, tweets in data:user = User.create(username=username)for tweet in tweets:Tweet.create(user=user, content=tweet)# Populate a few favorites for our users, such that:favorite_data = (('huey', ['whine']),('mickey', ['purr']),('zaizee', ['meow', 'purr']))for username, favorites in favorite_data:user = User.get(User.username == username)for content in favorites:tweet = Tweet.get(Tweet.content == content)Favorite.create(user=user, tweet=tweet)

这给了我们以下信息：

User	Tweet	Favorited by
huey	meow	zaizee
huey	hiss
huey	purr	mickey, zaizee
mickey	woof
mickey	whine	huey

在以下示例中，我们将执行一些查询。如果您不确定正在执行多少查询，您可以添加以下代码，它将所有查询记录到控制台：

import logging
logger = logging.getLogger('peewee')
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)

笔记

在 SQLite 中，默认情况下不启用外键。大多数事情，包括 Peewee 外键 API，都可以正常工作，但 ON DELETE
行为将被忽略，即使您on_delete在 ForeignKeyField. 结合默认 AutoField行为（可以重用已删除的记录
ID），这可能会导致细微的错误。为避免出现问题，我建议您在使用 SQLite 时启用外键约束，方法是
在实例化时设置.pragmas={‘foreign_keys’: 1}SqliteDatabase

# Ensure foreign-key constraints are enforced.
db = SqliteDatabase('my_app.db', pragmas={'foreign_keys': 1})

10.2 执行简单连接

作为学习如何使用 Peewee 执行连接的练习，让我们编写一个查询以打印出“huey”的所有推文。为此，我们将从 Tweet模型中选择并加入User模型，然后我们可以在字段上进行过滤 User.username：

>>> query = Tweet.select().join(User).where(User.username == 'huey')
>>> for tweet in query:
...     print(tweet.content)
...
meow
hiss
purr

笔记

我们不必明确指定连接谓词（“ON”子句），因为 Peewee 从模型中推断，当我们从 Tweet
连接到用户时，我们是Tweet.user在外键上连接的。

以下代码是等效的，但更明确：

query = (Tweet.select().join(User, on=(Tweet.user == User.id)).where(User.username == 'huey'))

如果我们已经有了User对“huey”对象的引用，我们可以使用User.tweets反向引用来列出所有huey的推文：

>>> huey = User.get(User.username == 'huey')
>>> for tweet in huey.tweets:
...     print(tweet.content)
...
meow
hiss
purr

仔细看huey.tweets，我们可以看到它只是一个简单的预过滤SELECT查询：

>>> huey.tweets
<peewee.ModelSelect at 0x7f0483931fd0>>>> huey.tweets.sql()
('SELECT "t1"."id", "t1"."content", "t1"."timestamp", "t1"."user_id"FROM "tweet" AS "t1" WHERE ("t1"."user_id" = ?)', [1])

10.3 连接多个表

让我们通过查询用户列表并获取他们创作的推文的数量被收藏来再次查看连接。这将需要我们加入两次：从用户到推文，以及从推文到收藏。我们将添加额外的要求，即应包括尚未创建任何推文的用户，以及推文未被收藏的用户。以 SQL 表示的查询将是：

SELECT user.username, COUNT(favorite.id)
FROM user
LEFT OUTER JOIN tweet ON tweet.user_id = user.id
LEFT OUTER JOIN favorite ON favorite.tweet_id = tweet.id
GROUP BY user.username

笔记

在上面的查询中，两个连接都是 LEFT OUTER，因为用户可能没有任何推文，或者，如果他们有推文，它们可能都没有被收藏。

Peewee 有一个join context的概念，这意味着每当我们调用该 join()方法时，我们都会隐式地加入先前加入的模型（或者如果这是第一次调用，则我们正在从中选择模型）。由于我们是直接加入的，从用户到推文，然后从推文到收藏，我们可以简单地写：

query = (User.select(User.username, fn.COUNT(Favorite.id).alias('count')).join(Tweet, JOIN.LEFT_OUTER)  # Joins user -> tweet..join(Favorite, JOIN.LEFT_OUTER)  # Joins tweet -> favorite..group_by(User.username))

迭代结果：

>>> for user in query:
...     print(user.username, user.count)
...
huey 3
mickey 1
zaizee 0

对于涉及多个连接和切换连接上下文的更复杂的示例，让我们查找 Huey 的所有推文以及它们被收藏的次数。为此，我们需要执行两次连接，并且我们还将使用聚合函数来计算收藏次数。

下面是我们如何在 SQL 中编写此查询：

SELECT tweet.content, COUNT(favorite.id)
FROM tweet
INNER JOIN user ON tweet.user_id = user.id
LEFT OUTER JOIN favorite ON favorite.tweet_id = tweet.id
WHERE user.username = 'huey'
GROUP BY tweet.content;

笔记

我们使用从推文到收藏夹的 LEFT OUTER 连接，因为推文可能没有任何收藏夹，但我们仍然希望在结果集中显示它的内容（以及零计数）。

使用 Peewee，生成的 Python 代码看起来非常类似于我们用 SQL 编写的代码：

query = (Tweet.select(Tweet.content, fn.COUNT(Favorite.id).alias('count')).join(User)  # Join from tweet -> user..switch(Tweet)  # Move "join context" back to tweet..join(Favorite, JOIN.LEFT_OUTER)  # Join from tweet -> favorite..where(User.username == 'huey').group_by(Tweet.content))

请注意对switch() 的调用，它指示 Peewee 将连接上下文设置回Tweet. 如果我们省略了对 switch 的显式调用，Peewee 将使用User（我们加入的最后一个模型）作为连接上下文，并使用 Favorite.user外键构造从用户到收藏夹的连接，这会给我们带来不正确的结果。

如果我们想省略连接上下文切换，我们可以改用该 join_from()方法。以下查询等效于前一个查询：

query = (Tweet.select(Tweet.content, fn.COUNT(Favorite.id).alias('count')).join_from(Tweet, User)  # Join tweet -> user..join_from(Tweet, Favorite, JOIN.LEFT_OUTER)  # Join tweet -> favorite..where(User.username == 'huey').group_by(Tweet.content))

我们可以遍历上述查询的结果以打印推文的内容和收藏次数：

>>> for tweet in query:
...     print('%s favorited %d times' % (tweet.content, tweet.count))
...
meow favorited 1 times
hiss favorited 0 times
purr favorited 2 times

10.4 从多个来源中选择

如果我们希望列出数据库中的所有推文及其作者的用户名，您可以尝试这样写：

>>> for tweet in Tweet.select():
...     print(tweet.user.username, '->', tweet.content)
...
huey -> meow
huey -> hiss
huey -> purr
mickey -> woof
mickey -> whine

上面的循环有一个大问题：它为每条推文执行一个额外的查询来查找tweet.user外键。对于我们的小表，性能损失并不明显，但我们会发现延迟随着行数的增加而增加。

如果您熟悉 SQL，您可能还记得可以从多个表中进行 SELECT，从而允许我们在单个查询中获取推文内容和用户名：

SELECT tweet.content, user.username
FROM tweet
INNER JOIN user ON tweet.user_id = user.id;

Peewee 使这很容易。事实上，我们只需要稍微修改一下我们的查询。我们告诉 Peewee 我们希望选择Tweet.content该User.username字段，然后我们包含从推文到用户的连接。为了更清楚地表明它正在做正确的事情，我们可以要求 Peewee 将行作为字典返回。

>>> for row in Tweet.select(Tweet.content, User.username).join(User).dicts():
...     print(row)
...
{'content': 'meow', 'username': 'huey'}
{'content': 'hiss', 'username': 'huey'}
{'content': 'purr', 'username': 'huey'}
{'content': 'woof', 'username': 'mickey'}
{'content': 'whine', 'username': 'mickey'}

现在我们将停止对“.dicts()”的调用并将行作为Tweet 对象返回。请注意，Peewee 将username值分配给 tweet.user.username- NOT tweet.username！因为从 tweet 到 user 有一个外键，并且我们从两个模型中选择了字段，所以 Peewee 将为我们重建模型图：

>>> for tweet in Tweet.select(Tweet.content, User.username).join(User):
...     print(tweet.user.username, '->', tweet.content)
...
huey -> meow
huey -> hiss
huey -> purr
mickey -> woof
mickey -> whine

如果我们愿意，我们可以通过在方法中User指定 an 来控制 Peewee 在上述查询中放置连接实例的位置：attrjoin()

>>> query = Tweet.select(Tweet.content, User.username).join(User, attr='author')
>>> for tweet in query:
...     print(tweet.author.username, '->', tweet.content)
...
huey -> meow
huey -> hiss
huey -> purr
mickey -> woof
mickey -> whine

相反，如果我们只是希望我们选择的所有属性都是Tweet实例的属性，我们可以objects()在查询末尾添加一个调用（类似于我们调用的方式dicts()）：

>>> for tweet in query.objects():
...     print(tweet.username, '->', tweet.content)
...
huey -> meow
(etc)

10.4.1 更复杂的例子

作为一个更复杂的示例，在此查询中，我们将编写一个查询，该查询选择所有收藏夹，以及创建收藏夹的用户、收藏的推文以及该推文的作者。

在 SQL 中，我们会写：

SELECT owner.username, tweet.content, author.username AS author
FROM favorite
INNER JOIN user AS owner ON (favorite.user_id = owner.id)
INNER JOIN tweet ON (favorite.tweet_id = tweet.id)
INNER JOIN user AS author ON (tweet.user_id = author.id);

请注意，我们从用户表中选择了两次——一次是在创建收藏夹的用户的上下文中，另一次是作为推文的作者。

使用 Peewee，我们使用Model.alias()别名模型类，以便可以在单个查询中引用它两次：

Owner = User.alias()
query = (Favorite.select(Favorite, Tweet.content, User.username, Owner.username).join(Owner)  # Join favorite -> user (owner of favorite)..switch(Favorite).join(Tweet)  # Join favorite -> tweet.join(User))   # Join tweet -> user

我们可以通过以下方式遍历结果并访问连接的值。请注意 Peewee 如何从我们选择的各种模型中解析字段并重建模型图：

>>> for fav in query:
...     print(fav.user.username, 'liked', fav.tweet.content, 'by', fav.tweet.user.username)
...
huey liked whine by mickey
mickey liked purr by huey
zaizee liked meow by huey
zaizee liked purr by huey

10.5 子查询

Peewee 允许您加入任何类似表的对象，包括子查询或公用表表达式 (CTE)。为了演示加入子查询，让我们查询所有用户及其最新推文。

这是SQL：

SELECT tweet.*, user.*
FROM tweet
INNER JOIN (SELECT latest.user_id, MAX(latest.timestamp) AS max_tsFROM tweet AS latestGROUP BY latest.user_id) AS latest_query
ON ((tweet.user_id = latest_query.user_id) AND (tweet.timestamp = latest_query.max_ts))
INNER JOIN user ON (tweet.user_id = user.id)

我们将通过创建一个选择每个用户及其最新推文时间戳的子查询来做到这一点。然后我们可以在外部查询中查询推文表，并从子查询中加入用户和时间戳组合。

# Define our subquery first. We'll use an alias of the Tweet model, since
# we will be querying from the Tweet model directly in the outer query.
Latest = Tweet.alias()
latest_query = (Latest.select(Latest.user, fn.MAX(Latest.timestamp).alias('max_ts')).group_by(Latest.user).alias('latest_query'))# Our join predicate will ensure that we match tweets based on their
# timestamp *and* user_id.
predicate = ((Tweet.user == latest_query.c.user_id) &(Tweet.timestamp == latest_query.c.max_ts))# We put it all together, querying from tweet and joining on the subquery
# using the above predicate.
query = (Tweet.select(Tweet, User)  # Select all columns from tweet and user..join(latest_query, on=predicate)  # Join tweet -> subquery..join_from(Tweet, User))  # Join from tweet -> user.

遍历查询，我们可以看到每个用户及其最新的推文。

>>> for tweet in query:
...     print(tweet.user.username, '->', tweet.content)
...
huey -> purr
mickey -> whine

在我们用于在本节中创建查询的代码中，有几件事您可能以前没有见过：

• 我们曾经join_from()明确指定连接上下文。我们写了，相当于 …join_from(Tweet, User).switch(Tweet).join(User)
• 我们使用魔法属性引用了子查询中的列.c，例如latest_query.c.max_ts. 该.c属性用于动态创建列引用。
• 我们没有将单个字段传递给Tweet.select()，而是传递了 TweetandUser模型。这是选择给定模型上所有字段的简写。

10.5.1 公用表表达式

在上一节中，我们加入了子查询，但我们也可以轻松地使用公共表表达式 (CTE)。我们将重复与之前相同的查询，列出用户及其最新的推文，但这次我们将使用 CTE 来完成。

这是SQL：

WITH latest AS (SELECT user_id, MAX(timestamp) AS max_tsFROM tweetGROUP BY user_id)
SELECT tweet.*, user.*
FROM tweet
INNER JOIN latestON ((latest.user_id = tweet.user_id) AND (latest.max_ts = tweet.timestamp))
INNER JOIN userON (tweet.user_id = user.id)

这个例子看起来与前面带有子查询的例子非常相似：

# Define our CTE first. We'll use an alias of the Tweet model, since
# we will be querying from the Tweet model directly in the main query.
Latest = Tweet.alias()
cte = (Latest.select(Latest.user, fn.MAX(Latest.timestamp).alias('max_ts')).group_by(Latest.user).cte('latest'))# Our join predicate will ensure that we match tweets based on their
# timestamp *and* user_id.
predicate = ((Tweet.user == cte.c.user_id) &(Tweet.timestamp == cte.c.max_ts))# We put it all together, querying from tweet and joining on the CTE
# using the above predicate.
query = (Tweet.select(Tweet, User)  # Select all columns from tweet and user..join(cte, on=predicate)  # Join tweet -> CTE..join_from(Tweet, User)  # Join from tweet -> user..with_cte(cte))

我们可以遍历结果集，其中包含每个用户的最新推文：

>>> for tweet in query:
...     print(tweet.user.username, '->', tweet.content)
...
huey -> purr
mickey -> whine

笔记

有关使用 CTE 的更多信息，包括有关编写递归 CTE 的信息，请参阅“查询”文档的公用表表达式部分。

10.6 同一模型的多个外键

当同一模型有多个外键时，最好明确指定要加入的字段。

回顾示例应用程序的模型，考虑关系模型，它用于表示一个用户何时关注另一个用户。这是模型定义：

class Relationship(BaseModel):from_user = ForeignKeyField(User, backref='relationships')to_user = ForeignKeyField(User, backref='related_to')class Meta:indexes = (# Specify a unique multi-column index on from/to-user.(('from_user', 'to_user'), True),)

由于User有两个外键，我们应该始终指定我们在连接中使用哪个字段。

例如，要确定我关注哪些用户，我会写：

(User.select().join(Relationship, on=Relationship.to_user).where(Relationship.from_user == charlie))

另一方面，如果我想确定哪些用户在关注我，我会加入from_user列并过滤关系的 to_user：

(User.select().join(Relationship, on=Relationship.from_user).where(Relationship.to_user == charlie))

10.7 加入任意字段

如果两个表之间不存在外键，您仍然可以执行连接，但您必须手动指定连接谓词。

在以下示例中， User和 ActivityLog之间没有显式外键，但ActivityLog.object_id字段和User.id之间存在隐含关系 。Field我们将使用加入，而不是加入特定 的Expression.

user_log = (User.select(User, ActivityLog).join(ActivityLog, on=(User.id == ActivityLog.object_id), attr='log').where((ActivityLog.activity_type == 'user_activity') &(User.username == 'charlie')))for user in user_log:print(user.username, user.log.description)#### Print something like ####
charlie logged in
charlie posted a tweet
charlie retweeted
charlie posted a tweet
charlie logged out

笔记

attr回想一下，我们可以通过在方法中指定参数来控制 Peewee 将分配连接实例的属性join()。在前面的示例中，我们使用了以下连接：

join(ActivityLog, on=(User.id == ActivityLog.object_id), attr='log')

然后在遍历查询时，我们能够直接访问加入的ActivityLog，而不会产生额外的查询：

for user in user_log:print(user.username, user.log.description)