FastAPI: от разработки до продакшена

За последний год FastAPI стал моим основным фреймворком для создания HTTP API. До этого были Flask и Django REST Framework — оба хороши в своих нишах, но FastAPI попал точно в мою потребность: быстрый асинхронный фреймворк с автоматической валидацией и документацией. В этой статье пройдём весь путь от структуры проекта до работающего деплоя за Nginx.

Почему FastAPI

Прежде чем погружаться в детали, коротко о том, чем FastAPI выигрывает в контексте API-сервисов:

Асинхронность из коробки. FastAPI построен поверх Starlette и поддерживает async/await нативно. Для I/O-bound задач (запросы к базе, вызовы внешних API) это даёт ощутимый прирост производительности.

Автоматическая валидация. Pydantic-модели описывают входные и выходные данные. Если клиент прислал невалидный JSON — фреймворк вернёт 422 с понятным описанием ошибки. Не нужно писать валидацию вручную.

OpenAPI-документация. Swagger UI и ReDoc генерируются автоматически по аннотациям типов. Фронтенд-разработчики и мобильная команда получают актуальную документацию без дополнительных усилий.

Dependency Injection. Встроенная система DI позволяет элегантно управлять зависимостями — сессия базы данных, текущий пользователь, настройки приложения.

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

Начнём с организации кода. Для средних и крупных проектов я использую такую структуру:

myapi/
├── app/
│   ├── __init__.py
│   ├── main.py              # Точка входа, создание приложения
│   ├── config.py            # Настройки через pydantic-settings
│   ├── database.py          # Подключение к БД, сессии
│   ├── dependencies.py      # Общие зависимости (DI)
│   ├── models/              # SQLAlchemy-модели
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/             # Pydantic-схемы
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── routers/             # Эндпоинты, сгруппированные по домену
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   ├── services/            # Бизнес-логика
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── item_service.py
│   └── middleware/          # Кастомные middleware
│       ├── __init__.py
│       └── logging.py
├── migrations/              # Alembic-миграции
│   ├── env.py
│   └── versions/
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_items.py
├── Dockerfile
├── docker-compose.yml
├── alembic.ini
├── pyproject.toml
└── .env.example

Ключевой принцип — разделение ответственности. Роутеры принимают запросы и возвращают ответы, сервисы содержат бизнес-логику, модели описывают структуру данных в базе, схемы — формат API.

Конфигурация через pydantic-settings

Первое, что делаю в любом проекте — настраиваю конфигурацию. Никаких os.getenv() разбросанных по коду:

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

    # Приложение
    app_name: str = "MyAPI"
    debug: bool = False
    api_prefix: str = "/api/v1"

    # База данных
    database_url: str = "postgresql+asyncpg://user:pass@localhost:5432/mydb"
    db_pool_size: int = 20
    db_max_overflow: int = 10

    # Безопасность
    secret_key: str = "change-me-in-production"
    access_token_expire_minutes: int = 30

    # CORS
    allowed_origins: list[str] = ["http://localhost:3000"]


settings = Settings()

Pydantic-settings автоматически подтягивает значения из переменных окружения. В продакшене переменные задаются через Docker или systemd — файл .env используется только локально.

Async-эндпоинты и роутеры

FastAPI поддерживает как sync, так и async-эндпоинты. Для I/O-операций всегда использую async:

# app/routers/items.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_session
from app.schemas.item import ItemCreate, ItemResponse, ItemList
from app.services.item_service import ItemService

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


@router.get("/", response_model=ItemList)
async def list_items(
    skip: int = 0,
    limit: int = 20,
    session: AsyncSession = Depends(get_session),
):
    """Получить список элементов с пагинацией."""
    service = ItemService(session)
    items = await service.get_items(skip=skip, limit=limit)
    total = await service.count_items()
    return ItemList(items=items, total=total)


@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(
    item_data: ItemCreate,
    session: AsyncSession = Depends(get_session),
):
    """Создать новый элемент."""
    service = ItemService(session)
    item = await service.create_item(item_data)
    return item


@router.get("/{item_id}", response_model=ItemResponse)
async def get_item(
    item_id: int,
    session: AsyncSession = Depends(get_session),
):
    """Получить элемент по ID."""
    service = ItemService(session)
    item = await service.get_item(item_id)
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item with id {item_id} not found",
        )
    return item

