asyncio 网络抓取 101：使用 aiohttp 获取多个 url

Posted 2023-02-23

【中文标题】asyncio 网络抓取 101：使用 aiohttp 获取多个 url

【英文标题】：asyncio web scraping 101: fetching multiple urls with aiohttp

【发布时间】：2016-06-25 21:54:45

【问题描述】：

在较早的问题中，aiohttp 的一位作者建议使用 Python 3.5 中的新 async with 语法来使用 fetch multiple urls with aiohttp 的方式：

import aiohttp
import asyncio

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

async def fetch_all(session, urls, loop):
    results = await asyncio.wait([loop.create_task(fetch(session, url))
                                  for url in urls])
    return results

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # breaks because of the first url
    urls = ['http://SDFKHSKHGKLHSKLJHGSDFKSJH.com',
            'http://google.com',
            'http://twitter.com']
    with aiohttp.ClientSession(loop=loop) as session:
        the_results = loop.run_until_complete(
            fetch_all(session, urls, loop))
        # do something with the the_results

但是，当session.get(url) 请求之一中断时（如上所述，因为http://SDFKHSKHGKLHSKLJHGSDFKSJH.com），错误不会得到处理，整个事情都会中断。

我寻找插入有关session.get(url) 结果的测试的方法，例如寻找try ... except ... 或if response.status != 200: 的位置，但我只是不明白如何使用async with，@ 987654332@ 和各种对象。

由于async with 还很新，所以例子不多。如果asyncio 向导可以展示如何做到这一点，这对许多人来说将是非常有帮助的。毕竟，大多数人想要使用asyncio 进行测试的第一件事就是同时获取多个资源。

目标

目标是我们可以检查the_results 并快速查看：

此 URL 失败（以及原因：状态代码，可能是异常名称），或者 这个网址有效，这是一个有用的响应对象

【参考方案1】：

我会使用gather 而不是wait，它可以将异常作为对象返回，而不会引发它们。然后您可以检查每个结果，如果它是某个异常的实例。

import aiohttp
import asyncio

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

async def fetch_all(session, urls, loop):
    results = await asyncio.gather(
        *[fetch(session, url) for url in urls],
        return_exceptions=True  # default is false, that would raise
    )

    # for testing purposes only
    # gather returns results in the order of coros
    for idx, url in enumerate(urls):
        print(': '.format(url, 'ERR' if isinstance(results[idx], Exception) else 'OK'))
    return results

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # breaks because of the first url
    urls = [
        'http://SDFKHSKHGKLHSKLJHGSDFKSJH.com',
        'http://google.com',
        'http://twitter.com']
    with aiohttp.ClientSession(loop=loop) as session:
        the_results = loop.run_until_complete(
            fetch_all(session, urls, loop))

测试：

$python test.py 
http://SDFKHSKHGKLHSKLJHGSDFKSJH.com: ERR
http://google.com: OK
http://twitter.com: OK

【讨论】：

太棒了，非常感谢！我需要消化这个，但是在玩了一下之后，它似乎很灵活。 +1，接受。 :)

很好的答案。我很好奇的一件事，因为您在执行asyncio.gather 之后立即迭代结果，在fetches 的列表中执行asyncio.as_completed 不是更好吗？这样您就可以立即迭代已完成的，而不是等待它们全部完成？

@dalanmiller：它需要异常处理，就像在 Padraic Cunningham 的回答中一样。但是，如果您需要立即为每个 Future 提供结果，那么这就是方法。

【参考方案2】：

我远不是 asyncio 专家，但你想捕捉你需要捕捉套接字错误的错误：

async def fetch(session, url):
    with aiohttp.Timeout(10):
        try:
            async with session.get(url) as response:
                print(response.status == 200)
                return await response.text()
        except socket.error as e:
            print(e.strerror)

运行代码并打印the_results：

Cannot connect to host sdfkhskhgklhskljhgsdfksjh.com:80 ssl:False [Can not connect to sdfkhskhgklhskljhgsdfksjh.com:80 [Name or service not known]]
True
True
(<Task finished coro=<fetch() done, defined at <ipython-input-7-535a26aaaefe>:5> result='<!DOCTYPE ht...y>\n</html>\n'>, <Task finished coro=<fetch() done, defined at <ipython-input-7-535a26aaaefe>:5> result=None>, <Task finished coro=<fetch() done, defined at <ipython-input-7-535a26aaaefe>:5> result='<!doctype ht.../body></html>'>, set())

您可以看到我们捕获了错误，并且进一步的调用仍然成功返回了 html。

我们可能真的应该捕捉到一个 OSError，因为 socket.error 是 A deprecated alias of OSError，因为 python 3.3：

async def fetch(session, url):
    with aiohttp.Timeout(10):
        try:
            async with session.get(url) as response:
                return await response.text()
        except OSError as e:
            print(e)

如果您还想检查响应是否为 200，请将您的 if 也放入 try 中，您可以使用 reason 属性获取更多信息：

async def fetch(session, url):
    with aiohttp.Timeout(10):
        try:
            async with session.get(url) as response:
                if response.status != 200:
                    print(response.reason)
                return await response.text()
        except OSError as e:
            print(e.strerror)

【讨论】：

非常感谢！两个很好的答案，我希望我可以同时选择。选择 @kwarunek 是因为它开箱即用，但是 +1，我会找到你的两个最佳答案来投票。 :)