async и await в Python: asyncio для начинающих

async и await в Python — синтаксис асинхронного кода: async def объявляет корутину, await приостанавливает её до получения результата. Пока корутина ждёт, цикл событий модуля asyncio выполняет другие корутины, поэтому программа ждёт несколько сетевых ответов одновременно. Минимальный пример: две «загрузки» по 1 и 2 секунды завершаются за 2 секунды вместо 3.

import asyncio
import time


async def fetch(name, delay):
    print(f"{name}: начало")
    await asyncio.sleep(delay)  # имитация сетевого запроса
    print(f"{name}: готово")
    return f"{name} за {delay} с"


async def main():
    start = time.perf_counter()
    results = await asyncio.gather(fetch("A", 1), fetch("B", 2))
    print(results)
    print(f"Всего: {time.perf_counter() - start:.1f} с")


asyncio.run(main())

A: начало
B: начало
A: готово
B: готово
['A за 1 с', 'B за 2 с']
Всего: 2.0 с

Ниже — как устроены корутины, как запускать задачи параллельно через gather и TaskGroup, ставить таймауты, что делать с блокирующими библиотеками и когда async вообще не нужен. Примеры проверены на Python 3.12, 3.14 и 3.15; время выполнения приведено для примера и на вашем компьютере может немного отличаться. Скрипты запускаются как обычно: py script.py в Windows или python3 script.py в macOS и Linux.

Синхронный и асинхронный код: в чём разница

Обычный (синхронный) код выполняет строки по очереди. Если функция ждёт ответа сервера, вся программа стоит:

import time


def fetch(name, delay):
    time.sleep(delay)  # программа просто ждёт
    return name


start = time.perf_counter()
fetch("A", 1)
fetch("B", 2)
print(f"Всего: {time.perf_counter() - start:.1f} с")  # → Всего: 3.0 с

В асинхронном коде ожидание не простаивает. Работой управляет цикл событий (event loop) — диспетчер внутри asyncio. Когда корутина доходит до await и ждёт сеть, таймер или диск, она возвращает управление циклу, и тот запускает следующую готовую корутину. Когда ответ пришёл, первая продолжает работу с того же места.

Всё это происходит в одном потоке. Корутины переключаются только в точках await, поэтому такую модель называют кооперативной многозадачностью: каждая корутина сама «уступает очередь». Отсюда два следствия, на которых строится вся статья:

• async ускоряет программы, которые много ждут: HTTP-запросы, базы данных, сокеты, чат-боты, парсеры.
• Код без await (вычисления или блокирующий вызов) не отдаёт управление и останавливает все остальные корутины.

Корутины: async def и объект корутины

Функция, объявленная через async def, называется корутинной функцией. Её вызов только создаёт объект корутины — «отложенную» работу. Тело функции выполнится, когда корутину запустят через await или цикл событий:

import asyncio


async def greet():
    return "привет"


coro = greet()
print(type(coro))          # → <class 'coroutine'>
print(asyncio.run(coro))   # → привет

Если вызвать корутину и забыть await, код внутри так и не запустится. Python сообщит об этом предупреждением:

async def save():
    print("сохраняю")


async def main():
    save()  # забыли await


asyncio.run(main())

RuntimeWarning: coroutine 'save' was never awaited

await разрешён только внутри async def. В обычной функции получите синтаксическую ошибку:

SyntaxError: 'await' outside async function

Помимо async def есть ещё две асинхронные конструкции. async with работает с асинхронными менеджерами контекста: сессиями HTTP-клиентов, соединениями с базой, семафорами. async for перебирает асинхронные итераторы, например асинхронный генератор — функцию async def с yield (как работает обычный yield, разобрано в статье о генераторах Python):

import asyncio


