Apis被描述为类，例如
class Categories(JsonEndpoint):
url = "http://127.0.0.1:8888/categories"
params = {"page": range(100), "language": "en"}
headers = {"User-Agent": get_user_agent}
results_key = "*.slug"
categories = Categories()
class Posts(JsonEndpoint):
url = "http://127.0.0.1:8888/categories/{category}/posts"
params = {"page": range(100), "language": "en"}
url_params = {"category": categories.iter_results()}
results_key = "posts"
async def comments(self, post):
comments = Comments(
self.session,
url_params={"category": post.url.params["category"], "id": post["id"]},
)
return [comment async for comment in comments]
posts = Posts()
参数和url_params可以包含函数（例如，这里的get_user_agent-返回随机useragent），范围，迭代器，等待和异步迭代器（因此您可以将它们链接在一起）。
参数标头和cookie也可以包含函数并且可以等待。
上例中的category api返回一个带有子对象的对象数组，迭代器将完全返回这些对象。通过将此迭代器放到帖子的url_params中，迭代器将递归地迭代所有类别以及每个类别中的所有页面。遇到404或其他错误时它将中止，然后转到下一个类别。
存储库中有一个用于这些类的aiohttp服务器示例，因此可以测试所有内容。
除了获取参数外，还可以将它们作为数据或json传递，并设置其他方法。
results_key被点起来，将尝试从结果中提取键。例如，“ comments。*。Text”将从注释内部的数组中返回每个注释的文本。
结果包装在具有url和params属性的包装器中。 url是从也具有参数的字符串派生的。因此，您可以找出使用了哪些参数来获得此结果，这在comment方法中得到了证明。
还有一个用于处理结果的基础Sink类。例如，将它们折叠到mq或数据库中。它在单独的任务中工作，并通过asyncio.Queue接收数据。
class LoggingSink(Sink):
def transform(self, obj):
return repr(obj)
async def init(self):
from loguru import logger
self.logger = logger
async def process(self, obj):
self.logger.info(obj)
return True
sink = LoggingSink(num_tasks=1)
最简单的接收器示例。 transform方法允许我们对对象进行一些操作，如果不适合我们，则返回None。那些。在主题中，您还可以进行验证。
Sink是一个异步上下文管理器，退出后，理论上将等待直到队列中的所有对象都处理完毕，然后取消其任务。
最后，为了将它们联系在一起，我开设了一个Worker类。它接受一个端点和多个接收器。例如，
worker = Worker(endpoint=posts, sinks=[loggingsink, mongosink])
worker.run()
运行将为辅助管道运行asyncio.run_until_complete。它还具有变换方法。
还有一个WorkerGroup类，它允许您一次创建多个工作程序并为其创建asyncio.gather。
该代码包含一个服务器示例，该服务器通过伪造者和其端点的处理程序生成数据。我认为这是最明显的。
所有这些都处于开发的早期阶段，到目前为止，我经常更改api。但是现在看来应该看起来如何了。我将合并请求和注释到我的代码中。