aiohttp: установить максимальное количество запросов в секунду

Начиная с версии 2.0, при использовании ClientSession aiohttp автоматически ограничивает количество одновременных подключений до 100.

Вы можете изменить ограничение, создав свой собственный TCPConnector и передав его в ClientSession. Например, для создания клиента до 50 одновременных запросов:

import aiohttp

connector = aiohttp.TCPConnector(limit=50)
client = aiohttp.ClientSession(connector=connector)

В случае, если он лучше подходит для вашего варианта использования, есть также параметр limit_per_host (который по умолчанию отключен), который вы можете передать, чтобы ограничить количество одновременных подключений к одной и той же «конечной точке». Согласно документам:

limit_per_host (int) - лимит одновременных подключений к одной конечной точке. Конечные точки одинаковы, если они имеют равную (host, port, is_ssl) тройку.

Пример использования:

import aiohttp

connector = aiohttp.TCPConnector(limit_per_host=50)
client = aiohttp.ClientSession(connector=connector)

Я нашел здесь одно возможное решение: http://compiletoi.net/fast-scraping-in-python-with-asyncio.html

Выполнять 3 запроса одновременно - это круто, но делать 5000 - не так хорошо. Если вы попытаетесь выполнить слишком много запросов одновременно, соединения могут начать закрываться, или вас могут даже забанить на сайте.

Чтобы этого избежать, можно использовать семафор. Это инструмент синхронизации, который можно использовать для ограничения количества сопрограмм, которые в какой-то момент что-то делают. Мы просто создадим семафор перед созданием цикла, передав в качестве аргумента количество одновременных запросов, которые мы хотим разрешить:

sem = asyncio.Semaphore(5)

Затем мы просто заменяем:

page = yield from get(url, compress=True)

тем же, но защищенным семафором:

with (yield from sem):
    page = yield from get(url, compress=True)

Это обеспечит одновременное выполнение не более 5 запросов.

Вы можете установить задержку для каждого запроса или сгруппировать URL-адреса в пакеты и регулировать пакеты для достижения желаемой частоты.

1. Задержка по запросу

Заставить скрипт ждать между запросами, используя asyncio.sleep

import asyncio
import aiohttp

delay_per_request = 0.5
urls = [
   # put some URLs here...
]

async def app():
    tasks = []
    for url in urls:
        tasks.append(asyncio.ensure_future(make_request(url)))
        await asyncio.sleep(delay_per_request)

    results = await asyncio.gather(*tasks)
    return results

async def make_request(url):
    print('$$$ making request')
    async with aiohttp.ClientSession() as sess:
        async with sess.get(url) as resp:
            status = resp.status
            text = await resp.text()
            print('### got page data')
            return url, status, text

Это можно запустить, например, results = asyncio.run(app()).

2. Дроссель дозирования

Используя make_request из приведенного выше, вы можете запрашивать и ограничивать пакеты URL-адресов следующим образом:

import asyncio
import aiohttp
import time

max_requests_per_second = 0.5
urls = [[
   # put a few URLs here...
],[
   # put a few more URLs here...
]]

async def app():
    results = []
    for i, batch in enumerate(urls):
        t_0 = time.time()
        print(f'batch {i}')
        tasks = [asyncio.ensure_future(make_request(url)) for url in batch]
        for t in tasks:
            d = await t
            results.append(d)
        t_1 = time.time()

        # Throttle requests
        batch_time = (t_1 - t_0)
        batch_size = len(batch)
        wait_time = (batch_size / max_requests_per_second) - batch_time
        if wait_time > 0:
            print(f'Too fast! Waiting {wait_time} seconds')
            time.sleep(wait_time)

    return results

Опять же, это можно запустить с asyncio.run(app()).

Это пример без aiohttp, но вы можете обернуть любой асинхронный метод или aiohttp.request с помощью декоратора Limit

import asyncio
import time


class Limit(object):
    def __init__(self, calls=5, period=1):
        self.calls = calls
        self.period = period
        self.clock = time.monotonic
        self.last_reset = 0
        self.num_calls = 0

    def __call__(self, func):
        async def wrapper(*args, **kwargs):
            if self.num_calls >= self.calls:
                await asyncio.sleep(self.__period_remaining())

            period_remaining = self.__period_remaining()

            if period_remaining <= 0:
                self.num_calls = 0
                self.last_reset = self.clock()

            self.num_calls += 1

            return await func(*args, **kwargs)

        return wrapper

    def __period_remaining(self):
        elapsed = self.clock() - self.last_reset
        return self.period - elapsed


@Limit(calls=5, period=2)
async def test_call(x):
    print(x)


async def worker():
    for x in range(100):
        await test_call(x + 1)


asyncio.run(worker())