Привет, Хабр! Когда начинаешь новый проект на 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
Что это даёт:
Единый источник истины: Валидация в одном месте.
Богатая модель: Можно добавлять методы, не создавая сервис на каждое действие.
Совместимость: Легко конвертировать в/из БД.
Эти паттерны не требуют переписывания всего приложения. Можно внедрять их постепенно:
Начните с кастомных исключений — это 30 минут работы.
При рефакторинге следующего модуля выделите сервисный слой.
Когда понадобится отправка 1000 писем — поставьте Redis и очередь задач.
Главный принцип: не усложнять раньше времени, но и не бояться структурировать код, когда он начинает расти.
А какие архитектурные решения спасли ваши проекты? Делитесь в комментариях
