Теория: Основы async/await

Определение корутин (async def)

Когда в Python появилась поддержка ключевых слов async и await, это стало большим шагом вперёд для работы с асинхронным кодом. До этого приходилось использовать генераторы и специальные библиотеки, что выглядело громоздко. Теперь же есть встроенный синтаксис, понятный даже новичкам.

Однако просто наличие ключевых слов async и await недостаточно, чтобы код автоматически стал асинхронным и конкурентным. Для того чтобы воспользоваться преимуществами новой модели выполнения, нужно понять, что такое корутина и как с ней работать.

Корутина — это особая функция, которая может быть приостановлена в точках await и позже возобновлена. В отличие от обычной функции, её выполнение может прерываться, позволяя выполняться другим задачам во время ожидания. Это позволяет писать асинхронный код в привычном стиле, похожем на последовательный.

Чтобы определить корутину, используется ключевое слово async. Например:

import asyncio


async def say_hello() -> None:
    print("Привет")

Здесь import asyncio подключает встроенную библиотеку Python для работы с асинхронным кодом. Она предоставляет инструменты для запуска и управления корутинами, включая функцию asyncio.run(), которая понадобится нам для выполнения асинхронного кода.

Если вызвать эту функцию напрямую, она не выполнится сразу. Вместо результата вы получите объект корутины:

import asyncio


async def say_hello() -> None:
    print("Привет")


coro = say_hello()
print(coro)
# => <coroutine object say_hello at 0x...>

При выводе вы увидите что-то вроде <coroutine object say_hello at 0x...>. Это намекает, что функция определена, но ещё не запущена. Управление её выполнением берёт на себя механизм asyncio, о котором мы поговорим чуть позже.

Важно: Если корутина создана, но никогда не была ожидаема (не было вызова await), Python выдаст предупреждение RuntimeWarning: coroutine 'say_hello' was never awaited. Это сигнал, что вы забыли запустить асинхронную функцию.

Использование await для переключения задач

Ключевое слово await можно понимать как «подожди выполнения этой операции, а пока займись чем-то другим». Оно применяется только внутри корутин. С его помощью одна корутина может уступить управление другой, не блокируя весь процесс.

Точки ожидания — это места в коде, где стоит await. Именно в этих точках корутина может быть приостановлена, и управление передаётся event loop. Если в корутине нет ни одного await, она выполняется синхронно от начала до конца, без возможности переключения на другие задачи.

Представим простой пример с искусственной задержкой. В стандартной библиотеке Python есть функция asyncio.sleep(), которая делает паузу, но не блокирует выполнение других задач.

import asyncio


async def step_one() -> None:
    print("Шаг 1 начался")
    await asyncio.sleep(2)  # Точка ожидания
    print("Шаг 1 завершился")


async def step_two() -> None:
    print("Шаг 2 начался")
    await asyncio.sleep(1)  # Точка ожидания
    print("Шаг 2 завершился")

Если запустить каждую из этих функций по очереди, то внутри каждой будет пауза. Но важно, что эта пауза не блокирует всё приложение. Пока первая функция ждёт, управление может перейти ко второй.

Таким образом, await — это не просто ожидание, а переключение. Он говорит интерпретатору: «Здесь я временно не нужен, можешь выполнить что-то другое».

Благодаря этому можно работать с большим количеством операций ввода-вывода, которые обычно медленные. Например, получать данные сразу с десятков сайтов. Если бы всё выполнялось синхронно, пришлось бы ждать каждый запрос. С await программа обрабатывает все запросы почти одновременно, экономя кучу времени.

Запуск программ с asyncio.run

Чтобы асинхронная программа действительно выполнилась, нужно её запустить. Если вызвать корутину напрямую, она не начнётся, а просто вернёт объект, который хранит план выполнения. Для запуска используется функция asyncio.run(). Она принимает корутину, выполняет её до конца и возвращает результат.

Простейший пример:

import asyncio


async def main() -> None:
    print("Начало работы")
    await asyncio.sleep(1)
    print("Завершение работы")


asyncio.run(main())

Здесь asyncio.run(main()) гарантирует, что программа начнётся, выполнит все шаги и завершится. Если убрать этот вызов, ничего бы не произошло.

Важно понимать взаимодействие async и await. Если функция объявлена через async def, но внутри неё нет await, она работает как обычная синхронная функция, только с другим синтаксисом.

Пример:

import asyncio


async def no_await() -> None:
    print("Я выполняюсь сразу, без ожиданий")


async def main() -> None:
    await no_await()


asyncio.run(main())

Выполнение происходит мгновенно, потому что внутри нет мест для переключения — нет точек ожидания.

Когда в корутине есть await, ситуация меняется. Такой код может временно уступать управление, и другие задачи смогут выполняться конкурентно, не блокируя друг друга во время операций ожидания.

Пример:

