Приклад створення простого API з використанням MongoDB

Сьогодні ми закріпимо отримані знання на практиці, створивши просте REST API, що взаємодіятиме з MongoDB. Це API стане базою для подальшого вивчення й розробки складніших застосунків.

Наша мета — створити повноцінний застосунок на базі FastAPI, який дозволяє виконувати CRUD-операції з MongoDB. Ми організуємо структуру проєкту, розробимо маршрути для API, протестуємо їх і обговоримо, як це можна застосувати в реальних проєктах.

Організація проєкту — перший крок до спрощення роботи з кодом. Зробимо мініпроєкт API для керування списком задач ("To-Do List"). Ось як виглядає структура папок і файлів:

todo-api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── routes.py
│   ├── database.py
├── .env
├── requirements.txt

• main.py: точка входу в застосунок.
• models.py: моделі даних для MongoDB.
• routes.py: визначення маршрутів API.
• database.py: налаштування підключення до MongoDB.
• .env: зберігання конфіденційних даних, таких як URL бази даних.
• requirements.txt: список залежностей проєкту.

Налаштування оточення

Відкрийте термінал, створіть віртуальне оточення й встановіть залежності:

python -m venv venv
source venv/bin/activate  # Для Windows: venv\Scripts\activate
pip install fastapi[all] motor python-dotenv

Тепер створимо файл requirements.txt для вказівки залежностей:

fastapi[all]
motor
python-dotenv

Додайте файл .env у корінь проєкту. Тут вкажемо рядок підключення до MongoDB:

MONGO_URL=mongodb://localhost:27017
DATABASE_NAME=todo_db

Підключення до MongoDB

Створіть файл database.py:

from motor.motor_asyncio import AsyncIOMotorClient
from dotenv import load_dotenv
import os

load_dotenv()

# Отримуємо змінні з файлу .env
MONGO_URL = os.getenv("MONGO_URL")
DATABASE_NAME = os.getenv("DATABASE_NAME")

# Створюємо MongoDB клієнт
client = AsyncIOMotorClient(MONGO_URL)
database = client[DATABASE_NAME]
todo_collection = database.get_collection("todos")  # Колекція для задач

Ми використовуємо бібліотеку Motor для асинхронних операцій з MongoDB. Тут створюється клієнт MongoDB і колекція, куди будуть зберігатися задачі.

Створення моделей даних

Створіть файл models.py:

from pydantic import BaseModel
from typing import Optional

# Модель для створення нової задачі
class TodoCreate(BaseModel):
    title: str
    description: Optional[str] = None
    is_completed: bool = False

# Модель для повернення задачі користувачу
class TodoResponse(TodoCreate):
    id: str

Ми використовуємо Pydantic для опису схем даних. Модель TodoCreate визначає поля для створення нової задачі, а модель TodoResponse — для повернення даних клієнту.

Розробка маршрутів

Тепер створимо файл routes.py, де визначимо маршрути для CRUD-операцій.

from fastapi import APIRouter, HTTPException
from bson import ObjectId
from app.models import TodoCreate, TodoResponse
from app.database import todo_collection

router = APIRouter()

# Допоміжна функція для перетворення даних з MongoDB
def serialize_todo(todo: dict) -> dict:
    return TodoResponse(
        id=str(todo["_id"]),
        title=todo["title"],
        description=todo.get("description"),
        is_completed=todo["is_completed"]
    )

# 1. Створення задачі
@router.post("/", response_model=TodoResponse)
async def create_task(todo: TodoCreate):
    new_task = todo.dict()
    result = await todo_collection.insert_one(new_task)
    created_task = await todo_collection.find_one({"_id": result.inserted_id})
    return serialize_todo(created_task)

# 2. Отримання всіх задач
@router.get("/", response_model=list[TodoResponse])
async def get_tasks():
    tasks = await todo_collection.find().to_list(100)
    return [serialize_todo(task) for task in tasks]

# 3. Оновлення задачі
@router.put("/{task_id}", response_model=TodoResponse)
async def update_task(task_id: str, todo: TodoCreate):
    if not ObjectId.is_valid(task_id):
        raise HTTPException(status_code=400, detail="Invalid ID")
    
    updated_task = await todo_collection.find_one_and_update(
        {"_id": ObjectId(task_id)},
        {"$set": todo.dict()},
        return_document=True
    )
    if updated_task:
        return serialize_todo(updated_task)
    
    raise HTTPException(status_code=404, detail="Task not found")

# 4. Видалення задачі
@router.delete("/{task_id}")
async def delete_task(task_id: str):
    if not ObjectId.is_valid(task_id):
        raise HTTPException(status_code=400, detail="Invalid ID")
    
    result = await todo_collection.delete_one({"_id": ObjectId(task_id)})
    if result.deleted_count:
        return {"message": "Task deleted"}
    
    raise HTTPException(status_code=404, detail="Task not found")

Запуск застосунку

Створіть файл main.py, який збере все разом:

from fastapi import FastAPI
from app.routes import router as todo_router

app = FastAPI()

# Підключення маршрутів
app.include_router(todo_router, prefix="/todos", tags=["To-Do List"])

@app.get("/")
async def root():
    return {"message": "Welcome to the To-Do List API!"}

Запустіть застосунок, використовуючи Uvicorn:

uvicorn app.main:app --reload

Перейдіть у браузер за адресою http://127.0.0.1:8000/docs, щоб перевірити роботу API через інтерфейс Swagger.

Тестування API

• POST /todos/: Створіть нову задачу за допомогою тіла запиту:

  {
    "title": "Купити молоко",
    "description": "Не забути про знижку на йогурт",
    "is_completed": false
  }

• GET /todos/: Отримайте список усіх задач.
• PUT /todos/{task_id}: Оновіть існуючу задачу, передавши її task_id в URL і нове тіло запиту.
• DELETE /todos/{task_id}: Видаліть задачу по її task_id.

Підсумок

Ми створили просте API для керування задачами, закріпивши знання про MongoDB, FastAPI та асинхронні запити. Цей застосунок може слугувати основою для складніших проєктів — від трекерів задач до великомасштабних систем управління даними.