async def ticks(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i  # асинхронный генератор


async def main():
    async for value in ticks(3):
        print(value)  # → 0, затем 1, затем 2
    squares = [v * v async for v in ticks(4)]
    print(squares)    # → [0, 1, 4, 9]


asyncio.run(main())

asyncio.run: точка входа в программу

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

Внутри уже работающего цикла asyncio.run() вызвать нельзя:

RuntimeError: asyncio.run() cannot be called from a running event loop

Эту ошибку чаще всего видят в Jupyter Notebook: там цикл событий уже запущен, поэтому в ячейке пишут просто await main() без asyncio.run(). В обычной консоли Python поэкспериментировать с await на верхнем уровне можно в асинхронном REPL: py -m asyncio в Windows или python3 -m asyncio в macOS и Linux.

Важно: в старых статьях встречается связка loop = asyncio.get_event_loop() и loop.run_until_complete(main()), а иногда и декоратор @asyncio.coroutine с yield from. Декоратор удалён в Python 3.11, а get_event_loop() без запущенного цикла с Python 3.14 выбрасывает RuntimeError: There is no current event loop in thread 'MainThread'. Используйте asyncio.run(), а внутри корутин — asyncio.get_running_loop().

await: ожидание результата

await можно поставить перед awaitable-объектом — тем, что умеет ждать: корутиной, задачей (Task) или футурой (Future, низкоуровневый «результат в будущем»). await возвращает результат корутины или выбрасывает её исключение, поэтому ошибки ловят обычным try/except.

Главная ловушка новичков — последовательные await. Каждый из них ждёт завершения, прежде чем следующая строка запустит новую работу:

import asyncio
import time


async def fetch(delay):
    await asyncio.sleep(delay)
    return delay


async def main():
    start = time.perf_counter()
    a = await fetch(1)  # ждём первую...
    b = await fetch(2)  # ...и только потом запускаем вторую
    print(a, b, f"{time.perf_counter() - start:.1f} с")  # → 1 2 3.0 с


asyncio.run(main())

Код асинхронный, а выигрыша нет: 3 секунды, как в синхронной версии. Чтобы операции шли одновременно, их нужно запустить как задачи — об этом следующий раздел.

Как запустить корутины одновременно

Задача (asyncio.Task) — корутина, которую цикл событий уже взял в работу. Она выполняется в фоне, пока текущая корутина делает что-то ещё. Создать задачи можно тремя основными способами.

asyncio.gather: дождаться списка корутин

gather() оборачивает каждую корутину в задачу, ждёт все и возвращает список результатов в порядке аргументов, даже если вторая корутина завершилась раньше первой. Пример в начале статьи использует именно его. Для списка корутин распакуйте его звёздочкой: await asyncio.gather(*coros).

По умолчанию первое исключение сразу передаётся в await, а остальные корутины продолжают работать. С return_exceptions=True исключения попадают в список результатов наравне с обычными значениями:

import asyncio


async def fetch(n):
    await asyncio.sleep(0.1)
    if n == 2:
        raise ValueError(f"ошибка в задаче {n}")
    return n


async def main():
    results = await asyncio.gather(
        fetch(1), fetch(2), fetch(3), return_exceptions=True
    )
    print(results)  # → [1, ValueError('ошибка в задаче 2'), 3]


asyncio.run(main())

asyncio.create_task: запустить и продолжить работу

create_task() ставит корутину в цикл событий сразу, а результат можно забрать позже:

import asyncio
import time


async def fetch(delay):
    await asyncio.sleep(delay)
    return delay


async def main():
    start = time.perf_counter()
    task_a = asyncio.create_task(fetch(1))  # задача стартует сразу
    task_b = asyncio.create_task(fetch(2))
    a = await task_a
    b = await task_b
    print(a, b, f"{time.perf_counter() - start:.1f} с")  # → 1 2 2.0 с


asyncio.run(main())

Сравните с примером из предыдущего раздела: те же два await, но задачи уже работают параллельно, поэтому 2 секунды вместо 3.

Важно: сохраняйте ссылку на задачу в переменной или коллекции. Цикл событий хранит задачи по слабым ссылкам, и задача без ссылки может быть удалена сборщиком мусора до завершения.

asyncio.TaskGroup (Python 3.11+): рекомендуемый способ

TaskGroup — асинхронный менеджер контекста: задачи создаются внутри блока async with, а выход из блока ждёт их все.

import asyncio


async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"{name}: {delay} с"


async def main():
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(fetch("A", 1))
        task_b = tg.create_task(fetch("B", 2))
    # сюда попадаем, когда обе задачи завершились
    print(task_a.result(), task_b.result())  # → A: 1 с B: 2 с


asyncio.run(main())

Главное отличие от gather — обработка ошибок. Если одна задача группы падает, TaskGroup отменяет остальные и выбрасывает все ошибки разом в виде ExceptionGroup. Ловят его конструкцией except*:

import asyncio


async def slow():
    try:
        await asyncio.sleep(5)
        print("slow завершилась")
    except asyncio.CancelledError:
        print("slow отменена")
        raise


async def broken():
    await asyncio.sleep(0.1)
    raise ValueError("сервер вернул ошибку")


async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(slow())
            tg.create_task(broken())
    except* ValueError as eg:
        for exc in eg.exceptions:
            print("Ошибка:", exc)


asyncio.run(main())

slow отменена
Ошибка: сервер вернул ошибку

Отменённая задача получает исключение asyncio.CancelledError в точке await. Если перехватываете его ради очистки ресурсов, обязательно пробрасывайте дальше через raise, как в примере: на отмене построены TaskGroup и таймауты. Подробнее об исключениях и except* — в статье о try/except в Python.

В Python 3.15 у группы появился метод tg.cancel(): он досрочно отменяет все задачи, и блок завершается без исключения. Пример — взять ответ от самого быстрого зеркала:

import asyncio


async def mirror(tg, name, delay, found):
    await asyncio.sleep(delay)
    found.append(name)
    tg.cancel()  # Python 3.15+: остановить всю группу


async def main():
    found = []
    async with asyncio.TaskGroup() as tg:
        tg.create_task(mirror(tg, "зеркало 1", 0.3, found))
        tg.create_task(mirror(tg, "зеркало 2", 0.1, found))
        tg.create_task(mirror(tg, "зеркало 3", 0.5, found))
    print("Первым ответило:", found[0])  # → Первым ответило: зеркало 2


asyncio.run(main())

В Python 3.14 и раньше этот код упадёт с AttributeError: 'TaskGroup' object has no attribute 'cancel'.

Результаты по мере готовности: as_completed

Если результат нужно обрабатывать сразу, как только он готов, используйте asyncio.as_completed():

import asyncio


async def fetch(name, delay):
    await asyncio.sleep(delay)
    return name


async def main():
    coros = [fetch("медленный", 2), fetch("быстрый", 0.5), fetch("средний", 1)]
    for next_done in asyncio.as_completed(coros):
        print(await next_done)  # → быстрый, затем средний, затем медленный


asyncio.run(main())

Что выбрать

Способ	Порядок результатов	Если задача упала	Когда использовать
asyncio.TaskGroup	через task.result() каждой задачи	остальные отменяются, ошибки собираются в ExceptionGroup	основной выбор в Python 3.11+
asyncio.gather	список в порядке аргументов	первое исключение уходит в await, остальные задачи работают дальше	нужен простой список результатов; return_exceptions=True для сбора ошибок
asyncio.create_task	забираете сами через await task	исключение придёт при await task	фоновая работа, пока корутина занята другим
asyncio.as_completed	в порядке завершения	исключение при await очередного результата	обработать каждый ответ сразу

Как ограничить число одновременных задач

Запустить тысячу запросов одним gather легко, но сервер может ответить ошибкой 429 или заблокировать вас. Ограничьте конкурентность семафором asyncio.Semaphore: внутри async with sem одновременно находится не больше заданного числа корутин.

import asyncio
import time


async def fetch(i, sem):
    async with sem:  # одновременно внутри не больше 3 задач
        await asyncio.sleep(1)
        return i


async def main():
    sem = asyncio.Semaphore(3)
    start = time.perf_counter()
    results = await asyncio.gather(*(fetch(i, sem) for i in range(9)))
    print(results, f"{time.perf_counter() - start:.1f} с")


asyncio.run(main())

[0, 1, 2, 3, 4, 5, 6, 7, 8] 3.0 с

Девять «запросов» по секунде прошли тремя волнами по три — 3 секунды.

Таймауты: asyncio.timeout

Сетевой вызов может зависнуть надолго, поэтому ожидание стоит ограничивать. С Python 3.11 для этого есть менеджер контекста asyncio.timeout(): если блок не уложился во время, текущая операция отменяется и выбрасывается встроенное исключение TimeoutError.

import asyncio


async def slow_request():
    await asyncio.sleep(10)
    return "ответ"


async def main():
    try:
        async with asyncio.timeout(2):
            result = await slow_request()
    except TimeoutError:
        print("Сервер не ответил за 2 секунды")


asyncio.run(main())

Один таймаут покрывает весь блок: несколько await внутри делят общие 2 секунды. Для абсолютного дедлайна есть asyncio.timeout_at(when), где when — время по часам цикла событий (loop.time()).

Для одной корутины подходит и asyncio.wait_for(), который встречается в коде, написанном до Python 3.11:

try:
    result = await asyncio.wait_for(slow_request(), timeout=2)
except TimeoutError:
    print("Таймаут")

До Python 3.11 wait_for() выбрасывал отдельное исключение asyncio.TimeoutError. Теперь это синоним встроенного TimeoutError, поэтому except TimeoutError ловит оба варианта.

Когда async ускоряет код, а когда нет

asyncio выигрывает время только там, где программа ждёт внешний мир: сеть, базу данных, диск, ответ другого процесса. Пока одна корутина ждёт, работают другие.

Вычисления на процессоре (CPU-bound задачи) asyncio не ускоряет: корутина без await занимает поток целиком, и остальные ждут её завершения. Две «асинхронные» функции подсчёта суммы работают столько же, сколько два обычных вызова:

import asyncio
import time


def count(n):
    total = 0
    for i in range(n):
        total += i
    return total


async def cpu_task(n):
    return count(n)  # внутри нет await: цикл событий занят


async def main():
    start = time.perf_counter()
    count(20_000_000)
    count(20_000_000)
    print(f"Обычные вызовы: {time.perf_counter() - start:.1f} с")

    start = time.perf_counter()
    await asyncio.gather(cpu_task(20_000_000), cpu_task(20_000_000))
    print(f"gather: {time.perf_counter() - start:.1f} с")


asyncio.run(main())

Обычные вызовы: 1.9 с
gather: 2.0 с

Для тяжёлых вычислений нужны отдельные процессы: multiprocessing или concurrent.futures.ProcessPoolExecutor. Из async-кода пул процессов вызывают через loop.run_in_executor():

import asyncio
from concurrent.futures import ProcessPoolExecutor


def count(n):
    total = 0
    for i in range(n):
        total += i
    return total


async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        results = await asyncio.gather(
            loop.run_in_executor(pool, count, 10_000_000),
            loop.run_in_executor(pool, count, 10_000_000),
        )
    print(results)  # → [49999995000000, 49999995000000]


if __name__ == "__main__":
    asyncio.run(main())

Проверка if __name__ == "__main__" здесь обязательна: в Windows и macOS дочерние процессы заново импортируют модуль, и без неё скрипт запустит сам себя повторно. Выигрыш от процессов появляется на нескольких ядрах и долгих вычислениях: запуск каждого процесса тоже занимает время.

Подход	Сколько ядер использует	Подходит для	Цена
asyncio	одно, один поток	тысячи одновременных сетевых операций	нужны асинхронные библиотеки, весь путь вызовов через async/await
threading, asyncio.to_thread	одно для Python-кода из-за GIL	десятки операций ввода-вывода через синхронные библиотеки	каждый поток занимает память, нужна синхронизация общих данных
multiprocessing, ProcessPoolExecutor	все	вычисления: обработка изображений, математика, парсинг больших файлов	дорогой запуск процесса, данные передаются между процессами копированием

GIL (Global Interpreter Lock) — блокировка интерпретатора CPython, из-за которой Python-код в один момент выполняет только один поток. С Python 3.14 официально поддерживается отдельная сборка без GIL (free-threaded), где потоки выполняют вычисления параллельно, но стандартный установщик ставит обычную сборку.

Блокирующие вызовы и asyncio.to_thread

Блокирующий вызов — обычная синхронная функция, которая ждёт: time.sleep(), requests.get(), драйвер базы данных без поддержки async, чтение большого файла. Внутри корутины он останавливает весь цикл событий. Посмотрите, как «тикер» должен печатать строку каждые 0,5 секунды, но замирает:

import asyncio
import time


async def ticker():
    for _ in range(3):
        print(f"тик {time.perf_counter() - START:.1f}")
        await asyncio.sleep(0.5)


async def bad_request():
    time.sleep(1.5)  # блокирующий вызов


async def main():
    await asyncio.gather(ticker(), bad_request())


START = time.perf_counter()
asyncio.run(main())

тик 0.0
тик 1.5
тик 2.0

Второй тик пришёл через 1,5 секунды вместо 0,5: всё это время цикл событий стоял. Слово async перед функцией ничего не меняет — если внутри нет await асинхронной операции, функция выполнится целиком, не отдавая управление.

Если асинхронного аналога у библиотеки нет, вынесите вызов в отдельный поток через asyncio.to_thread() (Python 3.9+):

import asyncio
import time


def blocking_io():
    time.sleep(1.5)  # например, requests.get() или чтение большого файла
    return "данные"


async def ticker():
    for _ in range(3):
        print(f"тик {time.perf_counter() - START:.1f}")
        await asyncio.sleep(0.5)


async def main():
    _, data = await asyncio.gather(ticker(), asyncio.to_thread(blocking_io))
    print(data, f"{time.perf_counter() - START:.1f} с")


START = time.perf_counter()
asyncio.run(main())

тик 0.0
тик 0.5
тик 1.0
данные 1.5 с

to_thread подходит для ожидания ввода-вывода. Для вычислений из-за GIL он не поможет — используйте пул процессов из предыдущего раздела.

Найти скрытые блокировки помогает режим отладки asyncio: asyncio.run(main(), debug=True), переменная окружения PYTHONASYNCIODEBUG=1 или запуск с -X dev. В нём цикл событий сообщает о шагах корутин дольше 100 мс, а к предупреждению о забытом await добавляет место, где создана корутина:

Executing <Task finished name='Task-1' coro=<bad_request() done, ...> ...> took 0.508 seconds

Асинхронные HTTP-запросы: httpx и aiohttp

Популярная библиотека requests синхронная: внутри корутины она блокирует цикл событий. Для асинхронных запросов берут httpx или aiohttp. Установка:

# Windows
py -m pip install httpx aiohttp

# macOS / Linux
python3 -m pip install httpx aiohttp

httpx: API как у requests

У httpx есть синхронный httpx.Client и асинхронный httpx.AsyncClient с почти одинаковыми методами, поэтому переход с requests проще всего. Пять запросов к тестовому сервису, каждый из которых отвечает через секунду:

import asyncio
import time

import httpx

URLS = [f"https://httpbin.org/delay/1?n={i}" for i in range(5)]


async def fetch(client, url):
    response = await client.get(url)
    response.raise_for_status()
    return response.json()["args"]["n"]


async def main():
    start = time.perf_counter()
    async with httpx.AsyncClient(timeout=10) as client:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(client, url)) for url in URLS]
    print([t.result() for t in tasks])
    print(f"5 запросов за {time.perf_counter() - start:.1f} с")


