Создание REST API с FastAPI и MongoDB

Создание REST API часто выглядит как набор рутинных шагов: настройка маршрутов, валидация, подключение к базе. FastAPI значительно упрощает этот процесс. В паре с MongoDB вы быстро получаете рабочий CRUD‑API, пригодный для прототипа или малого проекта.

В этой статье вы найдёте:

• пошаговую установку окружения;
• пример структуры проекта;
• рабочие маршруты для CRUD на Python с объяснениями;
• рекомендации по тестированию, безопасности и альтернативы (например, Motor для асинхронного доступа к MongoDB).

Важно: примеры ориентированы на Python 3.6+ и локальную MongoDB (порт по умолчанию 27017). Для production рекомендую скрывать строки подключения в переменных окружения.

Что такое FastAPI

FastAPI — лёгкий и быстрый Python‑фреймворк для создания API. Он автоматически валидирует входные данные через Pydantic, поддерживает асинхронные обработчики и генерирует документацию Swagger/OpenAPI.

Ключевые свойства (одно предложение каждое):

• Асинхронность: позволяет обрабатывать несколько запросов параллельно при использовании совместимых драйверов.
• Валидация: Pydantic автоматически проверяет типы и структуру JSON‑запросов.
• Документация: Swagger UI и ReDoc доступны из коробки.

Преимущества FastAPI по сравнению с другими фреймворками

• Быстрее в типичных задачах API благодаря использованию asyncio и uvicorn/ASGI.
• Меньше шаблонного кода: декларативные модели Pydantic и автогенерация схем упрощают разработку.
• Упрощённая документация и удобные инструменты разработки.

Совет: если вам нужен полностью синхронный стек — можно использовать Flask/Django, но при росте нагрузки FastAPI с асинхронным драйвером будет масштабироваться лучше.

Установка и настройка MongoDB

Для начала установите MongoDB локально или используйте облачный кластер (Atlas). Затем можно открыть MongoDB Compass для визуального управления базами данных.

Шаги в Compass:

