#fastapi #python #api #sqlalchemy #pydantic #backend

Создаём API мотивационных цитат на FastAPI

Константин Потапов

30 ноября 2025 г.

25 мин

Пошаговый гайд по созданию REST API для мотивационных цитат с помощью FastAPI, SQLAlchemy и Pydantic. От базовой структуры до production-ready решения с валидацией, миграциями и документацией.

В этом гайде создадим полноценное REST API для получения мотивационных цитат. Приложение будет готово к production: с валидацией, миграциями БД, автодокументацией и покрытием тестами.

Результат:

🚀 Быстрый FastAPI backend с async/await
🗄️ PostgreSQL с SQLAlchemy 2.0
✅ Pydantic для валидации данных
📚 Автогенерируемая OpenAPI документация
🧪 Покрытие тестами с pytest
🔄 Миграции с Alembic

Для кого: Разработчики знакомые с Python, хотят освоить FastAPI и современный подход к созданию API.

Архитектура приложения

Структура проекта: почему именно так?

quotes-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # Точка входа FastAPI
│   ├── config.py            # Настройки (DATABASE_URL, etc.)
│   ├── database.py          # Подключение к БД
│   ├── models.py            # SQLAlchemy модели (БД таблицы)
│   ├── schemas.py           # Pydantic схемы (DTO для API)
│   ├── crud.py              # CRUD операции
│   └── routers/
│       ├── __init__.py
│       └── quotes.py        # Endpoints для цитат
├── alembic/                 # Миграции БД
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_quotes.py
├── requirements.txt
├── .env                     # Переменные окружения
└── README.md

Важно понимать: models.py и schemas.py — это НЕ одно и то же!

models.py vs schemas.py: в чём разница?

Это частый источник путаницы для новичков в FastAPI. Разберём подробно:

models.py — SQLAlchemy модели (Database layer)

Описывают структуру таблиц в БД
Работают с базой данных напрямую
Содержат типы SQLAlchemy: Integer, String, DateTime
Пример: id: Mapped[int] = mapped_column(Integer, primary_key=True)

schemas.py — Pydantic модели (API layer, DTO)

Описывают формат данных для API (Data Transfer Objects)
Валидируют входящие/исходящие JSON
Содержат типы Python: int, str, datetime
Пример: id: int, content: str

Почему их разделяют?

Разные ответственности — модель БД может иметь поля, которые не должны возвращаться в API (например, password_hash)
Разная валидация — API может требовать email, а БД просто хранит строку
Гибкость — можно изменить формат API не меняя структуру БД и наоборот

Пример:

# models.py (БД)
class User(Base):
    id: Mapped[int]
    email: Mapped[str]
    password_hash: Mapped[str]  # ⚠️ Не должен попасть в API!
    created_at: Mapped[datetime]
 
# schemas.py (API)
class UserResponse(BaseModel):
    id: int
    email: str  # ✅ Без пароля!
    # created_at можно скрыть от пользователя

Почему не выносим DTO в отдельную папку `dto/`?

Для небольших проектов (как наш):

✅ Один файл schemas.py — проще навигация
✅ Все Pydantic схемы в одном месте
✅ Меньше вложенности (app/schemas.py вместо app/dto/quote_dto.py)

Для крупных проектов (10+ сущностей):

app/
├── quotes/
│   ├── models.py       # Quote SQLAlchemy модель
│   ├── schemas.py      # QuoteCreate, QuoteResponse
│   ├── crud.py         # CRUD для quotes
│   └── router.py       # Endpoints
├── users/
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   └── router.py

Рекомендация: Начинайте с плоской структуры (app/schemas.py), переходите к модульной при росте проекта (5+ файлов в app/).

Почему `crud.py` отдельно от `models.py`?

models.py — что (структура данных) crud.py — как (операции с данными)

# models.py — декларативно описываем таблицу
class Quote(Base):
    id: Mapped[int]
    content: Mapped[str]
 
# crud.py — императивно описываем логику
async def get_quote(db, quote_id):
    result = await db.execute(...)
    return result.scalar_one_or_none()

Зачем разделять?

Single Responsibility Principle — модель знает только о структуре
Тестируемость — легко мокировать CRUD функции
Переиспользование — одна модель, разные CRUD (admin vs user API)

Стек технологий

Компонент	Технология	Версия
Framework	FastAPI	0.109+
ORM	SQLAlchemy	2.0+
Validation	Pydantic	2.0+
Database	PostgreSQL	15+
Migrations	Alembic	1.13+
Testing	pytest + httpx	latest
Server	uvicorn	0.27+

Шаг 1: Установка зависимостей