asyncio.run(main())

['0', '1', '2', '3', '4']
5 запросов за 2.5 с

Те же пять запросов через синхронный httpx.Client в цикле заняли около 8 секунд. Создавайте один клиент на серию запросов: он переиспользует соединения.

Совет: развитие httpx в последнее время замедлилось, и команда Pydantic выпустила его продолжение — httpx2 с тем же API. Пакет ставится командой pip install httpx2, в коде достаточно заменить импорт на import httpx2 as httpx. Пример выше с httpx2 2.13 работает без других изменений.

aiohttp: клиент и сервер

aiohttp — асинхронная библиотека, в которую входит и HTTP-клиент, и веб-сервер. Запросы выполняются внутри сессии ClientSession, а ответ читается через async with:

import asyncio
import time

import aiohttp

URLS = [f"https://httpbin.org/delay/1?n={i}" for i in range(5)]


async def fetch(session, url):
    async with session.get(url) as response:
        response.raise_for_status()
        data = await response.json()
        return data["args"]["n"]


async def main():
    start = time.perf_counter()
    timeout = aiohttp.ClientTimeout(total=10)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        results = await asyncio.gather(*(fetch(session, url) for url in URLS))
    print(results)
    print(f"5 запросов за {time.perf_counter() - start:.1f} с")


