Асинхронна обробка даних: як це працює у FastAPI

Згадаймо для аналогії. Уявіть собі ресторан: ви заходите, робите замовлення, і офіціант одразу біжить на кухню, щоб приготувати кожну страву сам. Незручно, правда? Замовлення інших клієнтів доведеться чекати вічно. Асинхронне програмування — це як офіціант, який приймає замовлення, передає його кухарю і займається іншими клієнтами, поки кухня працює.

Асинхронність корисна, коли наш сервер не просто читає і пише пам’ять, а «зависає» в очікуванні (наприклад, поки база даних виконує запит або інший API повертає відповідь). Замість того, щоб сидіти без діла, він може обробляти інші запити. FastAPI — мастер у цій грі завдяки вбудованій підтримці асинхронності.

Як FastAPI працює з асинхронністю?

FastAPI використовує asyncio, вбудовану бібліотеку Python для роботи з асинхронним програмуванням. Давайте розберемо, чому це важливо:

• Асинхронні ендпоінти з допомогою async def: FastAPI розуміє, що ваш ендпоінт може бути "трохи зайнятий" (чекати відповіді від бази даних, наприклад). Використовуючи async і await, FastAPI дозволяє не блокувати виконання інших задач.
• Event Loop (цикл подій): FastAPI "жонглює" вашими задачами через Event Loop. Коли одна задача чекає, інші задачі продовжують виконуватись. Це суперефективно для API з високим навантаженням.
• Підтримка асинхронних бібліотек: FastAPI ідеально поєднується з асинхронними клієнтами, такими як httpx для роботи з API або asyncpg для роботи з PostgreSQL.

Приклад асинхронної обробки даних

Ось невеликий приклад, який показує запит даних до зовнішнього сервісу (наприклад, до API OpenWeather):

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/weather/{city}")
async def get_weather(city: str):
    # Асинхронний запит до зовнішнього API
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=ваш_ключ")
    if response.status_code == 200:
        # Повертаємо відповідь, якщо все пройшло добре
        return response.json()
    else:
        # Генеруємо помилку, якщо щось пішло не так
        return {"error": "Could not fetch weather data"}

1. Використовуємо бібліотеку httpx, яка підтримує асинхронні HTTP-запити.
2. Ключове слово async у функції get_weather каже FastAPI, що це асинхронний ендпоінт.
3. Ключове слово await вказує, що виконання призупиняється до отримання відповіді від зовнішнього API.
4. Завдяки асинхронності, поки ми чекаємо дані від OpenWeather, сервер може обробляти інші запити. Клас!

Переваги асинхронної обробки даних у FastAPI

1. Висока продуктивність: асинхронність дозволяє обробляти тисячі запитів одночасно, не блокуючи сервер.
2. Зменшення часу очікування: користувач бачить відповідь швидше, бо ресурси сервера використовуються оптимально.
3. Повноцінна підтримка вводу‑виводу: асинхронність працює ідеально з асинхронними базами даних і бібліотеками.

Асинхронність з базою даних

Тепер розглянемо випадок, коли дані потрібно запросити з бази даних.

Налаштування асинхронної бази даних (SQLAlchemy + asyncpg)

Asyncpg — асинхронний драйвер для PostgreSQL. Ось приклад налаштування FastAPI для асинхронної взаємодії з базою даних:

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from pydantic import BaseModel

app = FastAPI()

# Створення асинхронного движка SQLAlchemy
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)

# Pydantic модель для валідації даних
class Item(BaseModel):
    name: str
    description: str

async def get_db():
    async with SessionLocal() as session:
        yield session

@app.post("/items/")
async def create_item(item: Item, db: AsyncSession = next(get_db())):
    # Приклад асинхронного додавання запису у базу даних
    query = f"INSERT INTO items (name, description) VALUES ('{item.name}', '{item.description}')"
    await db.execute(query)
    await db.commit()
    return {"message": "Item created"}

Що тут відбувається?

1. AsyncPg: ми використовуємо асинхронний драйвер для роботи з PostgreSQL.
2. Pydantic: валідовуємо вхідні дані за допомогою моделі Item.
3. Асинхронна робота з базою: запити до бази даних виконуються через await, а сервер продовжує працювати з іншими задачами.

Асинхронне виконання задач з asyncio

Іноді потрібно запускати довгі задачі, що не залежать від виконання ендпоінтів. Наприклад, відправка повідомлень.

import asyncio

@app.get("/process/")
async def process_data():
    # Фонова задача
    asyncio.create_task(background_task())
    return {"message": "Processing started!"}

async def background_task():
    print("Починаємо довгу обробку...")
    await asyncio.sleep(10)  # Імітація тривалої задачі
    print("Обробка завершена!")

Такий підхід дозволяє одночасно повертати швидку відповідь клієнту і обробляти довгі операції "за лаштунками".

Поради та типові помилки

1. Змішування синхронного і асинхронного коду: якщо ви використовуєте асинхронні бібліотеки, уникайте вбудованих синхронних клієнтів. Наприклад, замість requests краще використовуйте httpx, а замість стандартних драйверів для бази даних — асинхронні аналоги (asyncpg).
2. Blocking Code: навіть якщо використовуєте async def, включення блокуючого коду (наприклад, читання великих файлів у пам’ять або синхронні операції) "заморожує" весь event loop.
3. Пропущений await: часта помилка, особливо у початківців. Якщо забути await, замість виконання задачі повернеться об’єкт coroutine, який не дасть результату.

Практичне застосування

Асинхронна обробка даних часто використовується в реальних проєктах:

• Мікросервіси: швидка взаємодія між компонентами через API.
• Інтеграція з зовнішніми сервісами: асинхронні запити до сторонніх API.
• Обробка великої кількості запитів: для високонавантажених систем, таких як соціальні платформи або фінансові інструменти, асинхронність — рятунок.

Якщо ви потрапите на співбесіду в компанію, що працює з високонавантаженими системами, знання асинхронного програмування і FastAPI буде величезною перевагою.

FastAPI робить асинхронне програмування дійсно зручним і доступним. Використання асинхронності в ваших проєктах забезпечує підтримку великих навантажень, оптимізацію часу виконання запитів і плавну інтеграцію з іншими сучасними технологіями.