asyncio.timeout: где ловить TimeoutError и как проверить очистку

Операция должна прекратить ожидание, если ответ не готов к сроку. В код добавили asyncio.timeout, но обработчик TimeoutError внутри блока не вызывается. Причина в том, как контекст превращает внутреннюю отмену во внешнюю ошибку времени.

Отмена происходит внутри, TimeoutError появляется снаружи

Контекст asyncio.timeout отменяет текущую задачу по истечении срока. Находящаяся внутри корутина получает CancelledError в точке await и выполняет finally. При выходе из контекста его собственная отмена преобразуется в TimeoutError. Поэтому обычное место обработки TimeoutError находится вокруг async with, а не внутри ожидающей операции.

Это различие важно для очистки. Низкоуровневая функция освобождает свой ресурс в finally и не решает, каким был общий бюджет вызывающего кода. Верхний уровень сообщает об истечении срока и выбирает дальнейшее действие. Так проверка времени не размножается в каждой вложенной функции.

Проверить без сетевого сервиса

Для воспроизводимого примера задаём срок в прошлом через timeout_at. Контекст всё равно даёт корутине дойти до точки ожидания, затем отменяет её на следующем обороте цикла событий. Событие внутри никогда не устанавливается. Проверки подтверждают, что ошибка поймана снаружи и очистка уже выполнена.

Укажите Python 3.11 или новее. В рабочем вызове срок обычно вычисляют как loop.time() плюс доступный бюджет. Это монотонные часы цикла событий, а не календарное время из datetime.now.

import asyncio

async def main():
    cleaned = []
    async def wait_for_data():
        try:
            await asyncio.Event().wait()
        finally:
            cleaned.append("closed")
    timed_out = False
    deadline = asyncio.get_running_loop().time() - 1
    try:
        async with asyncio.timeout_at(deadline):
            await wait_for_data()
    except TimeoutError:
        timed_out = True
    assert timed_out
    assert cleaned == ["closed"]
    print("timeout handled after cleanup")

asyncio.run(main())

Время ожидания и завершение работы

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

В обработчике истечения срока возвращайте результат, соответствующий договорённости вызывающего кода. Не превращайте все исключения в сообщение «долго»: ValueError или ошибка чтения до срока должны остаться различимыми. Узкий except TimeoutError сохраняет эту диагностику.

Закрепить ожидаемую ошибку

В открытом уроке pytest есть практика проверки исключений. Сначала закрепите обычный отказ функции, затем перенесите тот же принцип на внешний TimeoutError и состояние после очистки. Посмотрите программу курса, чтобы развить тестируемую структуру Python-кода.