Привет, Хабр! Когда начинаешь новый проект на FastAPI, всё кажется простым: пара моделей Pydantic, несколько эндпоинтов — и готово. Но через полгода и 20 000 строк кода оказывается, что базовая валидация размазана по всему приложению, бизнес-логика перемешана с обращениями к БД, а тесты пишутся со скрипом.

Сегодня я хочу поделиться конкретными архитектурными паттернами и приёмами, которые я неоднократно использовал в своих проектах. Они не усложняют простые задачи, но делают жизнь в долгосрочной перспективе несоизмеримо легче.

1. Service Layer: Отделяем бизнес-логику от эндпоинтов

Проблема: Бизнес-логика живёт в эндпоинтах. Хотите отменить бронь? Проверка прав, обновление статуса, запись в лог, отправка уведомления — всё это в одном роутере на 50 строк. Протестировать это отдельно? Почти невозможно.

Решение: Слой сервисов — это чистые классы Python с методами, которые знают правила вашего домена, но не знают о HTTP, базе данных или внешних API.

class BookingService:
    def __init__(self, booking_repo: AbstractBookingRepository):
        self._repo = booking_repo

    async def cancel_booking(self, booking_id: UUID, user_id: UUID) -> Booking:
        """Отмена бронирования с проверкой всех бизнес-правил."""
        booking = await self._repo.get_by_id(booking_id)
        
        # Всё, что касается правил отмены - здесь
        if booking.user_id != user_id:
            raise ForbiddenError("Нельзя отменять чужие брони")
        if booking.status != BookingStatus.CONFIRMED:
            raise BusinessError("Можно отменять только подтверждённые брони")
        if not booking.is_cancellable():
            raise BusinessError("Срок отмены истёк")
        
        booking.cancel()
        await self._repo.save(booking)
        return booking

# В эндпоинте остаётся только это:
@router.delete("/{booking_id}")
async def cancel_booking(
    booking_id: UUID,
    user: User = Depends(get_current_user),
    booking_service: BookingService = Depends(get_booking_service)
):
    """Эндпоинт теперь тонкий и только обрабатывает HTTP-контракт."""
    await booking_service.cancel_booking(booking_id, user.id)
    return {"status": "cancelled"}

Что это даёт:

• Тестируемость: Можно написать unit-тесты для BookingService, не поднимая FastAPI приложение.

• Переиспользование: Ту же логику отмены можно вызвать из фоновой задачи или CLI.

• Читаемость: По названию сервиса и метода сразу понятно, что происходит.

2. Dependency Injection не только для БД

Проблема: Depends() используют в основном для подключения к БД. Но его потенциал гораздо шире.

Решение: Используйте зависимости для всего, что может меняться или требует изоляции.

# 1. Получение текущего пользователя (очевидное)
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User: ...

# 2. Проверка прав (часто упускают!)
async def require_admin(user: User = Depends(get_current_user)) -> User:
    if not user.is_admin:
        raise HTTPException(status_code=403)
    return user

@router.get("/stats")
async def get_stats(user: User = Depends(require_admin)): ...

# 3. Валидация входных данных сверх Pydantic
def validate_time_range(
    start: datetime = Query(...),
    end: datetime = Query(...)
) -> tuple[datetime, datetime]:
    if end <= start:
        raise HTTPException(400, "Конец должен быть после начала")
    if (end - start).days > 30:
        raise HTTPException(400, "Максимальный диапазон — 30 дней")
    return start, end

@router.get("/report")
async def get_report(
    time_range: tuple = Depends(validate_time_range)
): ...

# 4. Логирование действий
class ActionLogger:
    def __init__(self, user: User = Depends(get_current_user)):
        self.user = user
    
    async def log(self, action: str, details: dict):
        await log_to_db(user=self.user, action=action, **details)

@router.post("/booking")
async def create_booking(
    logger: ActionLogger = Depends(),
    booking_data: BookingCreate = Body(...)
):
    await logger.log("booking_created", {"data": booking_data.dict()})

Что это даёт:

• Чистые эндпоинты: Вся побочная логика вынесена в зависимости.

• Гибкость: Можно легко подменить реализацию для тестов (AsyncMock для ActionLogger).

• Соблюдение SRP: Каждая зависимость делает одну вещь.

3. Кастомные статус-коды и ошибки, которые помогают фронту

Проблема: Все ошибки возвращаются как 500 или 400 с текстом, который парсится регулярками.

Решение: Создайте иерархию исключений и глобальный обработчик.

from fastapi import HTTPException
from typing import Any, Optional

class BaseAPIException(HTTPException):
    """База для всех наших исключений."""
    def __init__(
        self,
        status_code: int,
        detail: Any = None,
        error_code: Optional[str] = None,  # Машинный код ошибки
        extra: Optional[dict] = None
    ):
        super().__init__(status_code=status_code, detail=detail)
        self.error_code = error_code
        self.extra = extra or {}

