Створення асинхронних запитів до MongoDB через FastAPI

Сьогодні ми повністю зосередимося на створенні асинхронних запитів, обговоримо їхнє значення для високонавантажених систем, дізнаємося, як працює event loop, і напишемо кілька практичних прикладів. Поїхали!

Уяви, що ти замовив доставку з магазину через додаток. У синхронному сценарії — ти сидиш і дивишся в екран телефона, поки кур’єр везе замовлення. Нікуди не йдеш, ні на що не відволікаєшся — просто чекаєш.

В асинхронному варіанті — ти зробив замовлення і пішов далі займатися своїми справами: працюєш, дивишся фільм, готуєш вечерю. Коли кур’єр під’їхав — тобі приходить повідомлення, і ти виходиш за покупками. Тобто ти не втрачаєш час, поки щось десь обробляється. Твій "головний потік" залишається вільним і може робити інші важливі речі.

Ось це й є асинхронність: не блокувати процес очікування, а звільняти систему для інших задач, поки результат десь формується.

У термінах серверів: асинхронність = менше простаювання, більше клієнтів одночасно.

Асинхронність робить наші додатки більш відзивчивими і здатними обслуговувати більшу кількість клієнтів одночасно. Це досягається за рахунок використання вбудованого в Python механізму event loop та ключових слів async/await. Детальніше про event loop можна почитати в документації Python.

Асинхронне програмування в Python

Асинхронне програмування базується на event loop — циклі, який планує виконання задач. У Python асинхронність досягається за допомогою ключових слів async і await. Давайте заглибимося в базові поняття.

import asyncio

async def say_hello():
    print("Привіт, світ!")
    await asyncio.sleep(1)
    print("Минуло 1 секунду")

async def main():
    await say_hello()

# Запуск event loop
asyncio.run(main())

Тут async робить функцію асинхронною, а await використовується для позначення точок, де виконання може бути призупинене.

Важливий момент: ти не можеш викликати асинхронну функцію напряму, її завжди потрібно запускати через event loop — напряму або опосередковано.

---

Асинхронні маршрути в FastAPI для MongoDB

У FastAPI підтримка асинхронності вбудована "з коробки". Асинхронні маршрути дозволяють нам писати API, які ефективно працюють з базами даних, наприклад, MongoDB, використовуючи асинхронні бібліотеки на кшталт motor.

Крок 1: налаштування бібліотеки Motor

Якщо ти пропустив налаштування motor у попередній лекції, не біда. Встановимо її:

pip install motor

Підключаємо motor і створюємо з’єднання:

from motor.motor_asyncio import AsyncIOMotorClient

MONGO_DETAILS = "mongodb://localhost:27017"
client = AsyncIOMotorClient(MONGO_DETAILS)
database = client.my_database
collection = database.get_collection("my_collection")

Тепер у нас є колекція my_collection, в яку ми можемо вставляти і витягувати дані. Ніхто не втримається перед її потужністю!

Крок 2: асинхронне додавання даних

Створюємо маршрут для додавання записів у MongoDB:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str

@app.post("/items/")
async def create_item(item: Item):
    new_item = await collection.insert_one(item.dict())
    created_item = await collection.find_one({"_id": new_item.inserted_id})
    return created_item

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

1. Ми оголосили модель Item через Pydantic — це зручно і дозволяє валідовувати вхідні дані.
2. Використовуємо асинхронний метод insert_one для додавання нового запису в колекцію.
3. Метод find_one повертає доданий документ за його ID.

Крок 3: асинхронне читання даних

Додамо маршрут для читання всіх записів:

@app.get("/items/")
async def get_items():
    items = []
    async for item in collection.find():
        items.append(item)
    return items

Особливість MongoDB у тому, що find() повертає асинхронний курсор. Тому ми використовуємо async for для ітерації по результатах.

Крок 4: оновлення записів

Оновлювати дані теж нескладно:

@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
    update_result = await collection.update_one(
        {"_id": item_id}, {"$set": item.dict()}
    )
    if update_result.modified_count == 1:
        updated_item = await collection.find_one({"_id": item_id})
        if updated_item:
            return updated_item
    return {"error": "Не вдалося оновити елемент"}

Ми шукаємо документ по полю _id, оновлюємо його і повертаємо нове значення, якщо оновлення успішне.

Крок 5: видалення записів

Видалення простіше простого:

@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    delete_result = await collection.delete_one({"_id": item_id})
    if delete_result.deleted_count == 1:
        return {"message": "Елемент успішно видалено"}
    return {"error": "Елемент не знайдено"}

Тут ми використовуємо delete_one для видалення запису. Якщо елементів з таким ID не знайдено, повертається відповідна помилка.

---

3. Практичний приклад: Створюємо просте API

Ми побудуємо API для керування списком книг. Воно підтримуватиме операції: додавання, читання, оновлення і видалення книг.

Додамо модель для книги:

class Book(BaseModel):
    title: str
    author: str
    summary: str

Об’єднаємо наш попередній код в одному застосунку з маршрутом для кожної операції.

@app.post("/books/")
async def create_book(book: Book):
    new_book = await collection.insert_one(book.dict())
    return await collection.find_one({"_id": new_book.inserted_id})


@app.get("/books/")
async def list_books():
    books = []
    async for book in collection.find():
        books.append(book)
    return books


@app.put("/books/{book_id}")
async def update_book(book_id: str, book: Book):
    result = await collection.update_one({"_id": book_id}, {"$set": book.dict()})
    if result.modified_count == 1:
        return await collection.find_one({"_id": book_id})
    return {"error": "Книга не знайдена або не оновлена"}


@app.delete("/books/{book_id}")
async def delete_book(book_id: str):
    result = await collection.delete_one({"_id": book_id})
    if result.deleted_count == 1:
        return {"message": "Книгу успішно видалено"}
    return {"error": "Книга не знайдена"}

---

Типові помилки

1. Не використовувати асинхронні методи Motor: використання звичайних блокуючих методів може призвести до "зависання" застосунку. Будь уважний і використовуй тільки асинхронні методи (async def і await).
2. Неправильна структура MongoDB: MongoDB дозволяє зберігати довільні дані, але якщо ти забудеш узгодити структуру записів, відладка може стати кошмаром.
3. Відсутність індексів: без індексації твої запити можуть стати повільними, особливо при зростанні кількості записів.

---

Реальне застосування

Асинхронні запити широко застосовуються в високонавантажених додатках: від API для соціальних мереж до систем аналітики. MongoDB прекрасно підходить для зберігання даних у форматі JSON, а FastAPI дозволяє використовувати її можливості максимально ефективно.