Оглавление

1. Введение

2. Шпаргалка: Код из первой статьи

3. Dependency Injection — Punq

   1. Реализация

   2. Если сервис зависит от другого сервиса

4. Доменные модели и их логика

5. Бонус или «При чем здесь ELK?»

6. Заключение: Собираем пазл

Введение

Здравствуйте! Это вторая часть из серии статей «Сервисы — место, где живет бизнес логика». Если Вы еще не знакомы с первой частью, то рекомендую начать с нее, чтобы у вас сложилась общая картина. Сегодня мы постараемся ответить на все оставшиеся вопросы: познакомимся с прекрасной, легковесной DI-библиотекой, научимся «инжектить» в Django, посмотрим на несколько дашбордов в Кибане и поговорим про доменные модели.

Шпаргалка: Код из первой статьи

Чтобы у Вас под рукой была информация из предыдущей части, продублирую здесь код, на котором мы остановились:

• Сервисы с декоратором @dataclass и одним публичным методом __call__.

@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!")
  

• Декоратор @log_service_error, который по желанию можно вешать на методы __call__:

def log_service_error(__call__: Callable) -> Callable:
    @wraps(__call__)
    def wrapper(self, **kwargs) -> Any:
        try:
            return __call__(self, **kwargs)
        except BaseServiceError as error:
            logger.error(
                {
                    "error_in": self.__class__.__name__,
                    "error_name": error.__class__.__name__,
                    "error_message": error.message,
                    "error_context": dict(**error.context),
                },
            )
            raise error

    return wrapper

• Базовый класс исключений BaseServiceError, от которого должны наследоваться все сервисные исключения:

class BaseServiceError(Exception):
    def __init__(self, message: str = None, **context) -> None:
        self.message = message or self.__doc__
        self.context = context

class ProductNotAvailable(BaseServiceError):
    """Product not available now"""


class OutOfStockError(BaseServiceError):
    """Product is out of stock"""

• Как сервисные ошибки обрабатывались на уровне всего приложения:

from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import exception_handler
from apps.core.service import BaseServiceError


def service_exception_handler(exc, context):
    if isinstance(exc, BaseServiceError):
        return Response(
            data={
                "error": exc.message,
                "detail": dict(**exc.context),
            },
            status=status.HTTP_400_BAD_REQUEST,
        )
    return exception_handler(exc, context)

Dependency Injection — Punq

Мы перебрали большое количество возможных DI фреймворков — почти все из них выглядели очень неудобно. И, думаю, это не проблема конкретных фреймворков. Просто при разработке Django, изначально никто не закладывал возможность использования DI в привычном смысле этого слова. Фреймворк имеет свое видение того, как нужно организовывать код, и идти против него — значит постоянно сталкиваться с сопротивлением. Но все же у нас получилось найти подход, который достаточно гладко укладывается в существующую архитектуру Django.

«Назвался punq-ом — живи ... будь готов, что тебя никогда не выдаст Google по запросу «Depencdency Injection in Python». Но мы нашли.

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

Реализация

Разбираться будем на том же самом примере бизнес логики — есть условный товар, его можно купить. Для этого у нас есть BuyProductView, где вызывается BuyProductService— там и находится наша бизнес-логика.

Итак, предлагаю, что называется, пойти «от противного» и немного усложнить ситуацию. Представим, что:

1. У нас нет никакого DI фреймворка.

2. На вход наш сервис ожидает DTO-объект. То есть, перед тем как вызвать сервис, этот DTO нужно где-то создать.

Первое, что приходит в голову, это написать код так:

class BuyProductView(APIView):
    permission_classes = (CustomerRequired,)

    def post(self, request: Request) -> Response:
        serializer = BuyProductSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        result = BuyProductService(
            product=BuyProductIn(**serializer.data)
        )(customer=request.user.customer)

        return Response(
            data=result,
            status=status.HTTP_200_OK,
        )

Посмотрите внимательно на этот код, какие у него есть проблемы?

Первое — жесткая связь между классами. Наша BuyProductView знает слишком много о внутреннем устройстве BuyProductService: если вдруг, вместо BuyProductIn, у сервиса изменится конструктор, и нужно будет передавать, например, BuyProductInV2, то нам неизбежно придется вносить изменения в код этой «вьюшки». Это не очень хорошая история, особенно, если в коде сервис используется писят два раза, и везде потребуется такая замена.

