Приведи пример написания асинхронного кода

Асинхронный код в Python: async/await

Асинхронное программирование позволяет выполнять несколько задач "одновременно" без многопоточности. Это идеально для I/O-bound операций (HTTP запросы, БД).

Основная концепция

import asyncio

# Обычная функция (синхронная)
def fetch_user():
    print("Начинаю загрузку...")
    time.sleep(2)  # Ждём 2 секунды (БЛОКИРУЕТ!)
    print("Готово!")
    return {"id": 1, "name": "Alice"}

# Асинхронная функция (может ждать без блокировки)
async def fetch_user_async():
    print("Начинаю загрузку...")
    await asyncio.sleep(2)  # Ждём 2 секунды, но ЭВМ может делать другое
    print("Готово!")
    return {"id": 1, "name": "Alice"}

# Запустить асинхронную функцию
asyncio.run(fetch_user_async())

Проблема: синхронный код (медленный)

import time
import requests

def fetch_data():
    # Загрузить 3 пользователя
    user1 = requests.get('https://api.example.com/users/1').json()  # 1 сек
    user2 = requests.get('https://api.example.com/users/2').json()  # 1 сек
    user3 = requests.get('https://api.example.com/users/3').json()  # 1 сек
    # Итого: 3 секунды (выполняются последовательно!)
    return [user1, user2, user3]

start = time.time()
fetch_data()
print(f"Время: {time.time() - start}s")  # 3 сек

Решение: асинхронный код (быстрый)

import asyncio
import aiohttp

async def fetch_user_async(session, user_id):
    async with session.get(f'https://api.example.com/users/{user_id}') as resp:
        return await resp.json()

async def fetch_all_users():
    async with aiohttp.ClientSession() as session:
        # Запустить все 3 запроса одновременно
        tasks = [
            fetch_user_async(session, 1),
            fetch_user_async(session, 2),
            fetch_user_async(session, 3),
        ]
        users = await asyncio.gather(*tasks)  # Ждём все сразу
        return users

import time
start = time.time()
asyncio.run(fetch_all_users())
print(f"Время: {time.time() - start}s")  # ~1 сек вместо 3!

Ключевые ключевые слова

async def — определить асинхронную функцию

async def my_async_function():
    pass

await — ждать результат asynchronous функции

result = await some_async_func()  # Ждём результат
print(result)

asyncio.gather() — выполнить несколько задач одновременно

results = await asyncio.gather(
    fetch_user(1),
    fetch_user(2),
    fetch_user(3)
)  # Выполняются в параллель, ждём все

Практический пример: FastAPI с async

from fastapi import FastAPI
import aiohttp
import asyncio

app = FastAPI()

async def fetch_weather(city: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(f'https://api.weather.com/{city}') as resp:
            return await resp.json()

@app.get("/weather")
async def get_weather(city: str):
    """Endpoint возвращает погоду для города."""
    weather = await fetch_weather(city)
    return weather

# Когда 100 пользователей одновременно вызывают /weather?city=moscow
# Каждый запрос будет ждать асинхронно
# ЭВМ обработает 100 запросов без блокировок
# В синхронной версии это требовало 100 потоков!

Пример: параллельные запросы к БД

import asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/db"
)

async def get_user(session: AsyncSession, user_id: int):
    result = await session.execute(
        "SELECT * FROM users WHERE id = ?", (user_id,)
    )
    return result.fetchone()

async def fetch_multiple_users():
    from sqlalchemy.ext.asyncio import AsyncSession
    
    async with AsyncSession(engine) as session:
        # Запустить 3 запроса к БД одновременно
        users = await asyncio.gather(
            get_user(session, 1),
            get_user(session, 2),
            get_user(session, 3),
        )
        return users

Пример: timeout и обработка ошибок

async def fetch_with_timeout(url, timeout=5):
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=timeout) as resp:
                return await resp.json()
    except asyncio.TimeoutError:
        print("Запрос истёк по времени")
        return None
    except Exception as e:
        print(f"Ошибка: {e}")
        return None

async def main():
    # Запустить с timeout на всю операцию
    result = await asyncio.wait_for(
        fetch_with_timeout('https://api.example.com/data'),
        timeout=10  # 10 секунд на всё
    )
    return result

asyncio.run(main())

Пример: очередь задач (Queue)

import asyncio

async def worker(queue: asyncio.Queue):
    while True:
        task = await queue.get()  # Ждём задачу из очереди
        try:
            print(f"Обрабатываю задачу: {task}")
            await asyncio.sleep(1)  # Имитируем работу
            print(f"Готово: {task}")
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue()
    
    # Запустить 3 рабочих
    workers = [
        asyncio.create_task(worker(queue))
        for _ in range(3)
    ]
    
    # Добавить задачи в очередь
    for i in range(10):
        queue.put_nowait(f"task-{i}")
    
    # Ждём, пока все задачи будут обработаны
    await queue.join()
    
    # Отменить рабочих
    for w in workers:
        w.cancel()

asyncio.run(main())

Пример: Telegram бот (aiogram)

from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
import asyncio

bot = Bot(token="YOUR_TOKEN")
dp = Dispatcher()

@dp.message(Command("start"))
async def start_handler(message: types.Message):
    # Асинхронный обработчик
    await bot.send_message(
        chat_id=message.chat.id,
        text="Привет! 👋"
    )

@dp.message()
async def echo_handler(message: types.Message):
    # Fetch данные асинхронно
    user_data = await get_user_data(message.from_user.id)
    
    await message.reply(
        f"Ты написал: {message.text}\n"
        f"Твои данные: {user_data}"
    )

async def main():
    await dp.start_polling(bot)

asyncio.run(main())

Как выбрать между async и sync

Используйте ASYNC если:

• HTTP запросы (requests, aiohttp)
• Запросы к БД
• WebSocket
• Работа с файлами
• Telegram/Discord боты
• Много I/O операций

# Async: 100 HTTP запросов за 5 сек
await asyncio.gather(*[fetch(url) for url in urls])

# Sync: 100 HTTP запросов за 500 сек!
for url in urls:
    fetch(url)  # Каждый по 5 сек

Используйте SYNC если:

• Чистые CPU вычисления
• Обработка данных в памяти
• Machine learning
• Мало I/O операций

Ошибка: забыли await

# НЕПРАВИЛЬНО!
async def main():
    result = fetch_data()  # Забыли await!
    print(result)  # Это будет coroutine объект, не результат

# ПРАВИЛЬНО!
async def main():
    result = await fetch_data()  # await!
    print(result)  # Это результат

Вывод

Async/await в Python позволяет:

• Обрабатывать тысячи одновременных I/O операций
• Не использовать threads и их сложность
• Писать читаемый асинхронный код
• Масштабировать приложение без увеличения сложности

Правило: Если вы пишете код, который ждёт (HTTP, БД), используйте async/await!