asyncio.run(main())

['0', '1', '2', '3', '4']
5 запросов за 2.3 с

Критерий	httpx	aiohttp
Синхронный API	есть (httpx.Client)	нет
Похожесть на requests	высокая	своя модель: сессия и async with для ответа
HTTP/2	есть, через pip install httpx[http2]	нет
Веб-сервер в комплекте	нет	есть
Типичный выбор	новичкам, проектам на FastAPI, коду с sync- и async-частями	парсерам с большим числом запросов, проектам, где aiohttp уже используется

Оба клиента подходят для парсеров и интеграций с API. Если в проекте уже есть один из них, второй ради скорости ставить не нужно.

async в FastAPI и Django

Веб-фреймворки — главное место, где начинающий встречает async def на практике.

FastAPI запускается на ASGI-сервере, например Uvicorn, и принимает оба вида обработчиков. async def выполняется прямо в цикле событий, обычный def — в пуле потоков. Пишите async def, если внутри есть await асинхронного клиента или драйвера БД:

import asyncio

import httpx
from fastapi import FastAPI

app = FastAPI()


@app.get("/rates")
async def rates():
    async with httpx.AsyncClient(timeout=5) as client:
        usd, eur = await asyncio.gather(
            client.get("https://httpbin.org/get?currency=USD"),
            client.get("https://httpbin.org/get?currency=EUR"),
        )
    return {"usd": usd.json()["args"], "eur": eur.json()["args"]}

