asyncio в Python: асинхронное программирование

asyncio — стандартная библиотека Python для асинхронного кода. Позволяет выполнять несколько задач «одновременно» в одном потоке.

Аналогия

Синхронный повар: поставил стейк → стоит и ждёт → готово → режет овощи.

Асинхронный повар: поставил стейк → пока жарится — режет овощи → ставит воду → возвращается к стейку.

Один поток, несколько дел одновременно.

async / await

import asyncio

async def fetch_data(url: str) -> str:
    await asyncio.sleep(1)  # имитация HTTP запроса
    return f"Данные с {url}"

async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

asyncio.run(main())

• async def — объявляет корутину
• await — приостанавливает выполнение, отдаёт управление event loop
• asyncio.run() — запускает event loop

asyncio.gather() — параллельные задачи

import asyncio

async def task(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} готово"

async def main():
    # Запускаем три задачи параллельно
    results = await asyncio.gather(
        task("A", 1.0),
        task("B", 0.5),
        task("C", 1.5),
    )
    print(results)  # ['A готово', 'B готово', 'C готово']
    # Выполнится за 1.5 сек, не за 3.0!

asyncio.run(main())

AsyncAnthropic — пример с Claude

import asyncio
import anthropic

client = anthropic.AsyncAnthropic()

async def ask_claude(question: str) -> str:
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": question}]
    )
    return response.content[0].text

async def analyze_batch(texts: list[str]) -> list[str]:
    tasks = [ask_claude(f"Резюмируй: {text}") for text in texts]
    return await asyncio.gather(*tasks)

asyncio.run(analyze_batch(["текст 1", "текст 2", "текст 3"]))

asyncio vs threading