import asyncio


async def async_task(n: int) -> None:
    print(f"Задача {n} началась")
    await asyncio.sleep(2)
    print(f"Задача {n} завершилась")


async def main() -> None:
    await async_task(0)
    await async_task(1)
    await async_task(2)


asyncio.run(main())

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

Таким образом, asyncio.run() — это главный инструмент для запуска асинхронного кода. Он показывает, как связаны между собой ключевые слова async и await, и позволяет проверить их работу на практике. Даже если пока всё выглядит как последовательное выполнение, мы получаем фундамент для построения более сложных асинхронных сценариев.

Типичные ошибки при работе с async/await

При изучении асинхронного программирования часто допускаются следующие ошибки:

Забыли await перед вызовом корутины

import asyncio


async def fetch_data() -> str:
    await asyncio.sleep(1)
    return "Данные"


async def main() -> None:
    # Плохо: забыли await
    result = fetch_data()
    print(result)
    # => <coroutine object fetch_data at 0x...>
    # RuntimeWarning: coroutine 'fetch_data' was never awaited


asyncio.run(main())

Правильный вариант:

import asyncio


async def fetch_data() -> str:
    await asyncio.sleep(1)
    return "Данные"


async def main() -> None:
    # Хорошо: используем await
    result = await fetch_data()
    print(result)
    # => Данные


asyncio.run(main())

Попытка вызвать корутину из синхронного кода

import asyncio


async def async_function() -> None:
    await asyncio.sleep(1)


# Плохо: вызов корутины без запуска event loop
def sync_function() -> None:
    async_function()  # Не будет выполнено!

Правильный вариант — использовать asyncio.run() для запуска:

import asyncio


async def async_function() -> None:
    await asyncio.sleep(1)


def sync_function() -> None:
    asyncio.run(async_function())  # Правильно

Попытка запустить asyncio.run внутри уже работающего event loop

import asyncio


async def inner() -> None:
    await asyncio.sleep(1)


async def outer() -> None:
    # Плохо: нельзя вызывать asyncio.run внутри корутины
    asyncio.run(
        inner()
    )  # RuntimeError: asyncio.run() cannot be called from a running event loop


asyncio.run(outer())

Правильный вариант — просто использовать await:

import asyncio


async def inner() -> None:
    await asyncio.sleep(1)


async def outer() -> None:
    # Хорошо: просто await
    await inner()


asyncio.run(outer())

Асинхронные контекстные менеджеры и итераторы

Помимо базовых корутин, Python поддерживает асинхронные версии привычных конструкций.

Асинхронные контекстные менеджеры позволяют выполнять операции инициализации и очистки в асинхронном контексте. Они используют async with вместо обычного with:

import asyncio
from types import TracebackType
from typing import Self


class AsyncResource:
    async def __aenter__(self) -> Self:
        print("Открытие ресурса")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        print("Закрытие ресурса")
        await asyncio.sleep(0.1)


async def main() -> None:
    async with AsyncResource() as resource:
        print("Работа с ресурсом")


asyncio.run(main())

Асинхронные итераторы позволяют перебирать элементы, каждый из которых получается асинхронно. Они используют async for:

import asyncio
from typing import Self


class AsyncRange:
    def __init__(self, n: int) -> None:
        self.n = n
        self.i = 0

    def __aiter__(self) -> Self:
        return self

    async def __anext__(self) -> int:
        if self.i >= self.n:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.i += 1
        return self.i


async def main() -> None:
    async for number in AsyncRange(5):
        print(number)


asyncio.run(main())

Для создания асинхронных контекстных менеджеров можно использовать декоратор @asynccontextmanager из модуля contextlib:

import asyncio
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager


@asynccontextmanager
async def async_resource() -> AsyncGenerator[str, None]:
    print("Открытие ресурса")
    await asyncio.sleep(0.1)
    try:
        yield "ресурс"
    finally:
        print("Закрытие ресурса")
        await asyncio.sleep(0.1)


async def main() -> None:
    async with async_resource() as resource:
        print(f"Работа с {resource}")


asyncio.run(main())

Более подробно об этих конструкциях можно прочитать в документации Python.

Контрольный список при работе с async/await

Вызывайте asyncio.run() только один раз на верхнем уровне программы — это точка входа в асинхронный код

Не запускайте asyncio.run() внутри уже работающего event loop — используйте await для вызова корутин изнутри других корутин

Всегда используйте await перед вызовом корутины — иначе получите объект корутины вместо результата и предупреждение от Python

Импортируйте asyncio в каждом файле, где используете асинхронный код — не полагайтесь на неявные импорты

Используйте async with для асинхронных контекстных менеджеров — это гарантирует правильное освобождение ресурсов даже при ошибках

Используйте async for для асинхронных итераторов — обычный for не будет работать с асинхронными итераторами# Основы async/await