标签: aiohttp

正如标题所说，我的用例是这样的：

我有一个 aiohttp 服务器，它接受来自客户端的请求，当我收到请求时，我为它生成一个唯一的请求 ID，然后我将{req_id: req_pyaload}dict发送给一些工作人员（工作人员不在 python 中，因此在另一个进程中运行），当工人完成的工作，我回来的响应，并把它们放到一个结果字典是这样的：{req_id_1: res_1, req_id_2: res_2}。

然后我希望我的 aiohttp 服务器处理程序await在上面result dict，所以当特定的响应变得可用时（通过 req_id）它可以将它发送回来。

下面的示例代码，我建立以尽量模拟过程中，却被困在执行协程async def fetch_correct_res(req_id)应异步/ unblockly获取由正确的响应req_id。

import random
import asyncio
import shortuuid

n_tests = 1000

idxs = list(range(n_tests))

req_ids = []
for _ in range(n_tests):
    req_ids.append(shortuuid.uuid())

res_dict = {}

async def fetch_correct_res(req_id):
    pass

async def handler(req):
    res = await fetch_correct_res(req)
    assert req == res, "the correct res for the req should exactly …

想象一下，我有一个基于 Aiohttp 的 Web 应用程序：

from aiohttp import web
import asyncio
import logging

logger = logging.getLogger(__name__)

async def hello(request):
    logger.info('Started processing request')
    await asyncio.sleep(1)
    logger.info('Doing something')
    await asyncio.sleep(1)
    return web.Response(text="Hello, world!\n")

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s %(name)-14s %(levelname)s: %(message)s')

app = web.Application()
app.add_routes([web.get('/', hello)])
web.run_app(app)

它的输出是（例如）：

2019-11-11 13:37:14,757 __main__       INFO: Started processing request
2019-11-11 13:37:14,757 __main__       INFO: Started processing request
2019-11-11 13:37:15,761 __main__       INFO: Doing something
2019-11-11 13:37:15,761 __main__       INFO: Doing something
2019-11-11 13:37:16,765 aiohttp.access INFO: 127.0.0.1 [11/Nov/2019:12:37:14 +0000] "GET / HTTP/1.1" …

我正在使用 asyncio 和 aiohttp 来发出并发请求。我最近将 Python 升级到了 3.8.0 版，并且RuntimeError: Event loop is closed在程序运行后得到了一个。

import asyncio
import aiohttp

async def do_call(name, session):
    async with session.get('https://www.google.be') as response:
        await response.text()
        return 'ok - {}'.format(name)

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [do_call(str(i), session) for i in range(0,4)]
        results = await asyncio.gather(*tasks)
        print(results)

asyncio.run(main())

我确实从 asyncio.gather() 得到了一个有效的结果，但是在退出时会引发异常。我想更改代码，以免遇到异常。

回溯如下：

Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x000001E9A92079D0>
Traceback (most recent call last):
  File "C:\Users\Jonas\AppData\Local\Programs\Python\Python38\lib\asyncio\proactor_events.py", line 116, in __del__
    self.close()
  File …

场景：我需要从 Web 应用程序的 API 收集分页数据，该 API 的调用限制为每分钟 100 次。我需要返回的 API 对象每页包含 100 个项目，总共有 105 个页面，而且还在不断增加（总共约 10,500 个项目）。同步代码需要大约 15 分钟来检索所有页面，因此那时不必担心达到调用限制。但是，我想加快数据检索速度，因此我使用asyncio和实现了异步调用aiohttp。数据现在在 15 秒内下载 - 很好。

问题：我现在达到了呼叫限制，因此在最近 5 次左右的呼叫中收到 403 错误。

建议的解决方案我实现了try/except在get_data()函数中找到的。我拨打电话，然后当由于403: Exceeded call limit我退后back_off几秒钟而未成功拨打电话并重试retries多次时：

async def get_data(session, url):
    retries = 3
    back_off = 60  # seconds to try again
    for _ in range(retries):
        try:
            async with session.get(url, headers=headers) as response:
                if response.status != 200:
                    response.raise_for_status() …

我在做什么

我正在通过构建一个 REST api 来学习 aiohttp，我正在使用 Pytest（及其 async 和 aiohttp 插件）进行测试。

对于我的第一次测试（我从一开始就使用 TDD）我有以下代码：

@pytest.mark.asyncio
async def test_handle_user_create(
    aiohttp_client, init_test_app, create_test_user_table
):
    payload = {
        "email": "tintin@gmail.com",
        "username": "Tintin",
        "password": "y0u != n00b1e",
    }
    client = await aiohttp_client(init_test_app)
    resp = await client.post("/users/", json=payload)
    ...

• aiohttp_client 是客户端装置来自 pytest-aiohttp
• init_test_app 是一个固定装置，它基本上反映了我将要构建的应用程序
• create_test_user_table 是我在测试数据库中为用户创建表的夹具

它有什么问题

我的第一个测试是在上面代码块的最后一行抛出以下运行时错误：

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ …

这是基于一个错字和简单的错误。

不删除，因为它有 httpx 的示例代码。

我正在尝试利用asyncio并行化几个长时间运行的 Web 请求。因为我是从requests库中迁移过来的，所以我想使用这个httpx库，因为它有类似的 API。我的环境是 Python 3.7.7 Anaconda 发行版，其中安装了所有必需的软件包（Windows 10）。

然而，尽管能够httpx用于同步 Web 请求（或用于串行执行一个接一个运行的异步请求），但我无法成功地一次运行多个异步请求，尽管使用aiohttp库很容易做到这一点.

这是在aiohttp: (请注意，我在 Jupyter 中运行，所以我已经有一个事件循环，因此缺少asyncio.run().

import aiohttp
import asyncio
import time
import httpx

async def call_url(session):
    url = "https://services.cancerimagingarchive.net/services/v3/TCIA/query/getCollectionValues"        
    response = await session.request(method='GET', url=url)
    #response.raise_for_status() 
    return response

for i in range(1,5):
    start = time.time() # start time for timing event
    async with aiohttp.ClientSession() as session: #use aiohttp
    #async with httpx.AsyncClient as …

我需要抓取并获取许多（每天 5-10k）新闻文章的正文段落的原始文本。我已经编写了一些线程代码，但考虑到这个项目的高度 I/O 限制性质，我正在涉足asyncio. 下面的代码片段并不比 1 线程版本快，而且比我的线程版本差得多。谁能告诉我我做错了什么？谢谢你！

async def fetch(session,url):
    async with session.get(url) as response:
        return await response.text()

async def scrape_urls(urls):
    results = []
    tasks = []
    async with aiohttp.ClientSession() as session:
        for url in urls:
            html = await fetch(session,url)
            soup = BeautifulSoup(html,'html.parser')
            body = soup.find('div', attrs={'class':'entry-content'})
            paras = [normalize('NFKD',para.get_text()) for para in body.find_all('p')]
            results.append(paras)
    return results