Второе — мы рискуем нарваться на циклические импорты. Конечно, при правильной организации кода такой риск сводится к минимуму, но все же не к нулю. В Django это достаточно распространенная история, и, наверно, каждый, кто писал на этом замечательном фреймворке, сталкивался с такой проблемой.

Серебряной пулей здесь является объект Container с возможностью использования фабрик. Ниже пример, как это работает:

@final
@dataclass(kw_only=True, slots=True, frozen=True)
class BuyProductService:
    product: BuyProductIn

    @log_service_error
    def __call__(self, *, customer: Customer) -> BuyProductOut:
        ... 

def buy_product_service_factory(product: dict) -> BuyProductService:
    return BuyProductService(product=BuyProductIn(**product))


container.register("BuyProductService", factory=buy_product_service_factory)

Здесь мы регистрируем сервис под ключом "BuyProductService" и указываем способ его получения через фабричную функцию buy_product_service_factory. Заметьте, что фабричная функция принимает просто какой-то dict, а уже внутри функции создается DTO-объект BuyProductIn. Минус строковой регистрации — теряем некоторые возможности IDE.

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

class AbstractBuyProductService(abc.ABC):
  @abc.abstractmethod
  def __call__(self, *, customer: Customer) -> BuyProductOut:
     """Выполняет процесс покупки товара для указанного клиента.
        Args:
            customer: Объект клиента, совершающего покупку.
        
        Returns:
            Результат операции покупки, содержащий детали заказа.
        
        Raises:
            CustomerNotActiveError: Если клиент не активен.
            InsufficientFundsError: Если недостаточно средств.
            ProductNotAvailableError: Если товар отсутствует на складе.
        """

    
@final
@dataclass(kw_only=True, slots=True, frozen=True)
class BuyProductService(AbstractBuyProductService):
    product: BuyProductIn

    @log_service_error
    def __call__(self, *, customer: Customer) -> BuyProductOut:
        ... 


def buy_product_service_factory(product: dict) -> BuyProductService:
    return BuyProductService(product=BuyProductIn(**product))


container.register(AbstractBuyProductService, factory=buy_product_service_factory)

Это заставляет писать отдельный класс-абстракцию под каждую реализацию сервиса и немного раздувает кодовую базу. Однако в абстрактном классе вы можете «навалить» солидный док-стринг, в котором объясните, как работает ваш сервис.

Решение о том, какой из вариантов использовать — компромиссное. Если ваш домен сложен/специфичен я бы порекомендовал не отказываться от абстракций, хотя они и добавляют накладных расходов при написании кода.

Итак, вернемся к нашему примеру. Как теперь выглядит «вьюха»:

class BuyProductView(APIView):
    permission_classes = (CustomerRequired,)

    def post(self, request: Request) -> Response:
        serializer = BuyProductSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        service = container.resolve("BuyProductService", product=serializer.validated_data)
        result = service(customer=request.user.customer)

        return Response(
            data=result.asdict(),
            status=status.HTTP_200_OK,
        )

И последний, исключительно технический штрих, чтобы все заработало — необходимо импортировать ваши сервисы в методе ready, чтобы punq успел все зарегистрировать:

from django.apps import AppConfig


class ProductConfig(AppConfig):
    name = "apps.product"

    def ready(self):
        from apps.product import services # <--- тут

Итог:

1. Во «вьюху» больше не нужно импортировать ни сам сервис, ни его входной DTO. Мы отдаем внутрь словарь, а под капотом происходит какая-то магия, которая здесь нас больше не волнует. Внутри самого сервиса мы все еще имеем понятный контракт в виде BuyProductIn.

2. Чтобы получить нужный сервис, следует просто указать тот ключ, под которым его регистрировали. Это можно сделать в любой точке вашего приложения, в том числе в другом сервисе.

3. Если изменить BuyProductIn на BuyProductInV2 «вьюха» об этом не узнает. Все инкапсулировано внутри фабричной функции.

Если сервис зависит от другого сервиса

Распространенная история, когда один сервис нужно использовать внутри другого сервиса. На самом деле, все очень просто. Допустим, что при покупке товара, необходимо передавать информацию в какую-то внешнюю систему, с которой мы вынуждены общаться через HTTP-запросы. Пусть внешняя система называется «CRM».

Ниже пример сервиса, который взаимодействует с «CRM» (детали реализации нас здесь не сильно интересуют):

@final
@dataclass(kw_only=True, slots=True, frozen=True)
class SendProductCRMService:
  @log_service_error
  def __call__(self, *, product: BuyProductIn) -> None:
    ...
    
