Интеграция 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.