requirements.txt

# Core
fastapi==0.109.0
uvicorn[standard]==0.27.0
python-dotenv==1.0.0
 
# Database
sqlalchemy==2.0.25
psycopg2-binary==2.9.9
alembic==1.13.1
 
# Validation
pydantic==2.5.3
pydantic-settings==2.1.0
 
# Testing
pytest==7.4.4
pytest-asyncio==0.23.3
httpx==0.26.0

Установка

Важно: В примерах используем pip для совместимости и простоты, но в современных проектах рекомендуется использовать Poetry или uv для управления зависимостями.

С pip (традиционный способ):

# Создаём виртуальное окружение
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows
 
# Устанавливаем зависимости
pip install -r requirements.txt

С Poetry (рекомендуется для новых проектов):

# Инициализация проекта
poetry init
 
# Добавление зависимостей
poetry add fastapi uvicorn[standard] sqlalchemy psycopg2-binary
 
# Установка и активация окружения
poetry install
poetry shell

С uv (самый быстрый менеджер пакетов):

# Создание проекта
uv venv
source .venv/bin/activate
 
# Установка зависимостей
uv pip install -r requirements.txt
 
# Или напрямую
uv pip install fastapi uvicorn[standard] sqlalchemy

Подробнее о современных инструментах: См. Руководство по Poetry и uv — сравнение pip, Poetry, uv с практическими примерами.

Почему Poetry/uv лучше pip?

pip:

❌ Не гарантирует воспроизводимость (нет lock-файла)
❌ Медленная установка зависимостей
❌ Ручное управление requirements.txt и requirements-dev.txt

Poetry:

✅ poetry.lock гарантирует одинаковые версии
✅ Автоматическое разрешение конфликтов зависимостей
✅ Встроенное разделение dev/prod зависимостей

uv:

✅ В 10-100 раз быстрее pip (написан на Rust)
✅ Совместим с pip (замена drop-in)
✅ Автоматическое управление виртуальными окружениями

Рекомендация: Для обучения используйте pip, для production проектов — Poetry или uv.

Шаг 2: Настройка окружения

.env файл

# Database
DATABASE_URL=postgresql://postgres:password@localhost:5432/quotes_db
 
# App
APP_NAME="Quotes API"
APP_VERSION="1.0.0"
DEBUG=True
 
# CORS (для фронтенда)
ALLOWED_ORIGINS=http://localhost:3000,http://localhost:8000

app/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict
 
class Settings(BaseSettings):
    """Настройки приложения из переменных окружения"""
 
    # Database
    database_url: str
 
    # App
    app_name: str = "Quotes API"
    app_version: str = "1.0.0"
    debug: bool = False
 
    # CORS
    allowed_origins: list[str] = ["http://localhost:3000"]
 
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )
 
settings = Settings()

Шаг 3: Подключение к БД

app/database.py

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    create_async_engine,
    async_sessionmaker,
)
from sqlalchemy.orm import DeclarativeBase
from app.config import settings
 
# Создаём async engine
# Заменяем postgresql:// на postgresql+asyncpg://
DATABASE_URL = settings.database_url.replace(
    "postgresql://", "postgresql+asyncpg://"
)
 
engine = create_async_engine(
    DATABASE_URL,
    echo=settings.debug,  # Логирование SQL в dev
    future=True,
)
 
# Session factory
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
 
# Base class для моделей
class Base(DeclarativeBase):
    pass
 
# Dependency для получения сессии
async def get_db() -> AsyncSession:
    """Создаёт и закрывает сессию БД"""
    async with AsyncSessionLocal() as session:
        yield session

Почему async/await?

Традиционный sync подход:

# Блокирует весь процесс на время запроса к БД
result = db.execute(query)  # ⏸️ Все остальные запросы ждут

Async подход:

# Пока ждём БД, обрабатываем другие запросы
result = await db.execute(query)  # ⚡ Параллелим работу

Результат: FastAPI может обрабатывать тысячи одновременных запросов на одном воркере.

Зачем `expire_on_commit=False`?

По умолчанию SQLAlchemy инвалидирует все объекты после commit():

quote = Quote(content="Test")
db.add(quote)
await db.commit()
print(quote.id)  # ❌ DetachedInstanceError — объект отвязан от сессии!

С expire_on_commit=False:

quote = Quote(content="Test")
db.add(quote)
await db.commit()
print(quote.id)  # ✅ Работает — объект всё ещё доступен

Важно: В FastAPI это безопасно, потому что каждый request получает свою сессию.

Зачем `get_db()` как dependency?

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

Без dependency (плохо):