def send_product_crm_service_factory() -> SendProductCRMService:
    return SendProductCRMService()

container.register("SendProductCRMService", factory=send_product_crm_service_factory)

И вот как можно использовать его функционал внутри BuyProductService:

@final
@dataclass(kw_only=True, slots=True, frozen=True)
class BuyProductService:
    product: BuyProductIn
    crm_sender: SendProductCRMService

    @log_service_error
    def __call__(self, *, customer: Customer) -> BuyProductOut:
        ...

    def _send_product_to_crm(self) -> None:
      try:
        self.crm_sender(product=self.product)
      except BaseServiceError:
         raise ProductPurchaseFailed(product=dict(id=self.product.id, ...))
    
def buy_product_service_factory(product: dict) -> BuyProductService:
    return BuyProductService(
        product=BuyProductIn(**product),
        crm_sender=container.resolve("SendProductCRMService"),
    )


container.register("BuyProductService", factory=buy_product_service_factory)

Конструктор — это публичный контракт класса. Если сервису для работы нужен SendProductCRMService, то эта зависимость должна быть явной. При этом ни сам сервис, ни его пользователи «понятия не имеют», как эта зависимость в него попадает — все решается на уровне фабрики.

Заметьте, что при работе с вложенным сервисом внутри метода основного сервиса, вы можете обрабатывать исключения первого. Это удобно, если, допустим, функционал отправки не так критичен, и мы не должны завершать на этом бизнес процесс, однако должны сделать дополнительные действия (например, уведомить менеджеров и т.п.). Напоследок дополню, что здесь не нужно логировать исключения, все уже реализовано в декораторе log_service_error.

Если же функционал критичен и бизнес процесс нужно завершить — создайте и вызовите здесь другое исключение, как показано в примере выше. Следует взять за правило, что каждый сервис описывает свой бизнес-процесс и должен «разговаривать» на своем языке исключений.

Доменные модели и их логика

Доменная логика = бизнес логика? В целом — да, но я бы хотел немного разделять эти понятия, в рамках нашего подхода это можно представить так:

Доменная логика — это правила и поведение, инкапсулированные внутри самих доменных объектов (сущностей, агрегатов). Её цель — обеспечить целостность и консистентность модели. Критерий для вынесения кода в доменную логику —его многократное использование в разных сервисах.

Бизнес-логика — это координация доменных объектов, внешних систем, транзакций и обработка специфичных ошибок для реализации конкретного сценария (use case). Она принадлежит сервисному слою (сервисам приложения).

Рассмотрим следующий пример:

@final
@dataclass(kw_only=True, slots=True, frozen=True)
class BuyProductService:
    product: BuyProductIn

    @log_service_error
    def __call__(self, *, customer: Customer) -> BuyProductOut:
        product = Product.objects.get(pk=self.product.id)
        
        if customer.can_buy_max_count_of(product) < self.product.count:
            raise NotEnoughBalance(
                product=dict(id=product.pk, price=product.price),
                customer=dict(id=customer.pk, balance=customer.balance),
            )
            
        return self._buy(product=product, customer=customer)

    @transaction.atomic
    def _buy(self, *, product: Product, customer: Customer) -> BuyProductOut:
      ...

У объекта customer есть метод can_buy_max_count_of, куда передается товар. Метод подсчитывает сколько таких товаров может купить данный заказчик. В результате возвращает просто int:

class Customer(models.Model):
    balance = models.PositiveIntegerField("количество рублей на балансе")
    ...

    def can_buy_max_count_of(self, product: Product) -> int:
        if product.price > 0:
          return self.balance // product.price
        return 0

Определение количества доступных для покупки товаров конкретного заказчика — доменная логика модели Customer.

Или возьмем другой пример:

class Product(models.Model):
    title = models.CharField("наименование товара", max_length=256)
    price = models.PositiveIntegerField("цена товара")
    count = models.PositiveIntegerField("остаток на складе", default=0)
    status = models.PositiveSmallIntegerField(
        "статус товара",
        choices=ProductStatusEnum.choices(),
        default=ProductStatusEnum.AVAILABLE,
    )

    @property
    def is_available(self) -> bool:
        return self.status == ProductStatusEnum.AVAILABLE and self.count > 0

Свойство is_available — доменная логика модели Product. Скорее всего это свойство вы будете использовать множество раз, и каждый раз прописывать эти проверки в сервисе — плохая идея.

