asyncio.Semaphore: как ограничить одновременные запросы в Python

У вас есть список карточек, которые нужно получить из HTTP API. Последовательные запросы тратят время на ожидание ответа, поэтому вы запускаете несколько корутин. После этого API начинает отвечать ошибками или замедляется: одновременно к нему пришло больше обращений, чем он обслуживает. Нужна общая граница для всех корутин одного процесса.

Где поставить ограничение

asyncio.Semaphore(3) хранит три разрешения. Корутина входит в async with semaphore, когда получает одно из них. Пока она ждёт HTTP-ответ и читает его, разрешение занято; при выходе из блока освобождается, даже если внутри возникла ошибка. Остальные корутины могут быть созданы заранее, но до сетевого вызова они ждут своей очереди.

Семафор нужен именно вокруг всей операции, которая занимает место у внешнего сервиса. Если отпустить разрешение сразу после отправки запроса, но до чтения ответа, одновременно открытых обменов снова станет больше трёх. Один и тот же объект семафора передают всем задачам, обращающимся к этому API. Если создать новый семафор внутри каждой корутины, общего лимита не будет.

Без ограничения восемь задач могут дойти до сетевого ожидания одновременно. С семафором первые три занимают слоты, остальные ждут; освободившийся слот тут же получает следующая задача. Это не восемь отдельных потоков: event loop переключает корутины в точках await, пока сеть отвечает. Именно поэтому асинхронный код помогает использовать время ожидания, но не ускоряет тяжёлую вычислительную функцию внутри того же цикла.

Проверяем границу на измеримом примере

Для локальной проверки сеть заменена на asyncio.sleep(0.05): в этом месте настоящая корутина ожидала бы HTTP-ответ. Счётчик active увеличивается только после входа в ограниченный участок, а peak запоминает максимум. Между изменениями счётчиков нет await, поэтому в одном event loop другая корутина не вклинивается в этот короткий участок.

import asyncio

async def main():
    semaphore = asyncio.Semaphore(3)
    active = 0
    peak = 0

    async def fetch(item_id):
        nonlocal active, peak
        async with semaphore:
            active += 1
            peak = max(peak, active)
            try:
                await asyncio.sleep(0.05)  # имитация ожидания HTTP-ответа
                return f"item-{item_id}"
            finally:
                active -= 1

    result = await asyncio.gather(*(fetch(i) for i in range(8)))
    print("peak:", peak)
    print("result:", result)

asyncio.run(main())

Запуск печатает peak: 3 и восемь меток item-0 … item-7 в порядке входных аргументов. Семафор ограничивает одновременную работу, а asyncio.gather возвращает результаты в порядке переданных корутин, даже если отдельные операции завершаются в другом порядке. Блок finally здесь нужен для правильного счётчика при исключении; разрешение семафора освобождает сам async with.

В реальном коде замените await asyncio.sleep(...) на асинхронный HTTP-вызов и чтение ответа, оставив оба действия внутри блока. Сам по себе requests.get() сюда не подходит: он блокирует поток и не даёт event loop обслуживать другие корутины во время ожидания. Клиент должен предоставлять асинхронные операции. Отдельный таймаут нужен для обращения, которое зависло: иначе оно будет долго удерживать один из трёх слотов.

Почему три одновременно не означают три в секунду

Семафор считает активные операции, а не число стартов за интервал времени. Если каждый ответ приходит за 50 миллисекунд, освободившиеся слоты быстро займут новые задачи; за секунду может начаться намного больше трёх запросов. Когда API ограничивает именно количество запросов в секунду, нужен отдельный механизм учёта времени или серверный заголовок с правилами квоты. Сначала уточните, какое ограничение вам выдали: одновременность или частоту.

Лимит действует там, где разделяется один объект семафора. Если четыре процесса запускают по три запроса одновременно, внешний API может увидеть до двенадцати обращений. Для общего лимита нескольких воркеров потребуется координация между процессами или ограничение на стороне сервиса. В рамках одного event loop этот пример проверяет именно три активных операции, а не общую квоту всей системы.

Есть и граница по памяти. Выражение с gather удобно для восьми карточек, но создание миллионов корутин сразу потребует память ещё до выполнения HTTP. Для большого входного потока используйте ограниченное число воркеров и asyncio.Queue(maxsize=...): когда очередь заполнится, добавление новых элементов подождёт. Семафор при этом по-прежнему управляет одновременным доступом к API; очередь удерживает размер ожидающей работы.

Как углубить асинхронную практику

В платном курсе «Python для профессионалов» вы пройдёте async/await и asyncio, разберёте пример асинхронной обработки веб-запросов, а затем научитесь профилировать время выполнения. Эти темы помогают понять, где программа действительно ждёт сеть и как измерять эффект от конкурентности. Откройте программу курса и открытый урок по asyncio.

Если сгенерированный помощником код использует async, но продолжает блокировать цикл событий, посмотрите разбор блокировки event loop. А обзор продвинутого курса Python поможет соотнести асинхронность с остальными темами обучения.