Асинхронні ендпоінти 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 перед викликами асинхронних функцій.

---