1. Нажмите New Connection.
2. Вставьте URI (например mongodb://localhost:27017 или строку из Atlas).
3. Подключитесь и создайте базу данных и коллекцию для тестовых данных.

Примечание: используйте отдельную базу данных для разработки и для тестирования. Не храните секреты подключения в коде.

Подготовка окружения и установка зависимостей

1. Создайте папку проекта и откройте терминал в ней.
2. Убедитесь, что установлен Python 3.6 или выше:

python --version

3. Установите virtualenv (рекомендуется):

pip install virtualenv

4. Создайте виртуальное окружение и активируйте его:

На Unix / macOS:

virtualenv venv
source venv/bin/activate

На Windows (PowerShell):

virtualenv venv
.\venv\Scripts\Activate.ps1

5. Установите зависимости:

pip install fastapi pymongo uvicorn

Пояснения:

• pymongo — официальный блокирующий драйвер MongoDB для Python. Если вы планируете использовать асинхронные обработчики FastAPI, рассмотрите драйвер Motor (асинхронный).
• uvicorn — ASGI‑сервер для запуска FastAPI. При разработке используйте флаг –reload.

Минимальный сервер FastAPI

Создайте файл server.py в корне проекта и добавьте минимальный пример:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def home():
    return {"message": "Hello World"}

Запуск dev‑сервера:

uvicorn server:app --reload --port 8000

Откройте http://localhost:8000 для проверки и http://localhost:8000/docs для Swagger UI.

Важно: если вы используете PyMongo (блокирующий), ставьте блоки ввода/вывода в потоковый пул или используйте Motor для асинхронного доступа.

Структура проекта и файлы

Рекомендуемая простая структура:

├── config
├── models
├── routes
├── schemas
└── server.py

Эта структура отделяет конфигурацию, модели данных, маршруты и сериализаторы.

1. Конфигурация подключения к БД

Создайте config/db.py:

from pymongo import MongoClient

# Пример локального подключения. Для production используйте переменные окружения
db_connection = MongoClient("mongodb://localhost:27017")
db = db_connection.database_name
collection = db["collection_name"]

Пояснение: MongoClient принимает URI с хостом и портом. db и collection указывают на базу и коллекцию.

2. Модель данных (Pydantic)

Файл models/user_model.py:

from pydantic import BaseModel

class User(BaseModel):
    name: str
    role: str

Pydantic автоматически проверит, что в запросе придут строки для name и role.

3. Функции сериализации

Сериалиция нужна, чтобы преобразовать MongoDB документ (_id типа ObjectId) в JSON‑совместимый вид. Создайте schemas/user_schema.py:

from bson import ObjectId

def user_serializer(user) -> dict:
    return {
        'id': str(user["_id"]),
        'name': user["name"],
        'role': user["role"]
    }

def users_serializer(users) -> list:
    return [user_serializer(user) for user in users]

Учтите: ObjectId нужно преобразовывать в строку, иначе JSON сериализация упадёт.

4. Маршруты CRUD

Создайте routes/user_routes.py. Ниже — пример с импортами и четырьмя методами.

from fastapi import APIRouter, HTTPException
from models.user_model import User
from schemas.user_schema import users_serializer
from bson import ObjectId
from config.db import collection

user = APIRouter()

@user.post("/")
async def create_user(user: User):
    inserted = collection.insert_one(dict(user))
    created = users_serializer(collection.find({"_id": inserted.inserted_id}))
    return {"status": "Ok", "data": created}

@user.get("/")
async def find_all_users():
    users = users_serializer(collection.find())
    return {"status": "Ok", "data": users}

@user.get("/{id}")
async def get_one_user(id: str):
    try:
        user = users_serializer(collection.find({"_id": ObjectId(id)}))
        return {"status": "Ok", "data": user}
    except Exception:
        raise HTTPException(status_code=404, detail="User not found")

@user.put("/{id}")
async def update_user(id: str, user: User):
    collection.find_one_and_update(
        {"_id": ObjectId(id)},
        {"$set": dict(user)}
    )
    updated = users_serializer(collection.find({"_id": ObjectId(id)}))
    return {"status": "Ok", "data": updated}

@user.delete("/{id}")
async def delete_user(id: str):
    collection.find_one_and_delete({"_id": ObjectId(id)})
    return {"status": "Ok", "data": []}

Пояснения по коду:

• insert_one возвращает объект с полем inserted_id, которое используется для последующего поиска.
• find возвращает курсор; users_serializer преобразует все документы.
• Для простоты в примере используется HTTPException с кодом 404 при ошибках.

Инициализация маршрутов в server.py

Если у вас есть отдельный server.py с FastAPI‑объектом, подключите маршруты так:

from fastapi import FastAPI
from routes.user_routes import user

app = FastAPI()
app.include_router(user, prefix="/users", tags=["users"])

@app.get("/")
async def home():
    return {"message": "API is running"}

После запуска Uvicorn ваши маршруты будут доступны по /users.

Важные замечания и альтернативы

• PyMongo — блокирующий драйвер. Если вы используете async‑эндпоинты FastAPI и ждёте высокой нагрузки, рассмотрите Motor (асинхронный драйвер MongoDB). Иначе ставьте блокирующие операции в ThreadPool.
• Валидация: Pydantic валидирует вход, но проверку уникальности и дополнительные ограничения стоит выполнять на уровне БД или в бизнес-логике.
• Обработка ошибок: добавьте обработку ошибок подключения к БД, таймауты и логирование.

Когда предложенный подход не подходит:

• Нужна сложная транзакционная логика — реляционные СУБД (PostgreSQL) предпочтительнее.
• Если нужны сложные JOIN‑операции — MongoDB усложнит логику запросов.

Альтернативные подходы:

• Использовать Motor (асинхронный) вместо PyMongo.
• Использовать ODM (например, Beanie) поверх Motor для удобной работы с моделями.

Критерии приёмки

1. Все CRUD‑маршруты отвечают 200/201/404 в соответствии с ситуацией.
2. Валидация входных данных проходит (поля name и role — строки).
3. Swagger UI доступен и позволяет выполнять запросы.
4. Нет утечек секретов: строки подключения в переменных окружения.
5. Локальные тесты покрывают основные сценарии (создание, чтение, обновление, удаление).

Контрольные списки по ролям

Разработчик:

• Запустить проект локально и проверить /docs.
• Добавить unit‑тесты для логики сериализации.
• Проверить обработку ошибок при неверном ID.

DevOps:

• Секреты перенести в переменные окружения.
• Настроить сбор логов и мониторинг (uptime, ошибки).
• Настроить резервное копирование MongoDB.

QA:

• Проверить все эндпоинты в Swagger UI.
• Протестировать граничные случаи: пустые поля, неверный ObjectId.
• Проверить одновременные запросы (конкурентность).

Шпаргалка команд (cheat sheet)

• Создать виртуальное окружение:

virtualenv venv
source venv/bin/activate  # или .\venv\Scripts\Activate.ps1

• Установить зависимости:

pip install fastapi pymongo uvicorn

• Запустить dev‑сервер:

uvicorn server:app --reload --port 8000

• Проверить документацию: http://localhost:8000/docs

Мини‑методология разработки (короткий план)

1. Настройка окружения и БД.
2. Создание Pydantic‑моделей и сериализаторов.
3. Реализация CRUD‑маршрутов.
4. Локальное тестирование через Swagger UI.
5. Добавление логирования, обработчиков ошибок и CI тестов.
6. Подготовка к деплою (секреты, контейнеризация).

Пример простого рабочего сценария тестирования

1. POST /users/ с {“name”: “Alice”, “role”: “admin”} — вернуть созданного пользователя.
2. GET /users/ — убедиться, что Alice присутствует.
3. GET /users/{id} — получить Alice по ID.
4. PUT /users/{id} с изменённым role — проверить обновление.
5. DELETE /users/{id} — удалить Alice и убедиться, что GET /users/{id} возвращает 404.

Безопасность и рекомендации по харднингу

• Валидируйте и нормализуйте входные данные.
• Ограничьте размер тела запроса (rate limiting, body size limits).
• Используйте аутентификацию и авторизацию (JWT, OAuth2) для защищённых маршрутов.
• Скрывайте детальные сообщения об ошибках в production.

Факто‑бокс

• Требования: Python 3.6+; MongoDB порт по умолчанию 27017; Uvicorn служит на порту 8000 по умолчанию.
• Ключевые зависимости: fastapi, pymongo, uvicorn. Для асинхронного доступа — motor.

Decision flow (как выбрать драйвер MongoDB)

flowchart TD
  A[Нужна асинхронность?] -->|Да| B[Использовать Motor]
  A -->|Нет| C[Использовать PyMongo]
  B --> D[Применять async/await в обработчиках]
  C --> E[Выполнять запросы в ThreadPool или использовать синхронные обработчики]

Заключение

FastAPI вместе с MongoDB даёт быстрый путь к рабочему CRUD‑API. Для прототипа достаточно PyMongo и простых маршрутов, но для масштабируемости и полной асинхронности лучше выбрать Motor. Обязательно добавляйте обработку ошибок, логирование и тесты перед деплоем.

Коротко: начните с небольшой структуры проекта, покройте основные CRUD‑сценарии, затем улучшайте безопасность и производительность по мере роста нагрузки.

Важно: всегда храните секреты вне репозитория и используйте окружения/секрет‑менеджеры при деплое.