Aiohttp.TCPConnector(с предельным аргументом) против asyncio.Semaphore для ограничения количества одновременных соединений

Я думал, что хотел бы изучить новый синтаксис python async и, в частности, асинхронный модуль, создав простой script, который позволит вам загружать несколько ресурсов на одном.

Но теперь я застрял.

Во время исследования я столкнулся с двумя вариантами ограничения количества одновременных запросов:

• Передача aiohttp.TCPConnector(с предельным аргументом) на aiohttp.ClientSession или
• Использование asyncio.Semaphore.

Есть ли предпочтительный вариант или они могут использоваться взаимозаменяемо, если вы хотите ограничить количество одновременных подключений? Являются ли (примерно) равными по производительности?

Также оба имеют значение по умолчанию 100 одновременных подключений/операций. Если я использую только Семафор с пределом допустимости, скажем, что 500 будет внутренними aiohttp блокировать меня до 100 одновременных соединений неявно?

Это все очень новое и непонятное для меня. Пожалуйста, не стесняйтесь указывать на какие-либо недоразумения с моей стороны или недостатки в моем коде.

Вот мой код, содержащий в настоящее время оба параметра (который следует удалить?):

Бонусные вопросы:

• Как мне обрабатывать (желательно повторить x раз) coros, которые вызывают ошибку?
• Каков наилучший способ сохранить возвращаемые данные (сообщите мой DataHandler), как только закончится сверление? Я не хочу, чтобы все это было сохранено в конце, потому что я мог как можно скорее начать работу с результатами.

s

import asyncio
from tqdm import tqdm
import uvloop as uvloop
from aiohttp import ClientSession, TCPConnector, BasicAuth

# You can ignore this class
class DummyDataHandler(DataHandler):
    """Takes data and stores it somewhere"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def take(self, origin_url, data):
        return True

    def done(self):
        return None

class AsyncDownloader(object):
    def __init__(self, concurrent_connections=100, silent=False, data_handler=None, loop_policy=None):

        self.concurrent_connections = concurrent_connections
        self.silent = silent

        self.data_handler = data_handler or DummyDataHandler()

        self.sending_bar = None
        self.receiving_bar = None

        asyncio.set_event_loop_policy(loop_policy or uvloop.EventLoopPolicy())
        self.loop = asyncio.get_event_loop()
        self.semaphore = asyncio.Semaphore(concurrent_connections)

    async def fetch(self, session, url):
        # This is option 1: The semaphore, limiting the number of concurrent coros,
        # thereby limiting the number of concurrent requests.
        with (await self.semaphore):
            async with session.get(url) as response:
                # Bonus Question 1: What is the best way to retry a request that failed?
                resp_task = asyncio.ensure_future(response.read())
                self.sending_bar.update(1)
                resp = await resp_task

                await  response.release()
                if not self.silent:
                    self.receiving_bar.update(1)
                return resp

    async def batch_download(self, urls, auth=None):
        # This is option 2: Limiting the number of open connections directly via the TCPConnector
        conn = TCPConnector(limit=self.concurrent_connections, keepalive_timeout=60)
        async with ClientSession(connector=conn, auth=auth) as session:
            await asyncio.gather(*[asyncio.ensure_future(self.download_and_save(session, url)) for url in urls])

    async def download_and_save(self, session, url):
        content_task = asyncio.ensure_future(self.fetch(session, url))
        content = await content_task
        # Bonus Question 2: This is blocking, I know. Should this be wrapped in another coro
        # or should I use something like asyncio.as_completed in the download function?
        self.data_handler.take(origin_url=url, data=content)

    def download(self, urls, auth=None):
        if isinstance(auth, tuple):
            auth = BasicAuth(*auth)
        print('Running on concurrency level {}'.format(self.concurrent_connections))
        self.sending_bar = tqdm(urls, total=len(urls), desc='Sent    ', unit='requests')
        self.sending_bar.update(0)

        self.receiving_bar = tqdm(urls, total=len(urls), desc='Reveived', unit='requests')
        self.receiving_bar.update(0)

        tasks = self.batch_download(urls, auth)
        self.loop.run_until_complete(tasks)
        return self.data_handler.done()


### call like so ###

URL_PATTERN = 'https://www.example.com/{}.html'

def gen_url(lower=0, upper=None):
    for i in range(lower, upper):
        yield URL_PATTERN.format(i)   

ad = AsyncDownloader(concurrent_connections=30)
data = ad.download([g for g in gen_url(upper=1000)])