Twisted Requests (treq) 包是一个 HTTP 客户端，构建在流行的 Twisted 库之上，用于异步请求。异步库提供了并行执行大量网络请求的能力，且 CPU 影响相对较小。这对于需要在拥有所有所需信息之前发出多个请求的 HTTP 客户端非常有用。在本文中，我们将通过一个进行异步调用的示例来探索如何使用 treq。
定义要解决的问题
我喜欢玩实时战略游戏 Clash Royale。虽然它不是开源的，但它确实有一个公共 API，我们可以用它来展示异步请求的便利性。
Clash Royale 是一款移动战略玩家对玩家游戏，玩家在竞技场中出牌以获胜。每张卡牌都有不同的优势和劣势，不同的玩家喜欢不同的卡牌。Clash Royale 会记住玩家最常玩的卡牌；这是他们“最喜欢的”卡牌。玩家们聚集在部落中，互相帮助。Clash Royale 的开发者 Supercell 发布了一个基于 HTTP 的 API，可以查询不同的统计数据。
这是一个最适合异步回答的问题：我们如何编写一个程序来输出部落中最受欢迎的卡牌，以便我们可以开始了解我们的对手（并了解哪些卡牌在我们的部落成员中受欢迎）？
您可以注册一个帐户以跟随本教程，但即使您不注册，您仍然能够理解我们正在构建的内容。如果您确实想注册一个帐户，请通过 Clash Royale 开发者门户创建一个 API 令牌。然后选择您个人资料下的“创建新密钥”，并输入名称、描述和有效的 IP 地址。（需要确切的地址，所以我使用了这个网站来查找我的地址。）由于您永远不应将 API 密钥保存在代码中，请将其作为单独的文件保存在 ~/.crtoken 中
$ ls ~/.crtoken
/home/moshez/.crtokenTwisted 程序
运行基于 Twisted 的程序需要许多额外的软件包，以尽可能使体验顺畅。在本教程中，我不会介绍所有软件包，但每个软件包都值得探索以了解更多信息。
为了更容易了解正在发生的事情，让我们从这个打印 Hello world 的入门程序开始，然后我们将讨论它的作用
import collections, json, os, sys, urllib.parse
from twisted.internet import task, defer
import treq
with open(os.path.expanduser("~/.crtoken")) as fpin:
token = fpin.read().strip()
def main(reactor):
print("Hello world")
return defer.succeed(None)
task.react(main, sys.argv[1:])此代码导入的模块比“Hello world”示例所需的模块多得多。对于程序的最终版本，我们将需要这些模块，最终版本将完成更复杂的任务，即异步查询 API。导入后，程序从文件中读取令牌并将其存储在变量 token 中。（我们现在不会对令牌做任何事情，但很高兴看到该语法。）接下来是一个接受 Twisted reactor 的 main 函数。reactor 有点像 Twisted 包复杂机制的接口。在这种情况下，函数 main 作为参数发送，并为其提供了一个额外的参数。
main 返回 defer.succeed(None)。这就是它如何返回正确类型的值：一个延迟值，但该值已经“触发”或“调用”。因此，程序将在打印 Hello world 后立即退出，正如我们所需要的。
接下来，我们将了解 async 函数和 ensureDeferred 的概念
async def get_clan_details(clan):
print("Hello world", clan)
def main(reactor, clan):
return defer.ensureDeferred(get_clan_details(clan))
task.react(main, sys.argv[1:])在这个程序中，它应该从相同的导入开始，我们将所有逻辑移至异步函数 get_clan_details。就像常规函数一样，async 函数在末尾有一个隐式的 return None。但是，异步函数（有时称为协程）是一种与 Deferred 不同的类型。为了让自 Python 1.5.2 以来就存在的 Twisted 使用这个现代功能，我们必须使用 ensureDeferred 来适配协程。
虽然我们可以编写所有逻辑而无需使用协程，但使用 async 语法将使我们能够编写更易于理解的代码，并且我们将需要将更少的代码移入嵌入式回调中。
要介绍的下一个概念是 await。稍后，我们将 await 网络调用，但为了简单起见，现在，我们将 await 定时器。Twisted 有一个特殊函数 task.deferLater，它将在一段时间后使用给定的参数调用一个函数。
以下程序将花费五秒钟完成
async def get_clan_details(clan, reactor):
out = await task.deferLater(
reactor,
5,
lambda clan: f"Hello world {clan}",
clan
)
print(out)
def main(reactor, clan):
return defer.ensureDeferred(get_clan_details(clan, reactor))
task.react(main, sys.argv[1:])关于类型的说明：task.deferLater 返回一个 Deferred，大多数没有立即提供值的 Twisted 函数也是如此。运行 Twisted 事件循环时，我们可以 await Deferred 值和协程。
函数 task.deferLater 将等待五秒钟，然后调用我们的 lambda，计算要打印的字符串。
现在我们拥有编写高效部落分析程序所需的所有 Twisted 构建块了！
使用 treq 进行异步调用
由于我们将使用全局 reactor，因此我们不再需要在计算这些统计信息的函数中将 reactor 作为参数接受
async def get_clan_details(clan):使用令牌的方法是在标头中作为“bearer”令牌
headers={b'Authorization': b'Bearer '+token.encode('ascii')}我们希望发送部落标签，它将是字符串。部落标签以 # 开头，因此在将它们放入 URL 之前必须引用它们。这是因为 # 具有特殊含义“URL 片段”
clan = urllib.parse.quote(clan)第一步是获取部落的详细信息，包括部落成员
res = await treq.get("https://api.clashroyale.com/v1/clans/" + clan,
headers=headers)请注意，我们必须 await treq.get 调用。我们必须明确何时等待并获取信息，因为它是一个异步网络调用。仅使用 await 语法调用 Deferred 函数 不能 让我们充分发挥异步性的力量（稍后我们将看到如何做到这一点）。
接下来，在获取标头后，我们需要获取内容。treq 库为我们提供了一个辅助方法，可以直接解析 JSON
content = await res.json()内容包括一些关于部落的元数据，这些元数据对于我们当前的目的来说并不有趣，以及一个包含部落成员的 memberList 字段。请注意，虽然它包含一些关于玩家的数据，但当前最喜欢的卡牌不是其中的一部分。它确实包含唯一的“玩家标签”，我们可以使用它来检索更多数据。
我们收集所有玩家标签，并且由于它们也以 # 开头，因此我们对它们进行 URL 引用
player_tags = [urllib.parse.quote(player['tag'])
for player in content['memberList']]最后，我们来到了 treq 和 Twisted 的真正力量：一次生成所有玩家数据请求！这可以真正加快此类任务的速度，即一遍又一遍地查询 API。在具有速率限制的 API 的情况下，这可能会有问题。
有时我们需要体谅我们的 API 所有者，并且不要触及任何速率限制。有一些技术可以在 Twisted 中显式支持速率限制，但它们超出了本教程的范围。（一个重要的工具是 defer.DeferredSemaphore。）
requests = [treq.get("https://api.clashroyale.com/v1/players/" + tag,
headers=headers)
for tag in player_tags]题外话：await、Deferred 和回调
对于那些对返回对象的具体细节感到好奇的人，这里更仔细地看看正在发生的事情。
请记住，请求不会直接返回 JSON 正文。之前，我们使用 await，这样我们就不必担心请求返回的具体内容。它们实际上返回一个 Deferred。Deferred 可以附加一个 回调，它将修改 Deferred。如果回调 返回 Deferred，则 Deferred 的最终值将是返回的 Deferred 的 值。
因此，对于每个 deferred，我们附加一个回调，它将检索正文的 JSON
for request in requests:
request.addCallback(lambda result: result.json())将回调附加到 Deferred 是一种更手动的技术，这使得代码更难跟踪，但更有效地使用了异步功能。具体来说，因为我们同时附加所有回调，所以我们不需要等待网络调用（可能需要很长时间）来指示如何后处理结果。
从 Deferred 到值
在收集所有结果之前，我们无法计算最受欢迎的卡牌。我们有一个 Deferred 列表，但我们想要的是一个 获得列表值 的 Deferred。这种反转正是 Twisted 函数 defer.gatherResults 所做的
all_players = await defer.gatherResults(requests)这个看似无辜的调用是我们使用 Twisted 全部力量的地方。defer.gatherResults 函数立即返回一个 deferred，该 deferred 仅在所有组成 Deferred 都已触发时才会 触发，并将使用结果触发。它甚至为我们提供了免费的错误处理：如果任何 Deferred 错误退出，它将立即返回一个失败的 deferred，这将导致 await 引发异常。
现在我们有了所有玩家的详细信息，我们需要处理一些数据。我们开始使用 Python 最酷的内置函数之一 collections.Counter。这个类接受一个事物列表，并计算它看到每个事物的次数，这正是我们进行投票计数或人气竞赛所需要的。
favorite_card = collections.Counter([player["currentFavouriteCard"]["name"]
for player in all_players])最后，我们打印出来
print(json.dumps(favorite_card.most_common(), indent=4))将它们放在一起
所以，将它们放在一起，我们有
import collections, json, os, sys, urllib.parse
from twisted.internet import task, defer
import treq
with open(os.path.expanduser("~/.crtoken")) as fpin:
token = fpin.read().strip()
async def get_clan_details(clan):
headers = headers={b'Authorization': b'Bearer '+token.encode('ascii')}
clan = urllib.parse.quote(clan)
res = await treq.get("https://api.clashroyale.com/v1/clans/" + clan,
headers=headers)
content = await res.json()
player_tags = [urllib.parse.quote(player['tag'])
for player in content['memberList']]
requests = [treq.get("https://api.clashroyale.com/v1/players/" + tag,
headers=headers)
for tag in player_tags]
for request in requests:
request.addCallback(lambda result: result.json())
all_players = await defer.gatherResults(requests)
favorite_card = collections.Counter([player["currentFavouriteCard"]["name"]
for player in all_players])
print(json.dumps(favorite_card.most_common(), indent=4))
def main(reactor, clan):
return defer.ensureDeferred(get_clan_details(clan))
task.react(main, sys.argv[1:])感谢 Twisted 和 treq 的效率和富有表现力的语法，这就是我们需要的所有代码，用于对 API 进行异步调用。如果您想知道结果，我部落中最喜欢的卡牌列表是法师、超级骑士、瓦基丽和皇家巨人，按降序排列。
希望您喜欢使用 Twisted 编写更快的 API 调用！
1 条评论