@router.get("/quotes/{id}")
async def get_quote(id: int):
    db = AsyncSessionLocal()  # ❌ Нужно вручную закрывать!
    quote = await db.get(Quote, id)
    await db.close()  # ❌ Забыл — утечка соединений!
    return quote

С dependency (хорошо):

@router.get("/quotes/{id}")
async def get_quote(id: int, db: AsyncSession = Depends(get_db)):
    quote = await db.get(Quote, id)  # ✅ Сессия автоматически закроется
    return quote

Бонус: В тестах легко подменить get_db() на тестовую БД.

Важно: Используем asyncpg драйвер для async работы с PostgreSQL.

pip install asyncpg

Шаг 4: Модели БД

app/models.py

from datetime import datetime
from sqlalchemy import String, Text, DateTime, Integer
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base
 
class Quote(Base):
    """Модель цитаты в БД"""
 
    __tablename__ = "quotes"
 
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    author: Mapped[str] = mapped_column(String(200), nullable=False, index=True)
    category: Mapped[str | None] = mapped_column(String(100), nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        default=datetime.utcnow,
        nullable=False,
    )
 
    def __repr__(self) -> str:
        return f"<Quote(id={self.id}, author='{self.author}')>"

Почему SQLAlchemy 2.0 стиль?

Typed mappings с Mapped[T]
Лучшая поддержка type hints
Меньше runtime ошибок

Шаг 5: Pydantic схемы

app/schemas.py

from datetime import datetime
from pydantic import BaseModel, Field, ConfigDict
 
class QuoteBase(BaseModel):
    """Базовая схема цитаты"""
 
    content: str = Field(..., min_length=1, max_length=1000)
    author: str = Field(..., min_length=1, max_length=200)
    category: str | None = Field(None, max_length=100)
 
class QuoteCreate(QuoteBase):
    """Схема для создания цитаты"""
    pass
 
class QuoteUpdate(BaseModel):
    """Схема для обновления цитаты"""
 
    content: str | None = Field(None, min_length=1, max_length=1000)
    author: str | None = Field(None, min_length=1, max_length=200)
    category: str | None = Field(None, max_length=100)
 
class QuoteResponse(QuoteBase):
    """Схема ответа с цитатой"""
 
    id: int
    created_at: datetime
 
    model_config = ConfigDict(from_attributes=True)
 
class QuoteList(BaseModel):
    """Схема списка цитат"""
 
    quotes: list[QuoteResponse]
    total: int

Зачем три разные схемы?

Это паттерн Input-Output segregation. Разберём на примере:

1. QuoteCreate (Input для POST)

class QuoteCreate(BaseModel):
    content: str = Field(..., min_length=1)
    author: str = Field(..., min_length=1)

Почему все поля обязательны?

При создании цитаты обязаны передать контент и автора
Field(...) означает "обязательное поле"
min_length=1 защищает от пустых строк

2. QuoteUpdate (Input для PATCH)

class QuoteUpdate(BaseModel):
    content: str | None = None  # Опционально!
    author: str | None = None   # Опционально!

Почему все поля опциональны?

PATCH = частичное обновление
Хотим изменить только автора → передаём {"author": "New"}
Хотим изменить только контент → передаём {"content": "New"}

Без разделения схем:

# ❌ Плохо — одна схема для всего
await client.post("/quotes/", json={})  # Создаст пустую цитату!

3. QuoteResponse (Output для GET)

class QuoteResponse(QuoteBase):
    id: int              # ✅ Добавили ID (генерируется БД)
    created_at: datetime # ✅ Добавили timestamp

Зачем отдельная схема для ответа?

Клиент не отправляет id при создании (его генерирует БД)
Клиент не отправляет created_at (его ставит БД)
Но в ответе мы обязаны вернуть эти поля

model_config = ConfigDict(from_attributes=True) — магия Pydantic:

# Без from_attributes:
db_quote = Quote(id=1, content="Test")  # SQLAlchemy объект
return QuoteResponse(**db_quote)  # ❌ Ошибка!
 
# С from_attributes:
return QuoteResponse.model_validate(db_quote)  # ✅ Работает!

FastAPI автоматически вызывает model_validate() при response_model=QuoteResponse.

Шаг 6: CRUD операции

app/crud.py

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import Quote
from app.schemas import QuoteCreate, QuoteUpdate
 
async def create_quote(db: AsyncSession, quote: QuoteCreate) -> Quote:
    """Создаёт новую цитату"""
 
    # Конвертируем Pydantic модель в SQLAlchemy модель
    db_quote = Quote(**quote.model_dump())
 
    db.add(db_quote)           # Добавляем в сессию
    await db.commit()          # Сохраняем в БД
    await db.refresh(db_quote) # Обновляем объект (получаем ID)
 
    return db_quote
 
