Інтеграція MongoDB з FastAPI через бібліотеки (наприклад, motor)

Уяви собі фастфуд-ресторан. Якби кожного відвідувача обслуговували лише після того, як попередній завершить своє замовлення, черга рухалася б повільно, і клієнти почали б шукати інший заклад. Це — синхронна обробка запитів. Асинхронність же дає змогу обробляти запити паралельно: поки один клієнт читає меню, інший уже отримує своє замовлення. Асинхронні бібліотеки, як-от motor, дозволяють вашому FastAPI-додатку працювати за таким "ресторанним" принципом.

---

Бібліотека motor: асинхронна робота з MongoDB

motor — це асинхронний драйвер для роботи з MongoDB, побудований на основі популярної бібліотеки pymongo. Він дозволяє легко використовувати переваги асинхронного вводу/виводу в Python з такими фреймворками, як FastAPI.

Чому саме motor? Бо FastAPI підтримує асинхронні запити "з коробки". Тобто використання motor оптимально підходить для побудови швидких і масштабованих додатків.

Якщо в тебе ще не встановлений motor, додай його в проєкт за допомогою цієї команди:

pip install motor

---

Налаштування підключення MongoDB за допомогою motor

Створимо простий FastAPI-додаток з підключенням до MongoDB через motor.

Для зручності триматимемося такої структури:

.
├── app/
│   ├── main.py
│   ├── database.py
│   ├── models/
│   └── schemas/
└── requirements.txt

Підключення MongoDB в database.py

Почнемо зі створення файлу database.py, де буде логіка для роботи з базою даних, включно з налаштуванням підключення.

from motor.motor_asyncio import AsyncIOMotorClient

# URL для підключення до MongoDB
MONGO_DETAILS = "mongodb://localhost:27017"

# Створюємо клієнта для асинхронних запитів
client = AsyncIOMotorClient(MONGO_DETAILS)

# Підключаємося до бази даних
database = client.my_database  # Замініть "my_database" на ім'я вашої бази

# Підключаємося до колекції
user_collection = database.get_collection("users")

В цьому коді:

• AsyncIOMotorClient — асинхронний клієнт для підключення до MongoDB.
• MONGO_DETAILS — рядок підключення до MongoDB. (Не забудь замінити localhost і 27017, якщо в тебе інша конфігурація.)
• database.get_collection() — отримуємо колекцію users з бази даних.

Щоб переконатися в успішному підключенні, можна додати в main.py такий код:

from fastapi import FastAPI
from app.database import user_collection

app = FastAPI()

@app.get("/")
async def check_connection():
    # Перевіряємо, чи підключені ми до MongoDB, читаючи кількість документів
    count = await user_collection.count_documents({})
    return {"message": "Connection successful", "user_count": count}

Тепер, якщо сервер FastAPI запущений, перехід на http://127.0.0.1:8000/ має повернути щось на кшталт:

{
  "message": "Connection successful",
  "user_count": 0
}

---

CRUD-операції з використанням motor

Настав час використати потужність motor для виконання CRUD-операцій.

Створення користувача (Create)

Додамо можливість створювати користувача. Почнемо з простого маршруту:

from fastapi import FastAPI
from pydantic import BaseModel
from app.database import user_collection

app = FastAPI()

# Описуємо схему даних користувача
class User(BaseModel):
    name: str
    email: str

@app.post("/users/")
async def create_user(user: User):
    user_dict = user.dict()  # Перетворюємо об'єкт User у словник
    result = await user_collection.insert_one(user_dict)  # Вставляємо документ у MongoDB
    return {"id": str(result.inserted_id), "name": user.name}

Тепер можна відправити POST-запит на /users/ з тілом:

{
  "name": "Alice",
  "email": "alice@example.com"
}

І MongoDB додасть користувача з унікальним ідентифікатором _id.

Читання користувачів (Read)

Додамо можливість читати всі створені записи. Для цього використовується метод find:

@app.get("/users/")
async def get_users():
    users = []
    async for user in user_collection.find():  # Асинхронний перебір курсора
        user["_id"] = str(user["_id"])  # Перетворюємо ObjectId у рядок
        users.append(user)
    return users

Оновлення користувача (Update)

Додамо маршрут для оновлення даних користувача:

from bson import ObjectId

@app.put("/users/{user_id}")
async def update_user(user_id: str, user: User):
    if not ObjectId.is_valid(user_id):
        return {"error": "Invalid ID format"}
    
    result = await user_collection.update_one(
        {"_id": ObjectId(user_id)},
        {"$set": user.dict()}
    )
    if result.modified_count == 1:
        return {"message": "User updated successfully"}
    return {"message": "No changes made"}

Тут ми перевіряємо, щоб user_id був валідним ObjectId, а потім оновлюємо документ.

Видалення користувача (Delete)

Завершимо додаванням функціоналу для видалення користувача:

@app.delete("/users/{user_id}")
async def delete_user(user_id: str):
    if not ObjectId.is_valid(user_id):
        return {"error": "Invalid ID format"}
    
    result = await user_collection.delete_one({"_id": ObjectId(user_id)})
    if result.deleted_count == 1:
        return {"message": "User deleted successfully"}
    return {"message": "User not found"}

---

Використання connection pool для підвищення продуктивності

У великих проєктах важливо враховувати кількість одночасно відкритих з'єднань. motor дозволяє налаштовувати connection pool для роботи з великою кількістю запитів:

client = AsyncIOMotorClient(
    MONGO_DETAILS,
    maxPoolSize=10,  # Максимальна кількість з'єднань
    minPoolSize=1    # Мінімальна кількість з'єднань
)

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

---

Підсумки

Інтеграція FastAPI і MongoDB через бібліотеку motor дозволяє створювати потужні асинхронні додатки. Окрім стандартних CRUD-операцій, ти дізнався, як працювати з колекціями, налаштовувати connection pool і оптимізувати з'єднання. У наступній лекції ми заглибимося в створення асинхронних запитів і розробимо повноцінне API з MongoDB.