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

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

Это финальный урок трека. Соберём всё что прошли в production-ready REST API: TODO-приложение с JWT-аутентификацией, PostgreSQL через SQLAlchemy, миграциями Alembic, тестами на pytest и Docker-контейнеризацией. Будет много кода - это complete walkthrough реального backend.

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

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"
    jwt_secret: SecretStr = SecretStr("dev-secret-change-in-prod")
    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.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:
    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)
    await db.commit()
    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, 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),
    skip: int = 0,
    limit: int = 100,
) -> 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 до production-ready REST API. У тебя теперь есть навыки для:

• Писать чистый идиоматичный 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)

Следующие шаги для роста:

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

Что дальше

Модуль 10 завершён - финальный проект собран. В Модуле 11 разберём async messaging и consumers (RabbitMQ, Kafka) - последняя часть трека для современного backend.