async def get_quote(db: AsyncSession, quote_id: int) -> Quote | None:
    """Получает цитату по ID"""
 
    result = await db.execute(select(Quote).where(Quote.id == quote_id))
    return result.scalar_one_or_none()  # None если не найдена
 
async def get_quotes(
    db: AsyncSession,
    skip: int = 0,
    limit: int = 100,
    author: str | None = None,
    category: str | None = None,
) -> list[Quote]:
    """Получает список цитат с фильтрацией"""
 
    query = select(Quote)
 
    if author:
        query = query.where(Quote.author.ilike(f"%{author}%"))
 
    if category:
        query = query.where(Quote.category == category)
 
    query = query.offset(skip).limit(limit)
 
    result = await db.execute(query)
    return list(result.scalars().all())
 
async def get_quotes_count(
    db: AsyncSession,
    author: str | None = None,
    category: str | None = None,
) -> int:
    """Подсчитывает общее количество цитат"""
 
    query = select(func.count(Quote.id))
 
    if author:
        query = query.where(Quote.author.ilike(f"%{author}%"))
 
    if category:
        query = query.where(Quote.category == category)
 
    result = await db.execute(query)
    return result.scalar_one()
 
async def get_random_quote(db: AsyncSession) -> Quote | None:
    """Получает случайную цитату"""
 
    # PostgreSQL specific: ORDER BY RANDOM()
    query = select(Quote).order_by(func.random()).limit(1)
 
    result = await db.execute(query)
    return result.scalar_one_or_none()
 
async def update_quote(
    db: AsyncSession,
    quote_id: int,
    quote_update: QuoteUpdate,
) -> Quote | None:
    """Обновляет цитату"""
 
    db_quote = await get_quote(db, quote_id)
 
    if not db_quote:
        return None
 
    update_data = quote_update.model_dump(exclude_unset=True)
 
    for field, value in update_data.items():
        setattr(db_quote, field, value)
 
    await db.commit()
    await db.refresh(db_quote)
    return db_quote
 
async def delete_quote(db: AsyncSession, quote_id: int) -> bool:
    """Удаляет цитату"""
 
    db_quote = await get_quote(db, quote_id)
 
    if not db_quote:
        return False
 
    await db.delete(db_quote)
    await db.commit()
    return True

Ключевые моменты CRUD

1. Почему quote.model_dump() вместо quote.dict()?

В Pydantic v2 изменилась API:

# Pydantic v1 (старый способ)
quote.dict()  # ❌ Deprecated
 
# Pydantic v2 (новый способ)
quote.model_dump()  # ✅ Актуальный метод

2. Зачем await db.refresh(db_quote) после commit?

db_quote = Quote(content="Test")  # id = None (ещё не сохранили)
db.add(db_quote)
await db.commit()                 # БД генерирует id = 1
 
# Без refresh:
print(db_quote.id)  # None ❌
 
# С refresh:
await db.refresh(db_quote)
print(db_quote.id)  # 1 ✅

3. Почему scalar_one_or_none() а не first() или one()?

SQLAlchemy 2.0 style — три варианта:

result = await db.execute(select(Quote).where(...))
 
# .one() — строго одна запись, иначе ошибка
quote = result.scalar_one()  # ❌ NoResultFound если пусто
 
# .one_or_none() — одна запись или None
quote = result.scalar_one_or_none()  # ✅ None если не найдена
 
# .first() — первая запись или None (deprecated в 2.0)
quote = result.first()  # ⚠️ Лучше не использовать

4. Зачем query.offset(skip).limit(limit)?

Это пагинация (разбивка на страницы):

# Первая страница: 10 цитат
get_quotes(skip=0, limit=10)  # Записи 1-10
 
# Вторая страница: следующие 10
get_quotes(skip=10, limit=10)  # Записи 11-20
 
# Третья страница
get_quotes(skip=20, limit=10)  # Записи 21-30

Формула: skip = (page - 1) * limit

5. Почему update_data = quote_update.model_dump(exclude_unset=True)?

Без exclude_unset:

# Клиент отправил только {"author": "New"}
quote_update = QuoteUpdate(author="New")
 
quote_update.model_dump()
# {"author": "New", "content": None, "category": None}
# ❌ Затрёт content и category значениями None!

С exclude_unset=True:

quote_update.model_dump(exclude_unset=True)
# {"author": "New"}
# ✅ Обновит только author, остальное не трогает

Шаг 7: API Endpoints

