Мини-проект: API для блога (CRUD) - IMI-DS

Привет! Пришло время применить все знания на практике. Мы создадим полноценный CRUD API для блога — с созданием, чтением, обновлением и удалением постов. Это отличный способ закрепить навыки работы с FastAPI, Pydantic и тестированием.

В этом проекте мы создадим:

Модели данных для постов
CRUD эндпоинты
Валидацию данных
Тесты для API

Содержание

Что нужно знать перед началом
Основная часть
Структура проекта
Модели данных (schemas.py)
База данных (database.py)
CRUD операции (crud.py)
Основное приложение (main.py)
Тесты (tests/test_main.py)
Запуск проекта
Задачи для доработки
Нюансы и подводные камни
Хранение данных в памяти
Обработка ошибок
Валидация
Частые ошибки и как их избежать
Ошибка 1: Неправильный статус-код
Ошибка 2: Отсутствие проверки существования
Ошибка 3: Неочищенная база в тестах
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого проекта тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание Pydantic и эндпоинтов

Совет: Проект можно использовать как основу для реального блога.

Основная часть

Структура проекта

blog_api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   └── database.py
├── tests/
│   └── test_main.py
├── requirements.txt
└── .env

Совет: Используй виртуальное окружение для изоляции зависимостей.

Модели данных (schemas.py)

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

class PostCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200, description="Заголовок поста")
    content: str = Field(..., min_length=1, description="Содержимое поста")
    author: str = Field(..., min_length=1, description="Автор поста")
    tags: list[str] = Field(default=[], description="Теги поста")

class PostUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    content: Optional[str] = Field(None, min_length=1)
    author: Optional[str] = Field(None, min_length=1)
    tags: Optional[list[str]] = None

class PostResponse(BaseModel):
    id: int
    title: str
    content: str
    author: str
    tags: list[str]
    created_at: datetime
    updated_at: Optional[datetime]

    class Config:
        from_attributes = True

Совет: from_attributes = True позволяет преобразовывать ORM-объекты в Pydantic-модели.

База данных (database.py)

from datetime import datetime
from typing import List, Dict, Optional

# Заглушка базы данных
posts_db: Dict[int, dict] = {}
next_id = 1

def get_all_posts() -> List[dict]:
    return list(posts_db.values())

def get_post(post_id: int) -> Optional[dict]:
    return posts_db.get(post_id)

def create_post(data: dict) -> dict:
    global next_id
    now = datetime.now()
    post = {
        "id": next_id,
        **data,
        "created_at": now,
        "updated_at": None
    }
    posts_db[next_id] = post
    next_id += 1
    return post

def update_post(post_id: int, data: dict) -> Optional[dict]:
    post = posts_db.get(post_id)
    if not post:
        return None
    # Обновляем только переданные поля
    for key, value in data.items():
        if value is not None:
            post[key] = value
    post["updated_at"] = datetime.now()
    return post

def delete_post(post_id: int) -> bool:
    if post_id not in posts_db:
        return False
    del posts_db[post_id]
    return True

Совет: В реальном проекте используй настоящую базу данных (SQLite, PostgreSQL).

CRUD операции (crud.py)

from database import (
    get_all_posts, get_post, create_post,
    update_post, delete_post
)
from schemas import PostCreate, PostUpdate

def get_posts():
    return get_all_posts()

def get_post_by_id(post_id: int):
    return get_post(post_id)

def create_new_post(post: PostCreate):
    return create_post(post.model_dump())

def update_existing_post(post_id: int, post: PostUpdate):
    data = post.model_dump(exclude_unset=True)
    return update_post(post_id, data)

def delete_existing_post(post_id: int):
    return delete_post(post_id)

Совет: model_dump(exclude_unset=True) исключает поля, которые не были переданы.

Основное приложение (main.py)

from fastapi import FastAPI, HTTPException, status
from typing import List
from schemas import PostCreate, PostUpdate, PostResponse
import crud