Если обработчик вызывает синхронные библиотеки (requests, синхронный SQLAlchemy), объявляйте его через обычный def: блокирующий вызов в async def остановит обработку всех запросов сервера. Подробнее — в разделе «async def или def» руководства по FastAPI.

Django поддерживает асинхронные представления: достаточно объявить view через async def.

import asyncio

import httpx
from django.http import JsonResponse


async def rates(request):
    async with httpx.AsyncClient(timeout=5) as client:
        usd, eur = await asyncio.gather(
            client.get("https://httpbin.org/get?currency=USD"),
            client.get("https://httpbin.org/get?currency=EUR"),
        )
    return JsonResponse({"usd": usd.json()["args"], "eur": eur.json()["args"]})

Такое представление работает и под WSGI, но полную выгоду дают ASGI-сервер и асинхронные middleware. У ORM есть асинхронные методы с префиксом a: await Book.objects.aget(pk=1), await book.asave(), перебор async for book in Book.objects.filter(...). Синхронный ORM внутри async def вызывает ошибку SynchronousOnlyOperation — такой код оборачивают в sync_to_async() из пакета asgiref, который ставится вместе с Django. Транзакции в асинхронном режиме пока не работают, поэтому их выносят в синхронную функцию. WebSocket и фоновые соединения в Django делают через Django Channels.

