Что такое асинхронная функция в Python?

Что такое асинхронная функция в Python?

Асинхронная функция (async function) в Python — это функция, которая может быть приостановлена и возобновлена, позволяя другим кодам выполняться пока она ожидает (например, ответ от API или доступ к БД). Асинхронное программирование позволяет обрабатывать множество операций одновременно в одном потоке, значительно повышая производительность приложений с I/O операциями.

Основные концепции

1. Синхронный vs Асинхронный код

import time
import asyncio

# СИНХРОННЫЙ КОД (блокирующий)
def fetch_user(user_id):
    """Этот код заблокирует, пока ждет ответа"""
    time.sleep(2)  # Имитация сетевого запроса
    print(f"Получили пользователя {user_id}")
    return {"id": user_id, "name": "Иван"}

print("Начало")
for i in range(3):
    fetch_user(i)
print("Конец")
# Выполняется последовательно: 6 секунд (2+2+2)

# АСИНХРОННЫЙ КОД (неблокирующий)
async def fetch_user_async(user_id):
    """Эта функция может быть приостановлена"""
    await asyncio.sleep(2)  # Имитация сетевого запроса
    print(f"Получили пользователя {user_id}")
    return {"id": user_id, "name": "Иван"}

async def main():
    print("Начало")
    tasks = [fetch_user_async(i) for i in range(3)]
    results = await asyncio.gather(*tasks)  # Все выполняются параллельно
    print("Конец")

asyncio.run(main())
# Выполняется параллельно: ~2 секунды (все трое одновременно)

Синтаксис асинхронных функций

1. Определение асинхронной функции

import asyncio

# Обычная функция
def sync_function():
    return "Результат"

# Асинхронная функция
async def async_function():
    return "Результат"

# Асинхронная функция с ожиданием
async def async_with_await():
    result = await async_function()
    return result

2. Использование await

await используется только внутри асинхронной функции для ожидания выполнения другой асинхронной функции:

async def fetch_user(user_id):
    # await приостанавливает выполнение функции
    user_data = await get_user_from_api(user_id)
    posts = await get_user_posts(user_id)
    return {"user": user_data, "posts": posts}

async def get_user_from_api(user_id):
    await asyncio.sleep(1)
    return {"id": user_id, "name": "Иван"}

async def get_user_posts(user_id):
    await asyncio.sleep(1)
    return ["Пост 1", "Пост 2"]

3. Запуск асинхронного кода

# Способ 1: asyncio.run() (Python 3.7+, рекомендуется)
async def main():
    result = await async_function()
    return result

result = asyncio.run(main())

# Способ 2: используя event loop напрямую (старый способ)
loop = asyncio.get_event_loop()
result = loop.run_until_complete(main())

Практические примеры

1. Параллельные HTTP запросы

import asyncio
import aiohttp

async def fetch_user_data(session, user_id):
    """Получить данные одного пользователя"""
    url = f"https://api.example.com/users/{user_id}"
    async with session.get(url) as response:
        return await response.json()

async def fetch_all_users(user_ids):
    """Получить данные всех пользователей параллельно"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_user_data(session, uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
    return results

# Использование
user_ids = [1, 2, 3, 4, 5]
results = asyncio.run(fetch_all_users(user_ids))
# Вместо 5 последовательных запросов (5 секунд),
# они выполняются параллельно (~1 секунда)

2. Асинхронная работа с базой данных

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

async def get_user_from_db(session: AsyncSession, user_id: int):
    """Получить пользователя из БД асинхронно"""
    result = await session.execute(
        select(User).where(User.id == user_id)
    )
    return result.scalar_one_or_none()

async def get_multiple_users(session: AsyncSession, user_ids: list):
    """Получить нескольких пользователей параллельно"""
    tasks = [get_user_from_db(session, uid) for uid in user_ids]
    users = await asyncio.gather(*tasks)
    return users

3. Таймауты и обработка ошибок

async def fetch_with_timeout(url, timeout=5):
    """Получить данные с таймаутом"""
    try:
        async with asyncio.timeout(timeout):
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.json()
    except asyncio.TimeoutError:
        print(f"Таймаут при запросе к {url}")
        return None
    except aiohttp.ClientError as e:
        print(f"Ошибка сети: {e}")
        return None

# Использование
data = asyncio.run(fetch_with_timeout("https://api.example.com/data"))

4. asyncio.gather и asyncio.create_task

async def main():
    # Способ 1: asyncio.gather (рекомендуется для множественных задач)
    results = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
        return_exceptions=True  # Не прерывать при ошибках
    )
    
    # Способ 2: asyncio.create_task (для долгоживущих задач)
    task1 = asyncio.create_task(fetch_user(1))
    task2 = asyncio.create_task(fetch_user(2))
    
    result1 = await task1
    result2 = await task2
    
    # Способ 3: задачи можно отменить
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(5)
    task.cancel()  # Отменить задачу

5. Асинхронный контекстный менеджер

class AsyncDatabaseConnection:
    async def __aenter__(self):
        """Подключение к БД"""
        self.connection = await connect_to_db()
        return self.connection
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Закрытие соединения"""
        await self.connection.close()

# Использование
async def main():
    async with AsyncDatabaseConnection() as db:
        users = await db.query("SELECT * FROM users")
        return users

FastAPI с асинхронностью

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """Асинхронный endpoint"""
    # Может обрабатывать многих пользователей одновременно
    user = await fetch_user_from_db(user_id)
    posts = await fetch_user_posts(user_id)
    return {"user": user, "posts": posts}

@app.get("/users")
async def get_multiple_users(ids: list[int]):
    """Параллельная загрузка нескольких пользователей"""
    users = await asyncio.gather(*[fetch_user_from_db(uid) for uid in ids])
    return users

# Этот сервер может обслуживать тысячи одновременных запросов
# потому что он не блокируется на I/O операциях

Когда использовать асинхронность

# ✅ ИСПОЛЬЗУЙ асинхронность для:
# - HTTP запросов (requests, aiohttp)
# - Работы с БД (asyncpg, motor)
# - Файловой системы (aiofiles)
# - Web приложений (FastAPI, aiohttp)
# - Реального времени (WebSocket)

# ❌ НЕ используй асинхронность для:
# - Чистых вычислений (используй multiprocessing)
# - I/O блокирующихся операций (используй threading или multiprocessing)
# - Когда нет I/O операций вообще

Мониторинг асинхронного кода

import asyncio
import time

async def timed_task(name, delay):
    start = time.time()
    await asyncio.sleep(delay)
    elapsed = time.time() - start
    print(f"{name} завершена за {elapsed:.2f}с")
    return name

async def main():
    start_total = time.time()
    
    results = await asyncio.gather(
        timed_task("Задача 1", 2),
        timed_task("Задача 2", 3),
        timed_task("Задача 3", 1),
    )
    
    total_time = time.time() - start_total
    print(f"Общее время: {total_time:.2f}с (параллельное выполнение)")

asyncio.run(main())
# Результат: ~3 секунды вместо 6 (если бы выполнялось последовательно)

Заключение

Асинхронное программирование в Python критически важно для:

• Высокопроизводительных веб-приложений
• Приложений, работающих с внешними API
• Приложений, обслуживающих многих пользователей одновременно

Основное правило: используй async/await для I/O операций, которые могут быть приостановлены. Это позволяет твоему коду обрабатывать множество одновременных операций эффективно.