# Конкретные исключения
class NotFoundException(BaseAPIException):
    def __init__(self, detail: str = "Ресурс не найден"):
        super().__init__(404, detail, error_code="not_found")

class ValidationException(BaseAPIException):
    def __init__(self, detail: str, errors: Optional[list] = None):
        super().__init__(422, detail, error_code="validation_error")
        self.extra["fields"] = errors or []

class BusinessLogicException(BaseAPIException):
    def __init__(self, detail: str):
        super().__init__(409, detail, error_code="business_error")

# Глобальный обработчик в main.py
@app.exception_handler(BaseAPIException)
async def api_exception_handler(request, exc: BaseAPIException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.error_code,
                "message": exc.detail,
                **exc.extra
            }
        }
    )

# Использование в сервисе
async def get_booking(booking_id: UUID) -> Booking:
    booking = await repository.get(booking_id)
    if not booking:
        raise NotFoundException(f"Бронь {booking_id} не найдена")
    return booking

Что это даёт:

• Согласованный формат ошибок: Фронтенд знает, что всегда искать поле error.code.

• Лёгкая обработка на клиенте: Можно сделать перехватчик:

if (error.code === 'validation_error') {
  showFieldErrors(error.fields);
}

• Безопасность: Не пробрасываются внутренние детали из исключений БД.

4. Фоновые задачи, которые не сломаются при перезагрузке

Проблема: BackgroundTasks хороши для простых операций, но если сервер упадёт до выполнения — задача потеряется навсегда.

Решение: Используем связку Redis + RQ или Celery для важных операций.

from redis import Redis
from rq import Queue
from core.config import settings

redis = Redis.from_url(settings.REDIS_URL)
task_queue = Queue("default", connection=redis)

# Простая функция, которая будет выполнена воркером
def send_booking_confirmation_email(booking_id: str, user_email: str):
    # Здесь может быть сложная логика отправки
    email_service.send(
        to=user_email,
        subject="Бронь подтверждена",
        template="booking_confirm.html",
        booking_id=booking_id
    )

# В эндпоинте
@router.post("/booking")
async def create_booking(
    booking_data: BookingCreate,
    task_queue: Queue = Depends(get_task_queue)
):
    booking = await booking_service.create(booking_data)
    
    # Отправляем задачу в очередь вместо BackgroundTasks
    task_queue.enqueue(
        send_booking_confirmation_email,
        booking_id=str(booking.id),
        user_email=booking.user_email,
        # Можно добавить отложенное выполнение
        job_timeout=30  # seconds
    )
    
    return booking

Что это даёт:

• Надёжность: Задачи переживают перезагрузку сервера.

• Масштабирование: Воркеры можно запускать на отдельных машинах.

• Контроль: Есть панели мониторинга (RQ Dashboard, Flower для Celery).

5. Pydantic не только для запросов: Domain Models

Проблема: Мы используем Pydantic только для Request и Response моделей, а бизнес-сущности описываем классами с кучей свойств и методов.

Решение: Сделайте Pydantic-модели полноценными доменными объектами.

from pydantic import BaseModel, validator, root_validator
from enum import Enum
from datetime import datetime, timedelta

class BookingStatus(str, Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"

class Booking(BaseModel):
    """Доменная модель бронирования."""
    id: UUID
    user_id: UUID
    resource_id: UUID
    status: BookingStatus
    starts_at: datetime
    ends_at: datetime
    
    # Валидаторы на уровне домена
    @validator('ends_at')
    def validate_duration(cls, ends_at, values):
        starts_at = values.get('starts_at')
        if starts_at and ends_at:
            if ends_at <= starts_at:
                raise ValueError('Конец должен быть после начала')
            if ends_at - starts_at > timedelta(hours=24):
                raise ValueError('Максимум 24 часа')
        return ends_at
    
    # Бизнес-методы прямо в модели
    def cancel(self):
        if self.status != BookingStatus.CONFIRMED:
            raise BusinessError("Можно отменять только подтверждённые")
        self.status = BookingStatus.CANCELLED
    
    def is_cancellable(self) -> bool:
        return (
            self.status == BookingStatus.CONFIRMED
            and self.starts_at - datetime.now() > timedelta(hours=1)
        )
    
    class Config:
        orm_mode = True  # Для удобной конвертации из SQLAlchemy
        use_enum_values = True

Что это даёт:

• Единый источник истины: Валидация в одном месте.

• Богатая модель: Можно добавлять методы, не создавая сервис на каждое действие.

• Совместимость: Легко конвертировать в/из БД.

Эти паттерны не требуют переписывания всего приложения. Можно внедрять их постепенно:

1. Начните с кастомных исключений — это 30 минут работы.

2. При рефакторинге следующего модуля выделите сервисный слой.

3. Когда понадобится отправка 1000 писем — поставьте Redis и очередь задач.

Главный принцип: не усложнять раньше времени, но и не бояться структурировать код, когда он начинает расти.

А какие архитектурные решения спасли ваши проекты? Делитесь в комментариях