Финальный проект: REST API на FastAPI + SQLAlchemy + JWT + Docker

Финальный проект: REST API на FastAPI + SQLAlchemy + JWT + Docker

Это большой проект, в котором собирается всё из первых десяти модулей. После него остаётся Модуль 11 про очереди и consumers - он про другой тип задач и на этот проект не опирается. Соберём всё пройденное в рабочий REST API: TODO-приложение с JWT-аутентификацией, PostgreSQL через SQLAlchemy, миграциями Alembic, тестами на pytest и Docker-контейнеризацией. Кода будет много: разбираем проект целиком, а не по фрагментам.

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

pyproject.toml

[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"

[project]
name = "todo-api"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi[all]>=0.110",
    "sqlalchemy[asyncio]>=2.0",
    "asyncpg>=0.29",
    "alembic>=1.13",
    "pydantic-settings>=2.2",
    "python-jose[cryptography]>=3.3",
    "passlib[bcrypt]>=1.7",
    "python-multipart>=0.0.9",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "pytest-cov>=5.0",
    "httpx>=0.27",
    "ruff>=0.3",
    "mypy>=1.10",
]

[tool.setuptools.packages.find]
where = ["src"]

[tool.ruff]
line-length = 100
[tool.ruff.lint]
select = ["E", "F", "W", "I", "B", "UP"]

[tool.mypy]
strict = true

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

config.py

from pydantic_settings import BaseSettings
from pydantic import SecretStr

class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://test:test@localhost/todo"
    # У секрета дефолта быть не может.
    #
    # Дефолт означает, что приложение молча запустится без TODO_JWT_SECRET
    # и будет подписывать токены значением, опубликованным в уроке. Любой,
    # кто его знает, выпишет себе токен с чужим "sub" и получит полный
    # доступ. Без значения в окружении Pydantic не создаст Settings
    # и приложение упадёт на старте - это правильное поведение.
    jwt_secret: SecretStr
    jwt_algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    class Config:
        env_file = ".env"
        env_prefix = "TODO_"

settings = Settings()

db.py

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
from typing import AsyncGenerator
from todo_api.config import settings

class Base(DeclarativeBase):
    pass

engine = create_async_engine(settings.database_url, echo=False)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionFactory() as session:
        yield session

models.py

from datetime import datetime, timezone
from sqlalchemy import String, ForeignKey, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from todo_api.db import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

    todos: Mapped[list["Todo"]] = relationship(back_populates="user", cascade="all, delete-orphan")

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    title: Mapped[str] = mapped_column(String(200))
    done: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

    user: Mapped["User"] = relationship(back_populates="todos")

schemas.py

from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8)

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: EmailStr
    created_at: datetime

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    done: bool = False

class TodoUpdate(BaseModel):
    title: str | None = Field(None, min_length=1, max_length=200)
    done: bool | None = None

class TodoOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    done: bool
    created_at: datetime

auth.py

from datetime import datetime, timedelta, timezone
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from passlib.context import CryptContext
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from todo_api.config import settings
from todo_api.db import get_db
from todo_api.models import User

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
    to_encode.update({"exp": expire})
    return jwt.encode(
        to_encode,
        settings.jwt_secret.get_secret_value(),
        algorithm=settings.jwt_algorithm,
    )

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(
            token,
            settings.jwt_secret.get_secret_value(),
            algorithms=[settings.jwt_algorithm],
        )
        email = payload.get("sub")
        if not email:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    result = await db.scalars(select(User).where(User.email == email))
    user = result.first()
    if not user:
        raise credentials_exception
    return user

routers/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from todo_api.auth import create_access_token, hash_password, verify_password
from todo_api.db import get_db
from todo_api.models import User
from todo_api.schemas import Token, UserCreate, UserOut

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def register(data: UserCreate, db: AsyncSession = Depends(get_db)) -> User:
    # Предварительная проверка - для внятного ответа в обычном случае.
    # Полагаться на неё нельзя: между SELECT и INSERT проходит время,
    # и второй запрос с тем же email успевает вклиниться.
    existing = await db.scalars(select(User).where(User.email == data.email))
    if existing.first():
        raise HTTPException(status_code=409, detail="Email already registered")

    user = User(email=data.email, hashed_password=hash_password(data.password))
    db.add(user)
    try:
        await db.commit()
    except IntegrityError:
        # Источник правды - UNIQUE в схеме, а не проверка выше.
        # Rollback обязателен: после ошибки сессия непригодна,
        # и следующий запрос по ней упадёт с InvalidRequestError.
        await db.rollback()
        raise HTTPException(status_code=409, detail="Email already registered")

    await db.refresh(user)
    return user