Частые ошибки

• Забыли await — RuntimeWarning: coroutine '...' was never awaited, а вместо результата в переменной лежит объект корутины.
• Последовательные await вместо задач — код асинхронный, но работает так же долго, как синхронный. Запускайте операции через TaskGroup или gather.
• time.sleep(), requests и другие блокирующие вызовы в async def — замирает весь цикл событий. Используйте asyncio.sleep(), асинхронные библиотеки или asyncio.to_thread().
• asyncio.run() внутри работающего цикла — в Jupyter и внутри корутин пишите await.
• Задача без ссылки — результат create_task() не сохранили, задача пропала или её ошибка затерялась. Используйте TaskGroup.
• Перехват CancelledError без raise — ломаются отмена, таймауты и TaskGroup.
• Код из старых туториалов — @asyncio.coroutine, yield from, asyncio.get_event_loop() и параметр loop= в функциях удалены или устарели.

Частые вопросы

Как вызвать async-функцию из обычной функции?

В скрипте — через asyncio.run(coro()): он запустит цикл событий, дождётся результата и вернёт его. Делайте это один раз на верхнем уровне программы и не вызывайте внутри уже работающего цикла. В Django для этой задачи есть async_to_sync() из asgiref.

С какой версии Python есть async и await?

Ключевые слова async и await появились в Python 3.5, модуль asyncio — в 3.4. asyncio.run() добавлен в 3.7, asyncio.to_thread() — в 3.9, TaskGroup, asyncio.timeout() и except* — в 3.11, TaskGroup.cancel() — в 3.15. Для нового кода берите Python 3.12 или новее.