app/routers/quotes.py

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
 
from app import crud, schemas
from app.database import get_db
 
router = APIRouter(
    prefix="/quotes",
    tags=["quotes"],
)
 
@router.post(
    "/",
    response_model=schemas.QuoteResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Создать цитату",
)
async def create_quote(
    quote: schemas.QuoteCreate,
    db: AsyncSession = Depends(get_db),
):
    """Создаёт новую цитату"""
    return await crud.create_quote(db, quote)
 
@router.get(
    "/",
    response_model=schemas.QuoteList,
    summary="Получить список цитат",
)
async def list_quotes(
    skip: int = Query(0, ge=0, description="Пропустить N записей"),
    limit: int = Query(10, ge=1, le=100, description="Лимит записей"),
    author: str | None = Query(None, description="Фильтр по автору"),
    category: str | None = Query(None, description="Фильтр по категории"),
    db: AsyncSession = Depends(get_db),
):
    """Получает список цитат с пагинацией и фильтрацией"""
 
    quotes = await crud.get_quotes(
        db,
        skip=skip,
        limit=limit,
        author=author,
        category=category,
    )
 
    total = await crud.get_quotes_count(db, author=author, category=category)
 
    return schemas.QuoteList(quotes=quotes, total=total)
 
@router.get(
    "/random",
    response_model=schemas.QuoteResponse,
    summary="Получить случайную цитату",
)
async def random_quote(db: AsyncSession = Depends(get_db)):
    """Получает случайную цитату"""
 
    quote = await crud.get_random_quote(db)
 
    if not quote:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No quotes found",
        )
 
    return quote
 
@router.get(
    "/{quote_id}",
    response_model=schemas.QuoteResponse,
    summary="Получить цитату по ID",
)
async def get_quote(
    quote_id: int,
    db: AsyncSession = Depends(get_db),
):
    """Получает цитату по ID"""
 
    quote = await crud.get_quote(db, quote_id)
 
    if not quote:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Quote {quote_id} not found",
        )
 
    return quote
 
@router.patch(
    "/{quote_id}",
    response_model=schemas.QuoteResponse,
    summary="Обновить цитату",
)
async def update_quote(
    quote_id: int,
    quote_update: schemas.QuoteUpdate,
    db: AsyncSession = Depends(get_db),
):
    """Частично обновляет цитату"""
 
    quote = await crud.update_quote(db, quote_id, quote_update)
 
    if not quote:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Quote {quote_id} not found",
        )
 
    return quote
 
@router.delete(
    "/{quote_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Удалить цитату",
)
async def delete_quote(
    quote_id: int,
    db: AsyncSession = Depends(get_db),
):
    """Удаляет цитату"""
 
    deleted = await crud.delete_quote(db, quote_id)
 
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Quote {quote_id} not found",
        )

Почему именно такие декораторы?

1. response_model=schemas.QuoteResponse

@router.get("/quotes/{quote_id}", response_model=schemas.QuoteResponse)
async def get_quote(quote_id: int, db: AsyncSession = Depends(get_db)):
    return await crud.get_quote(db, quote_id)

Что делает response_model?

✅ Валидирует ответ (если функция вернёт что-то не то — ошибка)
✅ Конвертирует SQLAlchemy объект в JSON
✅ Фильтрует поля (если в модели есть password_hash, его не отдаст)
✅ Генерирует OpenAPI схему для документации

Без response_model (плохо):

return db_quote  # ❌ Вернёт SQLAlchemy объект, JSON не сериализуется!

2. status_code=status.HTTP_201_CREATED

REST API семантика:

# POST (создание) → 201 Created
@router.post("/", status_code=201)
 
# GET (чтение) → 200 OK (по умолчанию)
@router.get("/")
 
# PATCH (обновление) → 200 OK
@router.patch("/{id}")
 
# DELETE (удаление) → 204 No Content
@router.delete("/{id}", status_code=204)

Зачем status.HTTP_201_CREATED вместо 201?

# Магические числа (плохо)
@router.post("/", status_code=201)  # ❌ Что означает 201?
 
# Константы (хорошо)
@router.post("/", status_code=status.HTTP_201_CREATED)  # ✅ Понятно!

3. Query() для валидации параметров

skip: int = Query(0, ge=0, description="Пропустить N записей")
limit: int = Query(10, ge=1, le=100, description="Лимит записей")

Зачем Query() если можно просто skip: int = 0?

Без Query:

async def list_quotes(skip: int = 0, limit: int = 10):
    # ❌ Клиент может отправить skip=-10, limit=999999

С Query:

async def list_quotes(
    skip: int = Query(0, ge=0),      # ✅ >= 0
    limit: int = Query(10, ge=1, le=100)  # ✅ от 1 до 100
):

Если клиент отправит ?limit=500, получит 422 Unprocessable Entity:

{
  "detail": [
    {
      "loc": ["query", "limit"],
      "msg": "ensure this value is less than or equal to 100",
      "type": "value_error.number.not_le"
    }
  ]
}

4. Зачем summary и tags?

@router.get("/random", tags=["quotes"], summary="Получить случайную цитату")

Генерирует красивую документацию:

tags — группирует endpoints в Swagger UI
summary — краткое описание в списке
Docstring — подробное описание при раскрытии

Основные принципы:

Используем Depends(get_db) для инъекции сессии
Query() для параметров запроса с валидацией
HTTP статус коды (201, 404, 204)
Описания для автодокументации

Шаг 8: Главный файл приложения

app/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
 
from app.config import settings
from app.database import engine, Base
from app.routers import quotes
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifecycle events: startup and shutdown"""
 
    # Startup: создаём таблицы (только для dev!)
    # В production используем Alembic миграции
    if settings.debug:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
 
    yield
 
    # Shutdown: закрываем соединения
    await engine.dispose()
 
app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    description="REST API для мотивационных цитат",
    lifespan=lifespan,
)
 
# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
# Подключаем роутеры
app.include_router(quotes.router)
 
@app.get("/", tags=["root"])
async def root():
    """Health check endpoint"""
    return {
        "app": settings.app_name,
        "version": settings.app_version,
        "status": "running",
    }

Зачем нужен lifespan?

Проблема: Нужно выполнять код при старте/остановке приложения.

Старый способ (deprecated):

@app.on_event("startup")
async def startup():
    # Код при запуске
 
@app.on_event("shutdown")
async def shutdown():
    # Код при остановке

Новый способ (рекомендуемый):

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ⬇️ Код ДО yield — startup
    print("Starting up...")
    yield
    # ⬇️ Код ПОСЛЕ yield — shutdown
    print("Shutting down...")

Зачем await engine.dispose() в shutdown?

# Закрываем все соединения с БД при остановке
await engine.dispose()

Без dispose:

Соединения с PostgreSQL останутся открытыми
При перезапуске может исчерпаться connection pool
В production это утечка ресурсов

Зачем CORS middleware?

Проблема: Фронтенд на localhost:3000 не может обращаться к API на localhost:8000.

Браузер блокирует запросы:

Access to fetch at 'http://localhost:8000/quotes'
from origin 'http://localhost:3000' has been blocked by CORS policy

Решение:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Разрешаем фронтенд
    allow_credentials=True,
    allow_methods=["*"],  # Все HTTP методы (GET, POST, etc.)
    allow_headers=["*"],  # Все заголовки
)

Production настройка:

# .env
ALLOWED_ORIGINS=https://myapp.com,https://www.myapp.com
 
# config.py
allowed_origins: list[str] = ["https://myapp.com"]

⚠️ Никогда не используйте allow_origins=["*"] в production!

Шаг 9: Миграции с Alembic

Инициализация Alembic

alembic init alembic

alembic/env.py

Обновите env.py для поддержки async:

from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
 
from app.config import settings
from app.database import Base
from app.models import Quote  # Импортируем модели!
 
# Alembic Config
config = context.config
 
# Устанавливаем DATABASE_URL из настроек
config.set_main_option("sqlalchemy.url", settings.database_url)
 
if config.config_file_name is not None:
    fileConfig(config.config_file_name)
 
target_metadata = Base.metadata
 
def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
 
    with context.begin_transaction():
        context.run_migrations()
 
async def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
 
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
 
    await connectable.dispose()
 
def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
 
    with context.begin_transaction():
        context.run_migrations()
 
if context.is_offline_mode():
    run_migrations_offline()
else:
    import asyncio
    asyncio.run(run_migrations_online())

Создание первой миграции

# Автогенерация миграции
alembic revision --autogenerate -m "Create quotes table"
 
# Применение миграции
alembic upgrade head

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

Локальный запуск

# Development режим с hot reload
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

Заполнение тестовыми данными

Создайте scripts/seed_data.py:

import asyncio
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import AsyncSessionLocal
from app.models import Quote
 
quotes_data = [
    {
        "content": "Be yourself; everyone else is already taken.",
        "author": "Oscar Wilde",
        "category": "inspiration",
    },
    {
        "content": "The only way to do great work is to love what you do.",
        "author": "Steve Jobs",
        "category": "motivation",
    },
    {
        "content": "Life is what happens when you're busy making other plans.",
        "author": "John Lennon",
        "category": "life",
    },
]
 
async def seed_quotes():
    """Добавляет тестовые цитаты в БД"""
 
    async with AsyncSessionLocal() as db:
        for data in quotes_data:
            quote = Quote(**data)
            db.add(quote)
 
        await db.commit()
        print(f"✅ Added {len(quotes_data)} quotes")
 
if __name__ == "__main__":
    asyncio.run(seed_quotes())

Запуск:

python scripts/seed_data.py

Проверка API

Откройте автодокументацию:

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc

Шаг 11: Тестирование

tests/conftest.py

import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 
from app.main import app
from app.database import Base, get_db
 
# Тестовая БД (in-memory SQLite)
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
 
@pytest.fixture(scope="function")
async def db_session():
    """Создаёт тестовую БД сессию"""
 
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
 
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
 
    TestSessionLocal = async_sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )
 
    async with TestSessionLocal() as session:
        yield session
 
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
 
    await engine.dispose()
 
@pytest.fixture(scope="function")
async def client(db_session):
    """Создаёт тестовый HTTP клиент"""
 
    async def override_get_db():
        yield db_session
 
    app.dependency_overrides[get_db] = override_get_db
 
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac
 
    app.dependency_overrides.clear()

tests/test_quotes.py

import pytest
 
@pytest.mark.asyncio
async def test_create_quote(client):
    """Тест создания цитаты"""
 
    response = await client.post(
        "/quotes/",
        json={
            "content": "Test quote",
            "author": "Test Author",
            "category": "test",
        },
    )
 
    assert response.status_code == 201
    data = response.json()
    assert data["content"] == "Test quote"
    assert data["author"] == "Test Author"
    assert "id" in data
 
@pytest.mark.asyncio
async def test_get_quote(client):
    """Тест получения цитаты"""
 
    # Создаём цитату
    create_response = await client.post(
        "/quotes/",
        json={"content": "Get test", "author": "Author"},
    )
    quote_id = create_response.json()["id"]
 
    # Получаем цитату
    response = await client.get(f"/quotes/{quote_id}")
 
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == quote_id
    assert data["content"] == "Get test"
 
@pytest.mark.asyncio
async def test_get_nonexistent_quote(client):
    """Тест получения несуществующей цитаты"""
 
    response = await client.get("/quotes/999")
 
    assert response.status_code == 404
    assert "not found" in response.json()["detail"].lower()
 
@pytest.mark.asyncio
async def test_list_quotes(client):
    """Тест получения списка цитат"""
 
    # Создаём несколько цитат
    for i in range(3):
        await client.post(
            "/quotes/",
            json={"content": f"Quote {i}", "author": f"Author {i}"},
        )
 
    # Получаем список
    response = await client.get("/quotes/")
 
    assert response.status_code == 200
    data = response.json()
    assert data["total"] == 3
    assert len(data["quotes"]) == 3
 
@pytest.mark.asyncio
async def test_filter_quotes_by_author(client):
    """Тест фильтрации по автору"""
 
    await client.post(
        "/quotes/",
        json={"content": "Quote 1", "author": "Oscar Wilde"},
    )
    await client.post(
        "/quotes/",
        json={"content": "Quote 2", "author": "Steve Jobs"},
    )
 
    response = await client.get("/quotes/?author=Oscar")
 
    assert response.status_code == 200
    data = response.json()
    assert data["total"] == 1
    assert data["quotes"][0]["author"] == "Oscar Wilde"
 
@pytest.mark.asyncio
async def test_update_quote(client):
    """Тест обновления цитаты"""
 
    # Создаём цитату
    create_response = await client.post(
        "/quotes/",
        json={"content": "Original", "author": "Author"},
    )
    quote_id = create_response.json()["id"]
 
    # Обновляем
    response = await client.patch(
        f"/quotes/{quote_id}",
        json={"content": "Updated"},
    )
 
    assert response.status_code == 200
    data = response.json()
    assert data["content"] == "Updated"
    assert data["author"] == "Author"  # Не изменился
 
@pytest.mark.asyncio
async def test_delete_quote(client):
    """Тест удаления цитаты"""
 
    # Создаём цитату
    create_response = await client.post(
        "/quotes/",
        json={"content": "Delete me", "author": "Author"},
    )
    quote_id = create_response.json()["id"]
 
    # Удаляем
    response = await client.delete(f"/quotes/{quote_id}")
 
    assert response.status_code == 204
 
    # Проверяем что удалена
    get_response = await client.get(f"/quotes/{quote_id}")
    assert get_response.status_code == 404
 
@pytest.mark.asyncio
async def test_random_quote(client):
    """Тест получения случайной цитаты"""
 
    # Создаём цитату
    await client.post(
        "/quotes/",
        json={"content": "Random quote", "author": "Author"},
    )
 
    # Получаем случайную
    response = await client.get("/quotes/random")
 
    assert response.status_code == 200
    data = response.json()
    assert "content" in data
    assert "author" in data

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

# Установка зависимостей для SQLite
pip install aiosqlite
 
# Запуск тестов
pytest
 
# С coverage
pytest --cov=app --cov-report=term-missing

Production Deployment

Docker

Dockerfile:

FROM python:3.11-slim
 
WORKDIR /app
 
# Зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# Код приложения
COPY app ./app
COPY alembic ./alembic
COPY alembic.ini .
 
# Запуск миграций и сервера
CMD alembic upgrade head && \
    uvicorn app.main:app --host 0.0.0.0 --port 8000

docker-compose.yml:

version: "3.8"
 
services:
  db:
    image: postgres:15
    environment:
      POSTGRES_DB: quotes_db
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
 
  api:
    build: .
    environment:
      DATABASE_URL: postgresql://postgres:password@db:5432/quotes_db
      APP_NAME: "Quotes API"
      DEBUG: "False"
    ports:
      - "8000:8000"
    depends_on:
      - db
 
volumes:
  postgres_data:

Запуск:

docker-compose up --build

Production настройки

Рекомендации для production:

Отключите debug режим: DEBUG=False в .env
Используйте Alembic для миграций (не create_all())
Настройте connection pool:

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=0,
    pool_pre_ping=True,
)

Добавьте rate limiting (например, slowapi)
Настройте логирование:

import logging
 
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)

Используйте environment-specific конфиги:

class Settings(BaseSettings):
    environment: str = "development"
 
    @property
    def is_production(self) -> bool:
        return self.environment == "production"

Улучшения и расширения

1. Аутентификация (JWT)

pip install python-jose[cryptography] passlib[bcrypt]

Добавьте модель User, endpoints /auth/register, /auth/login.

2. Кеширование (Redis)

pip install redis[hiredis]

Кешируйте /quotes/random для ускорения:

@router.get("/random")
async def random_quote(
    db: AsyncSession = Depends(get_db),
    cache: Redis = Depends(get_redis),
):
    cached = await cache.get("random_quote")
 
    if cached:
        return json.loads(cached)
 
    quote = await crud.get_random_quote(db)
    await cache.setex("random_quote", 60, json.dumps(quote))
 
    return quote

3. Полнотекстовый поиск

Добавьте endpoint /quotes/search?q=motivation:

from sqlalchemy import or_
 
async def search_quotes(db: AsyncSession, query: str) -> list[Quote]:
    result = await db.execute(
        select(Quote).where(
            or_(
                Quote.content.ilike(f"%{query}%"),
                Quote.author.ilike(f"%{query}%"),
            )
        )
    )
    return list(result.scalars().all())

4. Фоновые задачи (Celery)

Для отправки email с цитатой дня:

pip install celery[redis]

from celery import Celery
 
celery = Celery("tasks", broker="redis://localhost:6379")
 
@celery.task
def send_daily_quote():
    # Получаем случайную цитату и отправляем email
    pass

5. Метрики и мониторинг

pip install prometheus-fastapi-instrumentator

from prometheus_fastapi_instrumentator import Instrumentator
 
app = FastAPI()
Instrumentator().instrument(app).expose(app)

Метрики доступны на /metrics.

Итоги

Вы создали production-ready REST API с:

FastAPI — современный async framework
SQLAlchemy 2.0 — typed ORM с async/await
Pydantic v2 — валидация данных
Alembic — миграции БД
pytest — покрытие тестами
OpenAPI — автодокументация
CORS — готовность к фронтенду
Docker — контейнеризация

Следующие шаги:

Добавьте JWT аутентификацию
Внедрите Redis кеширование
Настройте CI/CD (GitHub Actions, GitLab CI)
Добавьте мониторинг (Prometheus + Grafana)
Деплой на production (Railway, Fly.io, DigitalOcean)

Полезные ссылки:

Продолжайте изучать FastAPI — это мощный инструмент для создания быстрых и надёжных API!

← Все статьи Обсудить статью

Создаём API мотивационных цитат на FastAPI

Оглавление

Похожие материалы

pandas для бэкенд-разработчика: логи, Excel/1С, ETL и отчёты — почему я не использовал его раньше

Amazon SQS: от азов до production-ready архитектуры

Kafka + FastAPI: введение в event-driven архитектуру