Обратите внимание на Depends(get_session) — это dependency injection. Сессия базы создаётся для каждого запроса и закрывается после его обработки.

Pydantic-модели для валидации

Схемы описывают, что приходит в API и что уходит из него:

# app/schemas/item.py
from datetime import datetime
from pydantic import BaseModel, Field


class ItemCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str | None = Field(None, max_length=2000)
    price: float = Field(..., gt=0)
    is_active: bool = True


class ItemResponse(BaseModel):
    id: int
    title: str
    description: str | None
    price: float
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}


class ItemList(BaseModel):
    items: list[ItemResponse]
    total: int

Field(...) с валидаторами — мощный инструмент. min_length, max_length, gt, ge, lt, le, pattern — всё это работает автоматически и отражается в OpenAPI-спецификации.

Middleware: CORS и логирование

Точка сборки приложения — main.py:

# app/main.py
import time
import logging
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.database import engine
from app.routers import users, items

logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifecycle: startup и shutdown."""
    logger.info("Starting up...")
    yield
    logger.info("Shutting down...")
    await engine.dispose()


app = FastAPI(
    title=settings.app_name,
    lifespan=lifespan,
    docs_url="/api/docs" if settings.debug else None,
    redoc_url="/api/redoc" if settings.debug else None,
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# Middleware для логирования запросов
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    logger.info(
        "%s %s -> %d (%.3fs)",
        request.method,
        request.url.path,
        response.status_code,
        duration,
    )
    return response


# Health check
@app.get("/health")
async def health_check():
    return {"status": "ok"}


# Роутеры
app.include_router(users.router, prefix=settings.api_prefix)
app.include_router(items.router, prefix=settings.api_prefix)

Важный момент: в продакшене я отключаю Swagger UI (docs_url=None). Документация API не должна быть доступна публично.

Миграции с Alembic

Для управления схемой базы данных использую Alembic в async-режиме:

# Инициализация
alembic init -t async migrations

# Создание миграции
alembic revision --autogenerate -m "add items table"

# Применение
alembic upgrade head

# Откат на одну версию назад
alembic downgrade -1

В migrations/env.py нужно подключить модели и async-движок:

from app.database import Base, DATABASE_URL
from app.models import user, item  # noqa: F401 — импорт для autogenerate

config.set_main_option("sqlalchemy.url", DATABASE_URL)
target_metadata = Base.metadata

Всегда проверяйте автосгенерированные миграции перед применением. Alembic не всегда корректно определяет переименование колонок — иногда он генерирует drop + create вместо rename.

Gunicorn + Uvicorn workers

Для продакшена одного Uvicorn недостаточно. Используем Gunicorn как process manager с Uvicorn workers:

gunicorn app.main:app 
    --worker-class uvicorn.workers.UvicornWorker 
    --workers 4 
    --bind 0.0.0.0:8000 
    --access-logfile - 
    --error-logfile - 
    --timeout 120

Формула для количества workers: 2 * CPU_CORES + 1. Для сервера с 2 ядрами это 5 workers. Но если приложение потребляет много памяти, лучше начать с меньшего числа и мониторить потребление.

Конфиг Gunicorn удобно вынести в файл:

# gunicorn.conf.py
import multiprocessing

bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
timeout = 120
keepalive = 5
accesslog = "-"
errorlog = "-"
loglevel = "info"

Docker: multi-stage build

Dockerfile с multi-stage build для минимального размера образа:

# --- Stage 1: Builder ---
FROM python:3.12-slim AS builder

WORKDIR /build

# Зависимости отдельным слоем для кэширования
COPY pyproject.toml .
RUN pip install --no-cache-dir --prefix=/install .

# --- Stage 2: Runtime ---
FROM python:3.12-slim AS runtime

# Не запускаем от root
RUN groupadd -r appuser && useradd -r -g appuser appuser

WORKDIR /app

# Копируем только установленные пакеты
COPY --from=builder /install /usr/local

# Копируем код приложения
COPY app/ ./app/
COPY migrations/ ./migrations/
COPY alembic.ini .
COPY gunicorn.conf.py .

# Переключаемся на непривилегированного пользователя
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --retries=3 
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["gunicorn", "app.main:app", "-c", "gunicorn.conf.py"]

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

• Multi-stage — в финальный образ не попадают build-зависимости и кэш pip.
• Непривилегированный пользователь — контейнер работает не от root.
• HEALTHCHECK — Docker (и оркестраторы) могут проверять состояние приложения.
• Слои кэшируются — pyproject.toml копируется отдельно, чтобы зависимости пересобирались только при их изменении.

Docker Compose для локальной разработки:

services:
  api:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      db:
        condition: service_healthy
    volumes:
      - ./app:/app/app  # Hot reload в разработке

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

volumes:
  pgdata:

Nginx как reverse proxy

В продакшене FastAPI-приложение всегда стоит за Nginx:

upstream fastapi_backend {
    server 127.0.0.1:8000;
    keepalive 32;
}

server {
    listen 80;
    server_name api.example.com;
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name api.example.com;

    ssl_certificate /etc/letsencrypt/live/api.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;

    # Ограничение размера тела запроса
    client_max_body_size 10m;

    # Таймауты
    proxy_connect_timeout 10s;
    proxy_read_timeout 60s;
    proxy_send_timeout 60s;

    location / {
        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # Для WebSocket (если нужен)
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }

    # Статика — отдаём напрямую через Nginx
    location /static/ {
        alias /app/static/;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }

    # Не отдаём скрытые файлы
    location ~ /. {
        deny all;
    }
}

Почему Nginx, а не напрямую в интернет:

• TLS termination — Nginx эффективнее обрабатывает SSL.
• Статика — отдаётся без нагрузки на Python.
• Rate limiting — настраивается на уровне Nginx.
• Буферизация — Nginx буферизует медленных клиентов, освобождая worker FastAPI.

Тестирование с pytest и httpx

Для тестирования API использую pytest с httpx (async HTTP-клиент):

# tests/conftest.py
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.database import get_session, Base

TEST_DB_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(TEST_DB_URL)
TestSession = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


@pytest.fixture(autouse=True)
async def setup_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture
async def client():
    async def override_session():
        async with TestSession() as session:
            yield session

    app.dependency_overrides[get_session] = override_session

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.clear()

Сами тесты:

# tests/test_items.py
import pytest


@pytest.mark.anyio
async def test_create_item(client):
    response = await client.post(
        "/api/v1/items/",
        json={
            "title": "Test Item",
            "description": "A test item",
            "price": 29.99,
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Test Item"
    assert data["price"] == 29.99
    assert "id" in data


@pytest.mark.anyio
async def test_create_item_invalid_price(client):
    response = await client.post(
        "/api/v1/items/",
        json={
            "title": "Bad Item",
            "price": -10,
        },
    )
    assert response.status_code == 422


@pytest.mark.anyio
async def test_get_nonexistent_item(client):
    response = await client.get("/api/v1/items/99999")
    assert response.status_code == 404


@pytest.mark.anyio
async def test_list_items_pagination(client):
    # Создаём несколько элементов
    for i in range(5):
        await client.post(
            "/api/v1/items/",
            json={"title": f"Item {i}", "price": 10.0 + i},
        )

    response = await client.get("/api/v1/items/?skip=0&limit=3")
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 3
    assert data["total"] == 5

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

pytest tests/ -v --tb=short

Чек-лист перед деплоем

Перед выкаткой в продакшен прохожу по этому списку:

1. Переменные окружения — все секреты заданы, .env не попал в образ.
2. DEBUG выключен — debug=False, Swagger UI недоступен снаружи.
3. CORS настроен — только реальные домены, не *.
4. Health check работает — /health отвечает 200.
5. Миграции применены — alembic upgrade head выполняется перед стартом приложения.
6. Логирование настроено — structured logging, уровень INFO для продакшена.
7. Rate limiting — настроен на Nginx или через middleware.
8. Бэкап базы — перед каждым деплоем с миграциями.

Итого

FastAPI отлично подходит для API-сервисов любого масштаба. Связка FastAPI + Pydantic + SQLAlchemy (async) + Alembic покрывает 90% потребностей бэкенда. Gunicorn управляет воркерами, Docker обеспечивает воспроизводимость, Nginx закрывает вопросы TLS и статики.

Главное — не усложнять на старте. Начните с простой структуры, добавляйте слои по мере роста проекта. Один сервис с десятком эндпоинтов не нуждается в DDD и event sourcing. А вот тесты, миграции и Docker нужны с первого дня.