app = FastAPI(
    title="Blog API",
    description="Простой CRUD API для блога",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Welcome to Blog API!"}

# ======== CREATE ========
@app.post("/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED)
def create_post(post: PostCreate):
    return crud.create_new_post(post)

# ======== READ ========
@app.get("/posts", response_model=List[PostResponse])
def get_posts():
    return crud.get_posts()

@app.get("/posts/{post_id}", response_model=PostResponse)
def get_post(post_id: int):
    post = crud.get_post_by_id(post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

# ======== UPDATE ========
@app.put("/posts/{post_id}", response_model=PostResponse)
def update_post(post_id: int, post: PostUpdate):
    updated = crud.update_existing_post(post_id, post)
    if not updated:
        raise HTTPException(status_code=404, detail="Post not found")
    return updated

# ======== DELETE ========
@app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(post_id: int):
    if not crud.delete_existing_post(post_id):
        raise HTTPException(status_code=404, detail="Post not found")
    return None

Совет: response_model помогает документировать API и валидировать ответы.

Тесты (tests/test_main.py)

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.database import posts_db

client = TestClient(app)

class TestBlogAPI:
    def setup_method(self):
        # Очищаем базу перед каждым тестом
        posts_db.clear()
        global next_id
        next_id = 1

    def test_create_post(self):
        response = client.post(
            "/posts",
            json={
                "title": "First Post",
                "content": "This is my first post",
                "author": "Anna",
                "tags": ["python", "fastapi"]
            }
        )
        assert response.status_code == 201
        data = response.json()
        assert data["title"] == "First Post"
        assert data["author"] == "Anna"
        assert "id" in data
        assert "created_at" in data

    def test_get_posts(self):
        # Создаём пост
        client.post("/posts", json={
            "title": "Test Post",
            "content": "Content",
            "author": "Test"
        })
        response = client.get("/posts")
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["title"] == "Test Post"

    def test_get_post_not_found(self):
        response = client.get("/posts/999")
        assert response.status_code == 404
        assert response.json()["detail"] == "Post not found"

    def test_update_post(self):
        # Создаём пост
        create_resp = client.post("/posts", json={
            "title": "Old Title",
            "content": "Old content",
            "author": "Author"
        })
        post_id = create_resp.json()["id"]

        # Обновляем
        response = client.put(
            f"/posts/{post_id}",
            json={"title": "New Title", "content": "New content"}
        )
        assert response.status_code == 200
        data = response.json()
        assert data["title"] == "New Title"
        assert data["content"] == "New content"
        assert data["updated_at"] is not None

    def test_delete_post(self):
        # Создаём пост
        create_resp = client.post("/posts", json={
            "title": "To Delete",
            "content": "Content",
            "author": "Author"
        })
        post_id = create_resp.json()["id"]

        # Удаляем
        response = client.delete(f"/posts/{post_id}")
        assert response.status_code == 204

        # Проверяем, что поста нет
        response = client.get(f"/posts/{post_id}")
        assert response.status_code == 404

    def test_create_post_validation_error(self):
        # Пустой заголовок
        response = client.post("/posts", json={
            "title": "",
            "content": "Content",
            "author": "Author"
        })
        assert response.status_code == 422

    def test_update_post_partial(self):
        # Создаём пост
        create_resp = client.post("/posts", json={
            "title": "Original",
            "content": "Original content",
            "author": "Original author",
            "tags": ["original"]
        })
        post_id = create_resp.json()["id"]

        # Обновляем только заголовок
        response = client.put(
            f"/posts/{post_id}",
            json={"title": "Updated Title Only"}
        )
        assert response.status_code == 200
        data = response.json()
        assert data["title"] == "Updated Title Only"
        assert data["content"] == "Original content"
        assert data["author"] == "Original author"
        assert data["tags"] == ["original"]

Совет: Тесты покрывают основные сценарии, включая ошибки и частичные обновления.

Запуск проекта

Установка зависимостей (requirements.txt):

fastapi==0.104.1
uvicorn==0.24.0
pytest==7.4.3
httpx==0.25.1

Запуск приложения:

uvicorn app.main:app --reload

Запуск тестов:

pytest tests/

Тестирование в Swagger:

Открой http://localhost:8000/docs

Задачи для доработки

Задача 1. Добавь поле published (bool) для постов.

Задача 2. Добавь поиск по заголовку: GET /posts?search=python.

Задача 3. Добавь пагинацию: GET /posts?skip=0&limit=10.

Задача 4. Создай эндпоинт GET /posts/{post_id}/comments.

Задача 5. Добавь фильтрацию по тегам.

Ответы:

Задача 1.

# В PostCreate и PostUpdate добавляем:
published: bool = False

Задача 2.

@app.get("/posts")
def get_posts(search: str = None):
    posts = crud.get_posts()
    if search:
        posts = [p for p in posts if search.lower() in p["title"].lower()]
    return posts

Задача 3.

@app.get("/posts")
def get_posts(skip: int = 0, limit: int = 10):
    posts = crud.get_posts()[skip:skip+limit]
    return posts

Задача 4.

@app.get("/posts/{post_id}/comments")
def get_comments(post_id: int):
    # Заглушка
    return [{"id": 1, "text": "Nice post!"}]

Задача 5.

@app.get("/posts")
def get_posts(tag: str = None):
    posts = crud.get_posts()
    if tag:
        posts = [p for p in posts if tag in p.get("tags", [])]
    return posts

Нюансы и подводные камни

Хранение данных в памяти

Данные теряются при перезапуске. Используй SQLite или PostgreSQL для продакшена.

Обработка ошибок

Всегда обрабатывай 404 и другие статусы.

Валидация

Используй Pydantic для валидации входных данных.

Частые ошибки и как их избежать

Ошибка 1: Неправильный статус-код

Используй 201 для создания, 204 для удаления.

Ошибка 2: Отсутствие проверки существования

Всегда проверяй, существует ли пост перед обновлением или удалением.

Ошибка 3: Неочищенная база в тестах

Используй setup_method для очистки данных.

Шпаргалка

Эндпоинт	Метод	Назначение
`/posts`	POST	Создать пост
`/posts`	GET	Получить все посты
`/posts/{id}`	GET	Получить пост
`/posts/{id}`	PUT	Обновить пост
`/posts/{id}`	DELETE	Удалить пост