Какие есть особенности написания асинхронных эндпоинтов в FastAPI?

FastAPI — это современный веб-фреймворк для Python, который изначально поддерживает асинхронное программирование благодаря своей основе на стандарте ASGI (Asynchronous Server Gateway Interface). Асинхронные эндпоинты позволяют серверу эффективно обрабатывать множество одновременных соединений, особенно когда запросы связаны с ожиданием операций ввода-вывода (I/O-bound), таких как сетевые запросы, чтение файлов или запросы к базам данных.

Объявление и выполнение

Асинхронный эндпоинт объявляется с помощью async def. Когда FastAPI получает запрос к такому эндпоинту, он не блокирует весь поток выполнения, ожидая его завершения. Вместо этого, при встрече с операцией await (например, await some_io_operation()), управление временно возвращается циклу событий, который может начать обрабатывать другой запрос. Это значительно повышает пропускную способность приложения при большом количестве одновременных подключений.

Ключевые особенности и лучшие практики

• Использование async/await: Внутри эндпоинта все операции ввода-вывода должны быть асинхронными и вызываться с await.
• Избегание блокирующего кода: Любой синхронный, блокирующий вызов (например, тяжелые вычисления CPU-bound или вызов синхронной библиотеки для БД) внутри async-функции сведёт на нет преимущества асинхронности и может "заморозить" всё приложение. Для такого кода следует использовать фоновые задачи или запускать их в отдельном потоке через asyncio.to_thread или исполнители.
• Асинхронные клиенты и драйверы: Для взаимодействия с внешними сервисами или базами данных необходимо использовать асинхронные библиотеки (например, httpx для HTTP-запросов, asyncpg для PostgreSQL, motor для MongoDB).
• Зависимости (Dependencies): Зависимости также могут быть асинхронными. FastAPI корректно их разрешает.
• Фоновые задачи: Для операций, которые должны выполняться после отправки ответа клиенту, используйте BackgroundTasks.

Пример кода

from fastapi import FastAPI, HTTPException
import httpx
import asyncio

app = FastAPI()

# Асинхронный эндпоинт для получения данных из внешнего API
@app.get("/fetch-data/")
async def fetch_external_data(item_id: int):
    # Используем асинхронный HTTP-клиент
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(f"https://api.example.com/items/{item_id}", timeout=10.0)
            response.raise_for_status()
            return response.json()
        except httpx.RequestError:
            raise HTTPException(status_code=503, detail="External service unavailable")

# Эндпоинт, демонстрирующий проблему с блокирующим кодом и её решение
@app.get("/process/")
async def process_data():
    # ПЛОХО: Синхронная, блокирующая операция внутри async-функции.
    # time.sleep(5)  # Это заблокирует весь event loop!

    # ХОРОШО: "Усыпляем" асинхронно, не блокируя цикл событий.
    await asyncio.sleep(1)

    # Если необходимо выполнить синхронный CPU-bound код, запускаем его в отдельном потоке.
    result = await asyncio.to_thread(cpu_intensive_calculation, 1000)
    return {"result": result}

def cpu_intensive_calculation(n: int):
    # Тяжёлые синхронные вычисления
    return sum(i * i for i in range(n))

Вывод: Асинхронные эндпоинты в FastAPI следует применять для построения высокомасштабируемых API, которые часто выполняют операции ввода-вывода. Они позволяют эффективно использовать ресурсы сервера, обслуживая тысячи одновременных соединений. Однако важно следить за тем, чтобы внутри таких эндпоинтов не было синхронных блокирующих вызовов, и использовать соответствующие асинхронные библиотеки для всех операций с ожиданием.