Асинхронные эндпоинты FastAPI: создание GET-запроса

FastAPI прекрасно поддерживает асинхронное программирование "из коробки". Когда мы добавляем к функции ключевое слово async, это позволяет FastAPI автоматически интегрироваться с event loop Python'а и обрабатывать запросы эффективнее, особенно когда дело касается операций ввода-вывода, например работы с базами данных, взаимодействий с внешними API или чтения файлов.

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

---

Пример базового асинхронного эндпоинта

Давайте начнём с самого простого примера асинхронного эндпоинта, реализующего обработку GET-запроса.

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello-async")
async def say_hello():
    return {"message": "Привет, асинхронный мир!"}

Здесь мы используем декоратор @app.get для создания эндпоинта, который будет доступен по адресу /hello-async. Вы заметили, что функция say_hello помечена как async? Это указание FastAPI, что функция асинхронна и может приостанавливать своё выполнение для ожидания завершения операций.

Теперь запустим сервер, чтобы убедиться, что эндпоинт работает. Если вы забыли, как это делается, напомню:

uvicorn main:app --reload

После запуска откройте в браузере: http://127.0.0.1:8000/hello-async. Вы увидите:

{
  "message": "Привет, асинхронный мир!"
}

Ура, мы сделали первый асинхронный эндпоинт!

---

Реализация асинхронности

Асинхронные эндпоинты раскрывают свой потенциал, когда мы добавляем операции, которые действительно выигрывают от асинхронности. Вот пример с использованием функции asyncio.sleep, чтобы симулировать долгую операцию.

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/long-task")
async def long_task():
    await asyncio.sleep(5)  # Симулируем длительную задачу
    return {"message": "Готово! Я подождал 5 секунд."}

Когда вы откроете http://127.0.0.1:8000/long-task, запрос выполнится через 5 секунд, но сервер не будет блокирован. Это значит, что другие пользователи смогут одновременно отправлять запросы и сервер будет их обрабатывать сразу, а не ждать завершения текущей задачи.

А теперь по-настоящему…

Давайте сделаем что-то более интересное: симулируем запрос к внешнему API, используя библиотеку httpx, которая поддерживает асинхронность.

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/external-api")
async def get_external_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return {"external_data": response.json()}

Здесь мы выполняем асинхронный HTTP-запрос к тестовому API https://jsonplaceholder.typicode.com, который возвращает фиктивные данные.

Попробуйте открыть http://127.0.0.1:8000/external-api. Вы увидите ответ от внешнего API:

{
  "external_data": {
    "userId": 1,
    "id": 1,
    "title": "sunt aut facere repellat provident occaecati excepturi optio reprehenderit",
    "body": "quia et suscipit\nsuscipit recusandae consequuntur expedita et cum..."
  }
}

---

Когда асинхронность реально приносит пользу?

Асинхронность сияет, когда мы имеем дело с задачами, которые занимают много времени, такими как:

1. Чтение или запись данных в базу данных.
2. Вызов внешних API.
3. Чтение больших файлов.
4. Выполнение сложных вычислений.

Например, представьте, что у вас есть эндпоинт, который делает несколько запросов одновременно. С помощью асинхронности мы можем значительно ускорить выполнение, так как запросы будут выполняться параллельно.

from fastapi import FastAPI
import httpx

app = FastAPI()

async def fetch_url(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

@app.get("/parallel-requests")
async def parallel_requests():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2"
    ]
    responses = await asyncio.gather(*(fetch_url(url) for url in urls))
    return {"results": responses}

Здесь мы используем asyncio.gather, чтобы сразу отправить два запроса. FastAPI дождётся завершения всех операций и вернёт результат.

---

Типичные ошибки при работе с асинхронными эндпоинтами

Ошибка номер один, которую совершают новички, — это смешивание синхронного и асинхронного кода. Например, если вы используете синхронный запрос в асинхронном эндпоинте:

import requests  # Синхронная библиотека
from fastapi import FastAPI

app = FastAPI()

@app.get("/wrong")
async def wrong_function():
    response = requests.get("https://jsonplaceholder.typicode.com/posts/1")  # ОШИБКА
    return {"data": response.json()}

Этот код не вызовет ошибки сразу, но синхронный вызов requests.get заморозит ваш event loop, и весь сервер перестанет реагировать на другие запросы. Используйте асинхронные библиотеки, например httpx.

Ещё одно распространённое недоразумение: забыть поставить await перед асинхронной функцией. Это приведёт к тому, что функция не выполнится корректно и вернёт объект coroutine вместо результата.

@app.get("/forgot-await")
async def forgot_await():
    result = asyncio.sleep(5)  # ОШИБКА: забыли await
    return {"result": result}

Решение: всегда помните про await перед вызовами асинхронных функций.

---