Введение
Здравствуйте! Идея написать эту статью пришла мне в голову абсолютно спонтанно. Я работаю в компании и, так сложилось, что нас имеет мы имеем DRF монолит на писят два миллиона строк кода. И вот однажды, чью-то светлую голову посетила мысль — «а давайте писать код одинаково». Идея прозвучала чертвоски просто и соблазнительно. С этого момента мы завели себе ишака по имени «Django Service Layer», и все дружно начали на него наваливать. Теперь навалю и вам. Би-бу-бип.
ДИСКЛЕЙМЕР: Статья не претендует на звание серебряной пули. Не является инвестиционной рекомендацией. Учитывайте ваши риски и архитектурные возможности.
Сервисы — место, где живет бизнес-логика
Мы старались, насколько это возможно, не изобретать велосипед. В основу легли проверенные паттерны и подходы: шаблонный метод, DTO, сервисный класс с декоратором @dataclass и еще по мелочи. Если говорить о сути, то вся концепция строится на простоте и доступности. Ее легко объяснить, ее легко понять, предлагаю нам всем в этом убедиться.
Требования к сервису
Чтобы разработчикам было легче проводить ревью, мы также разработали ряд требований, которым должен соответствовать каждый сервис.
Единая ответственность. Каждый сервис реализует законченный кусок бизнес-логики — от простого создания объекта до комплексных операций с вызовами других сервисов или внешних API.
Единственный публичный метод. Вся работа с сервисом ведётся через метод
__call__. Избавляемся от проблемы когда разработчики начинают придумыватьrun-ы, execute-ы, process-ыи т.д.Только именованные аргументы. Все методы сервиса принимают либо
kwargs, либо не имеют аргументов вовсе.Строгая типизация. Во всех методах обязательно указываются типы возвращаемых значений. Это не просто «для mypy» — это документация.
Именование по функции. Название сервиса должно сразу говорить, что он делает:
SendSmsService,CreateAppointmentService,BuyProductService.Свой огород исключений. Сервис определяет специфичные исключения (например,
NotEnoughBalanceError), которые документируют бизнес-правила и делают обработку ошибок осмысленной.Чёткие контракты через DTO. Результатом работы сервиса является Data Transfer Object или примитивный тип данных (
int, float, bool, None).Реализация через dataclass. Это выбор по умолчанию: минимум boilerplate, автоматическая реализация
__init__,__repr__, поддержка frozen-режима для иммутабельности + slots=True для экономии памяти.
Реализация
@dataclass(kw_only=True, slots=True, frozen=True)
class BuyProductService:
price: int
amount: int
@log_service_error
def __call__(self) -> SoldProductDTO:
if self.amount < self.price:
raise NotEnoughBalanceError(price=price, amount=amount)
# ...other business logic...
return SoldProductDTO(message="Thank you for byuing!")Чертовски просто, но все-таки позвольте пару уточнений:
SoldProductDTO— датакласс, в данном случае я сделал заглушку сmessage. Вместо него может быть все, что валидно в рамках вашей бизнес логики. Он может аккумулировать количество оставшихся товаров/остаточный баланс пользователя и все в таком духе.Исключение
NotEnoughBalanceErrorкоторое документирует бизнес-правило и сразу завершает работу сервиса при его нарушении.С помощью декоратора
@log_service_errorмы разделяем зону ответственности и переносим логирование ошибок в отдельную небольшую функцию. Цель — не засорять основной код логированием и не раздувать тем самым кодовую базу.
Хотел бы поподробнее остановиться на работе с исключениями внутри класса-сервиса — далее об этом.
Обработка ошибок внутри сервиса
Исключение внутри сервиса должно вызываться при нарушении какого-либо бизнес-правила. Цели, которые мы преследуем при рейзе ошибки:
Сообщить вызывающей стороне о том, что бизнес-правило не прошло.
Залогировать нарушение бизнес-правила.
Унифицировать процесс логирования и сделать одинаковую структуру лога для всех ошибок бизнес-правил (да поможем эластику более эффективно индексировать логи).
Позволить вызывающей стороне решить, как правильно обработать данную ошибку. Другими словами — сервис не в праве решать, вернуть 400-ый или 200-ый код ответа, например, при вызове во «вьюхе».
Чтобы добиться поставленных целей, внедряем два простых компонента:
Общий класс для сервисных исключений
Это первый компонент. От данного класса должны наследоваться все другие сервисные исключения.
class BaseServiceError(Exception):
"""Base service error."""
def __init__(self, *, message: str = None, **context) -> None:
self.message = self.__doc__ or message
self.context = context
class NotEnoughBalanceError(BaseServiceError):
"""There are insufficient funds in the balance to complete the transaction."""Немого пояснений:
self.message = self.__doc__ or message— заставляем разработчиков описывать док-стринги для всех исключений и избавляемся от необходимости каждый раз прописывать одинаковое сообщение об ошибке. Код самодокументируется — и это прекрасно.**context— немного забегая вперед скажу, что так мы добавляем гибкость при логировании ошибок. Например, если у пользователя не хватило денег на балансе, чтобы купить товар — мы залогируем ошибку и дополнительно добавим контекст, что у него было «столько-то» рублей, а товар стоил «столько-то» денег.
Специальный сервисный декоратор
Это второй компонент. Он умеет работать только с нашими сервисами и только с ошибками, которые унаследованы от BaseServiceError.
def log_service_error(__call__: Callable) -> Callable:
"""Use this decorator on __call__ service method."""
@wraps(__call__)
def wrapper(self, **kwargs) -> Any:
try:
return __call__(self, **kwargs)
except BaseServiceError as exc:
logger.error(
{
"error_in": self.__class__.__name__,
"error_name": exc.__class__.__name__,
"error_message": exc.message,
**exc.context,
}
)
raise exc
return wrapperТакой декоратор залогирует всю необходимую информацию об ошибке:
Он залогирует ГДЕ произошла ошибка (
error_in)Он залогирует КАКАЯ произошла ошибка (
error_name)Он залогирует ПОЧЕМУ произошла ошибка (
error_message)Он залогирует ДОПОЛНИТЕЛЬНЫЙ КОНТЕКСТ, который передал разработчик при рейзе исключения (
**exc.context)
Преимущества такого подхода:
Удобно фильтровать и искать нужные логи в кибане.
Можно строить дашборды и смотреть, какие сервисы часто «падают» по их названиям, при желании — провести более глубокую аналитику.
Плюс к чистоте кода — не засоряем сервисный слой постоянным логированием ошибок.
Он универсален для любого сервиса, но не обязателен. Мы без проблем можем не вешать декоратор, если нечего логировать или мы не хотим этого делать по какой-либо причине.
В конце концов мы экономим время разработчика, ему не нужно думать о том в каком формате залогировать ошибки, и как это лучше сделать. Достаточно дописать одну строчку над методом
__call__.Мы заранее готовы к некоторым расширениям. Возможно, в будущем вы захотите замерять время работы вашей бизнес логики? Создайте новый декоратор и вешайте его поверх логгера. Самое главное — не вмешивайтесь в логику работы сервиса.
Ну, и последнее, что хочется сказать про обработку — это использование встроенного в DRF хендлера — централизированный на уровне API способ обрабатывать сервисные ошибки и отдавать валидные ответы в одинаковом формате. Ссылка на доку и пример ниже:
def service_exception_handler(exc, context):
if isinstance(exc, BaseServiceError):
return Response(
data={
"error_message": exc.message,
"error_context": dict(**exc.context),
},
status=status.HTTP_400_BAD_REQUEST,
)
return exception_handler(exc, context)Что-то подобное есть у каждого фреймворка, так что, думаю, при желании можно найти альтернативу.
С учетом всего вышесказанного, нам, в большинстве случаев, не нужно писать никакие try-except-ы внутри нашихviews. Также нам не нужно задумываться над логированием ошибок. В то же время такой код очень легко тестируется. При необходимости любой из сервисом можно «замокать». При помощи декоратора @dataclass(kw_only=True, slots=True, frozen=True) мы, насколько это возможно, делаем код потокобезопасным, более понятным и, в конце концов, такой класс весит меньше памяти. Можете в этом сами убедится при помощи sys.getsizeof. Далее — немного практики, чтобы еще лучше понять идею.
Как не нужно делать — примеры
1. Нельзя чтобы ваш сервис возвращал:
Более чем одно значение.
Словарь/список/кортеж и тем подобные структуры.
ПРИМЕР:
@dataclass(kw_only=True, slots=True, frozen=True)
class GetCoordinateService:
@log_service_error
def __call__(self) -> tuple[int, int]:
# ...business logic...
return x, y # ❌ Возвращает кортеж, неявный контрактВместо этого следует использовать DTO для возврата структурированных данных:
@dataclass(kw_only=True, slots=True, frozen=True)
class GetCoordinateService:
@log_service_error
def __call__(self) -> PointDTO:
# ...business logic...
return PointDTO(x=x, y=y) # ✅ возвращает DTO, явный контракт2. Нельзя чтобы ваш сервис принимал в методе __call__:
Более чем два параметра
Словарь/неограниченное количество аргументов
ПРИМЕР:
@dataclass(kw_only=True, slots=True, frozen=True)
class CalculatePaymentService:
@log_service_error
def __call__(self, **kwargs) -> int:
deposit = kwargs["deposit"]
percent = kwargs["percent"]
term = kwargs["term"]
# ❌ Очень неявное поведениеВместо этого следует использовать DTO:
@dataclass(kw_only=True, slots=True, frozen=True)
class CalculatePaymentService:
@log_service_error
def __call__(self, *, calculate_data: CalculateDTO) -> int:
# ✅ понятный контрактКак нужно делать — примеры
1) Использование абстрактных классов и шаблонного метода:
class AbstractExportIssueService(abc.ABC):
system: ExportToEnum
@log_service_error
def __call__(self, *, issue: Issue) -> None:
"""Export issue to external system."""
built_export_model = self._build_export_model(issue)
self._export(to_export=built_export_model, issue=issue)
logger.info(
{
"message": "Issue successfully exported.",
"export_system": self.system,
"export_id": issue.pk,
"export_type": issue.type,
"export_data": built_export_model.model_dump(exclude_unset=True),
}
)
@final
def _build_export_model(self, issue: Issue) -> ExportDTO:
builder_class = self._get_builder_class(issue)
return builder_class()(issue=issue)
@abc.abstractmethod
def _get_builder_class(self, issue: Issue) -> Type[IExportBuildService]:
"""Resolving builder class."""
@abc.abstractmethod
def _export(self, to_export: ExportDTO, issue: Issue) -> None:
"""Business logic implementation."""
@final
@dataclass(kw_only=True, slots=True, frozen=True)
class JiraExportService(AbstractExportIssueService):
system = ExportToEnum.JIRA
def _export(self, to_export: JiraExportDTO, issue: Issue) -> None:
create_order(to_export.model_dump(by_alias=True))
issue.is_export_to_jira = True
issue.save(update_fields=["is_export_to_jira"])
def _get_builder_class(self, issue: Issue) -> Type[IExportBuildService]:
if issue.export_to_jira_kb:
return JiraBuildExportKBDataService
return JiraBuildExportDataService
@final
@dataclass(kw_only=True, slots=True, frozen=True)
class AnotherOneExportService(AbstractExportIssueService):
system = ExportToEnum.ONE_MORE_SYSTEM
def _export(self, to_export: AnotherSystemDTO, issue: Issue) -> None:
# ...other business logic impl...
def _get_builder_class(self, issue: Issue) -> Type[IExportBuildService]:
# ...resolving builder class...
2) Использование декоратора @final при наследовании от абстрактного сервиса.
3) Разделение логики на небольшие приватные методы, вместо «раздувания» одного большого метода __call__.
4) Декомпозиция бизнес-логики на уровне сервисов, определение зоны ответственности конкретного сервиса. Не стоит смешивать в одном сервисе следующую логику:
обращение во внешний сервис
сбор данных перед запросом
любая сложная калькуляция
Итого
Если после прочтения вашей статьи у вас остался привкус незавершенности — так и должно быть. Мы ни слова не сказали про DI (Dependency Injection). Почти ничего не сказали про то, как лучше строить общение между сервисами. Однако, это уже не так важно. Я надеюсь, что у меня получилось донести идею нашего подхода. Если вам в целом симпатизирует то, о чем вы прочитали, но в голове родились предложения по улучшению — мы можем обсудим это в моем гитхаб репозитории. Заводите ишуес. Би-бу-бип.