@router.post("/login", response_model=Token)
async def login(
    form: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db),
) -> Token:
    result = await db.scalars(select(User).where(User.email == form.username))
    user = result.first()
    if not user or not verify_password(form.password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token({"sub": user.email})
    return Token(access_token=token)

routers/todos.py

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from todo_api.auth import get_current_user
from todo_api.db import get_db
from todo_api.models import Todo, User
from todo_api.schemas import TodoCreate, TodoOut, TodoUpdate

router = APIRouter(prefix="/todos", tags=["todos"])

@router.get("/", response_model=list[TodoOut])
async def list_todos(
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Границы обязательны. Без ge=0 отрицательный skip уходит в OFFSET -1
    # и база отвечает ошибкой; без le клиент запросит limit=1000000
    # и выгребет всю таблицу одним запросом.
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=200),
) -> list[Todo]:
    stmt = select(Todo).where(Todo.user_id == user.id).offset(skip).limit(limit)
    result = await db.scalars(stmt)
    return list(result.all())

@router.post("/", response_model=TodoOut, status_code=status.HTTP_201_CREATED)
async def create_todo(
    data: TodoCreate,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Todo:
    todo = Todo(user_id=user.id, **data.model_dump())
    db.add(todo)
    await db.commit()
    await db.refresh(todo)
    return todo

@router.get("/{todo_id}", response_model=TodoOut)
async def get_todo(
    todo_id: int,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Todo:
    todo = await db.get(Todo, todo_id)
    if not todo or todo.user_id != user.id:
        raise HTTPException(404, "Todo not found")
    return todo

@router.patch("/{todo_id}", response_model=TodoOut)
async def update_todo(
    todo_id: int,
    data: TodoUpdate,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Todo:
    todo = await db.get(Todo, todo_id)
    if not todo or todo.user_id != user.id:
        raise HTTPException(404, "Todo not found")

    update_data = data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(todo, field, value)

    await db.commit()
    await db.refresh(todo)
    return todo

@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(
    todo_id: int,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> None:
    todo = await db.get(Todo, todo_id)
    if not todo or todo.user_id != user.id:
        raise HTTPException(404, "Todo not found")
    await db.delete(todo)
    await db.commit()

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from todo_api.routers import auth, todos

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    yield
    # Shutdown
    pass

app = FastAPI(
    title="Todo API",
    version="0.1.0",
    description="REST API for personal todos with JWT auth",
    lifespan=lifespan,
)

app.include_router(auth.router)
app.include_router(todos.router)

@app.get("/health")
def health():
    return {"status": "ok"}

Тесты

conftest.py

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from todo_api.db import Base, get_db
from todo_api.main import app

TEST_DB = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture
async def db():
    engine = create_async_engine(TEST_DB)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    SessionFactory = async_sessionmaker(engine, expire_on_commit=False)
    async with SessionFactory() as session:
        yield session
    await engine.dispose()

@pytest_asyncio.fixture
async def client(db):
    async def get_test_db():
        yield db
    app.dependency_overrides[get_db] = get_test_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

test_auth.py

import pytest

@pytest.mark.asyncio
async def test_register(client):
    response = await client.post(
        "/auth/register",
        json={"email": "alice@example.com", "password": "password123"},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "alice@example.com"
    assert "id" in data

@pytest.mark.asyncio
async def test_register_duplicate(client):
    await client.post("/auth/register", json={"email": "a@b.c", "password": "12345678"})
    response = await client.post("/auth/register", json={"email": "a@b.c", "password": "12345678"})
    assert response.status_code == 409

@pytest.mark.asyncio
async def test_login(client):
    await client.post("/auth/register", json={"email": "a@b.c", "password": "12345678"})
    response = await client.post(
        "/auth/login",
        data={"username": "a@b.c", "password": "12345678"},
    )
    assert response.status_code == 200
    assert "access_token" in response.json()

@pytest.mark.asyncio
async def test_login_wrong_password(client):
    await client.post("/auth/register", json={"email": "a@b.c", "password": "12345678"})
    response = await client.post(
        "/auth/login",
        data={"username": "a@b.c", "password": "wrong"},
    )
    assert response.status_code == 401

test_todos.py

import pytest

async def get_token(client, email="user@example.com", password="password123"):
    await client.post("/auth/register", json={"email": email, "password": password})
    response = await client.post("/auth/login", data={"username": email, "password": password})
    return response.json()["access_token"]

@pytest.mark.asyncio
async def test_create_todo(client):
    token = await get_token(client)
    response = await client.post(
        "/todos/",
        json={"title": "Buy milk"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 201
    assert response.json()["title"] == "Buy milk"

@pytest.mark.asyncio
async def test_list_todos(client):
    token = await get_token(client)
    headers = {"Authorization": f"Bearer {token}"}
    await client.post("/todos/", json={"title": "Task 1"}, headers=headers)
    await client.post("/todos/", json={"title": "Task 2"}, headers=headers)
    response = await client.get("/todos/", headers=headers)
    assert response.status_code == 200
    assert len(response.json()) == 2

@pytest.mark.asyncio
async def test_todos_isolation(client):
    """Юзеры видят только свои todos"""
    token_a = await get_token(client, "alice@a.b", "password123")
    token_b = await get_token(client, "bob@a.b", "password123")

    await client.post("/todos/", json={"title": "Alice task"}, headers={"Authorization": f"Bearer {token_a}"})

    response = await client.get("/todos/", headers={"Authorization": f"Bearer {token_b}"})
    assert response.json() == []   # Bob не видит todos Alice

@pytest.mark.asyncio
async def test_unauthorized(client):
    response = await client.get("/todos/")
    assert response.status_code == 401

Dockerfile

FROM python:3.12-slim AS builder

WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

COPY pyproject.toml ./
RUN pip install --user --no-warn-script-location -e .

FROM python:3.12-slim

WORKDIR /app
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH

COPY src/ ./src/
COPY alembic.ini .
COPY alembic/ ./alembic/

RUN useradd -m -u 1000 app && chown -R app:app /app
USER app

EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=10s CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "todo_api.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yml

version: "3.9"

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      TODO_DATABASE_URL: postgresql+asyncpg://todo:secret@db:5432/todo
      TODO_JWT_SECRET: dev-secret-key
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: todo
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: todo
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U todo"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  postgres_data:

Запуск

# Установка
pip install -e ".[dev]"

# Локально с SQLite
TODO_DATABASE_URL="sqlite+aiosqlite:///./dev.db" \
TODO_JWT_SECRET="dev-secret" \
uvicorn todo_api.main:app --reload

# Тесты
pytest

# Docker
docker-compose up

# Миграции
alembic revision --autogenerate -m "initial schema"
alembic upgrade head

# Тестовый запрос
curl http://localhost:8000/health

Что осталось для production

Этот проект - solid foundation, но для real production добавь:

1. Rate limiting - slowapi или nginx
2. Структурированные логи - JSON logs с request_id
3. Метрики - Prometheus через prometheus-fastapi-instrumentator
4. Tracing - OpenTelemetry для distributed tracing
5. Refresh tokens - продление session без логина
6. Email verification - подтверждение email при register
7. Password reset flow - забыли пароль
8. CORS configuration - правильно для frontend домена
9. Тесты на race conditions - concurrent requests
10. Load testing - locust или k6
11. Backup strategy для БД
12. Monitoring - alerts на errors, latency

Что уже умеешь

Путь от установки Python до работающего REST API с базой, тестами и Docker пройден. У тебя теперь есть навыки для:

• Писать чистый идиоматичный Python (modules 1-5)
• Применять ООП правильно (module 6)
• Понимать concurrency: asyncio, threading, multiprocessing (module 7)
• Управлять packaging, stdlib, logging (module 8)
• Тестировать с pytest, ruff, mypy (module 9)
• Создавать REST API на FastAPI с auth, БД, миграциями, Docker (module 10)

Куда расти дальше, за пределами трека:

• WebSockets для real-time
• GraphQL как альтернатива REST
• Реальный production проект - применить всё на практике
• Глубокое погружение в performance, profiling, optimization
• DevOps: Kubernetes, observability

Что дальше в треке

Модуль 10 завершён, API собран и работает. Остался Модуль 11: очереди и consumers - RabbitMQ, Kafka, идемпотентность, retry и outbox.

Это отдельный тип задач, а не продолжение этого проекта. Синхронный API умеет отвечать на запрос здесь и сейчас; очереди нужны там, где работу надо отложить, распределить между воркерами или пережить недоступность соседнего сервиса. Проект выше от них не зависит, поэтому Модуль 11 можно пройти сразу, а можно вернуться к нему, когда такая задача появится в работе.