Таким образом, доменная логика — частный случай бизнес логики, который удобно отнести к конкретным доменным моделям, над которыми эта логика производится.

Как правило, такой код вы будете часто переиспользовать в нескольких разных сервисах. Старайтесь, чтобы ваша доменная логика представляла из себя набор небольших методов. Правильно определяйте контекст доменной модели, так она «не раздуется» и останется, что, называется, «в хорошей форме».

Бонус или «При чем здесь ELK»?

Насколько сильно Вас тревожит, как в вашей команде пишутся логи? По своему опыту могу сказать, что на логи смотрят, только если «что-то сломалось» и нужно восстановить причинно-следственную связь.

Когда я учился в школе, я узнал про понятие «альтернативные издержки». Простыми словами: представьте, что вы стоите на развилке. Выбрав одну дорогу, вы навсегда отказываетесь от всех сокровищ и возможностей, которые таила в себе другая.

Подход команды к логированию — это одна из таких развилок, где разработчик выбирает, как ему относится к своему коду. Глобально есть два пути:

1. Код — прикладной инструмент. Просто реализует нужную бизнес-логику.

2. Код — прикладной и исследовательский инструмент. Может и должен служить объектом для анализа поведения пользователей и всей системы в целом.

Знаете ли вы, сколько раз пользователь, например, «не купил товар потому что у него не было денег на счете»? Или сколько раз «товара не было на складе»? А бизнес вообще знает, что он может собрать такую метрику почти бесплатно? И даже в динамике. А ведь может.

При этом, унифицированная структура лога положительно скажется на вашем поиске в Кибане. Напомню, что в нашем подходе для решения всех этих проблем используется декоратор, который «вешается» на сервисы

@final
@dataclass(kw_only=True, slots=True, frozen=True)
class BuyProductService:
    product: BuyProductIn

    @log_service_error # <--- тут
    def __call__(self, *, customer: Customer) -> BuyProductOut:
        ...

И благодаря этому, мы можем строить, например, такие «дашики»:

А теперь представьте, что можно проводить аналитику для каждого сервиса. Ваш класс в понимании бизнеса становится не просто куском кода, а конкретным бизнес-процессом, с которого можно собирать метрики.

В каком-то смысле это UL в рамках сервисной логики: вы всегда будете говорить об одном и том же бизнес-процессе, имея в виду конкретный класс и его метрики. Вы можете изменить структуру лога под себя, я показал привел лишь пример. Добавьте то, что необходимо Вам и проводите нужную Вам аналитику.

«Потыкаться» с дашбордами можно в репозитории, там уже все настроено, как говорится, «из коробки».

Заключение: Собираем пазл

В этой и предыдущей статьях мы прошли путь от классических Django-паттернов к более структурированному подходу, который лучше масштабируется в сложных приложениях. Самое главное преимущество которое я вижу — можно начать использовать этот подход здесь и сейчас — с учетом всего что мы сказали, он простой и понятный. Вам не нужно перепиливать всю архитектуру, а бизнесу выделять огромные деньги на рефакторинг, чтобы начать внедрять сервисный слой. Давайте подытожим:

Что мы получили в итоге?

• Аккуратные доменные модели с бизнес-логикой, но без ответственности за процессы.

• Сервисы как координаторы бизнес-процессов с явными зависимостями.

• DI-контейнер как инструмент борьбы с жесткими связями и циклическими импортами

• Фабричные функции как способ инкапсуляции сложной логики создания объектов

• Разделение ответственности, где каждый компонент делает одну вещь и делает её хорошо

• Структурированное и унифицированное логирование для сервисных ошибок

Что дальше?

Этот подход открывает дорогу к:

• Простому тестированию (зависимости легко подменяются моками)

• Гибкой конфигурации (разные реализации для разных окружений)

• Постепенному рефакторингу (можно внедрять поэтапно)

Я сознательно не стал углубляться в тему тестирования — она заслуживает отдельной статьи. Но поверьте, с таким подходом ваши тесты станут проще и надежнее. А пайп тестов на Гитлабе не будет занимать 20 минут.

Повторю, что я сказал в начале первой статьи «Представленный подход — не серебряная пуля». Это набор инструментов, которые стоит иметь в арсенале. Используйте их там, где это имеет смысл: в сложных бизнес-процессах, часто меняющихся требованиях, больших командах.

Если у вас есть вопросы, замечания или собственный опыт применения подобных подходов — жду вас в комментариях! Обмен опытом делает нас всех лучше.

Спасибо за внимание!