Можно ли асинхронно читать и записывать файлы?

В asyncio нет асинхронных функций для обычных файлов. Небольшие файлы читайте как обычно — это быстро. Большие файлы или медленные сетевые диски читайте через asyncio.to_thread() или сторонний пакет aiofiles, который делает то же самое в пуле потоков.

Как работать с базой данных асинхронно?

Нужен асинхронный драйвер: asyncpg или psycopg 3 для PostgreSQL, aiosqlite для SQLite. SQLAlchemy 2.x поддерживает AsyncSession поверх таких драйверов — пример связки в статье о FastAPI и SQLAlchemy. В Django используйте асинхронные методы ORM (aget, acreate, async for).

Как понять, что библиотека асинхронная?

В документации функции вызываются через await, клиенты открываются через async with, а имена часто начинаются с aio или async: aiohttp, aiosqlite, asyncpg. Если примеров с await нет, библиотека синхронная, и async def вокруг её вызова ничего не ускорит.

Чем asyncio отличается от многопоточности?

Потоки переключает операционная система в любой момент, поэтому общие данные приходится защищать блокировками. Корутины переключаются только на await в одном потоке: гонок данных меньше, а тысячи корутин занимают меньше памяти, чем тысячи потоков. Зато каждая библиотека в цепочке вызовов должна поддерживать async.

Итог

async def создаёт корутину, await ждёт результат и отдаёт управление циклу событий, а asyncio.run() запускает всю программу. Чтобы операции шли одновременно, запускайте их задачами: через TaskGroup в Python 3.11+ или gather. Ограничивайте ожидание через asyncio.timeout(), число одновременных запросов — семафором. async ускоряет только ожидание ввода-вывода: блокирующие вызовы выносите в asyncio.to_thread(), вычисления — в пул процессов. Полный список функций модуля — в официальной документации asyncio, а остальной базовый синтаксис языка собран в шпаргалке по синтаксису Python.