Сегодня мы закрепим полученные знания на практике, создав простое 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 и асинхронных запросах. Это приложение может использоваться как основа для более сложных проектов — от трекеров задач до крупномасштабных систем управления данными.
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