Use Cases

Применение бизнес-логики

Application Layer в Clean Architecture

Что такое Use Case?

Use Case — описание того, как система должна вести себя в ответ на запрос пользователя или системы для достижения определённой цели.

🔹 Проблема: логика в контроллере

# ❌ Вся логика в контроллере
class UserController:
    def create_user(self, name, email, password):
        # Валидация
        if not name:
            raise ValueError("Name is required")
        if "@" not in email:
            raise ValueError("Invalid email")
        
        # Проверка существования
        if user_repository.find_by_email(email):
            raise ValueError("User already exists")
        
        # Создание
        user = User(name, email, password)
        user_repository.save(user)
        
        # Отправка email
        email_service.send_welcome_email(email)
        
        # Логирование
        logger.info(f"User created: {email}")
        
        return user

# Проблемы:
# - Контроллер делает слишком много
# - Логику сложно переиспользовать
# - Сложно тестировать

✅ Решение: Use Case

# ✅ Логика в Use Case
class CreateUserUseCase:
    def __init__(self, user_repository, email_service, logger):
        self.user_repository = user_repository
        self.email_service = email_service
        self.logger = logger
    
    def execute(self, name: str, email: str, password: str) -> UserDTO:
        # Валидация
        self._validate(name, email, password)
        
        # Проверка существования
        if self.user_repository.find_by_email(email):
            raise ValueError("User already exists")
        
        # Создание
        user = User.create(name, email, password)
        self.user_repository.save(user)
        
        # Отправка email
        self.email_service.send_welcome_email(email)
        
        # Логирование
        self.logger.info(f"User created: {email}")
        
        return UserDTO.from_entity(user)
    
    def _validate(self, name, email, password):
        if not name:
            raise ValueError("Name is required")
        if "@" not in email:
            raise ValueError("Invalid email")

# Контроллер становится тонким
class UserController:
    def __init__(self, create_user_use_case):
        self.create_user_use_case = create_user_use_case
    
    def create_user(self, request: CreateUserRequest):
        return self.create_user_use_case.execute(
            request.name, request.email, request.password
        )

Структура Use Case

🔹 Базовый Use Case

from abc import ABC, abstractmethod

class UseCase(ABC):
    @abstractmethod
    def execute(self, *args, **kwargs):
        pass

class CreateUserUseCase(UseCase):
    def __init__(self, user_repository, email_service):
        self.user_repository = user_repository
        self.email_service = email_service
    
    def execute(self, request: CreateUserRequest) -> UserDTO:
        # 1. Валидация входных данных
        self._validate(request)
        
        # 2. Проверка бизнес-правил
        self._check_business_rules(request)
        
        # 3. Выполнение операции
        result = self._perform_operation(request)
        
        # 4. Побочные эффекты (опционально)
        self._handle_side_effects(request, result)
        
        # 5. Возврат результата
        return result

🔹 Слои приложения

┌─────────────────────────────────┐
│   Presentation Layer            │
│   (Controllers, API)            │
└──────────────┬──────────────────┘
               │
               ↓
┌─────────────────────────────────┐
│   Application Layer             │
│   (Use Cases)                   │
│   - CreateUserUseCase           │
│   - GetUserUseCase              │
│   - UpdateUserUseCase           │
└──────────────┬──────────────────┘
               │
               ↓
┌─────────────────────────────────┐
│   Domain Layer                  │
│   (Entities, Domain Services)   │
└─────────────────────────────────┘

Примеры Use Cases

🔹 CreateUserUseCase

class CreateUserUseCase:
    def __init__(
        self,
        user_repository: UserRepository,
        email_service: EmailService,
        password_hasher: PasswordHasher,
        logger: Logger
    ):
        self.user_repository = user_repository
        self.email_service = email_service
        self.password_hasher = password_hasher
        self.logger = logger
    
    def execute(self, request: CreateUserRequest) -> UserDTO:
        # 1. Валидация
        if not request.name or len(request.name) < 2:
            raise ValueError("Name must be at least 2 characters")
        if not self._is_valid_email(request.email):
            raise ValueError("Invalid email")
        if len(request.password) < 8:
            raise ValueError("Password must be at least 8 characters")
        
        # 2. Проверка существования
        if self.user_repository.find_by_email(request.email):
            raise ValueError("User with this email already exists")
        
        # 3. Создание пользователя
        hashed_password = self.password_hasher.hash(request.password)
        user = User.create(request.name, request.email, hashed_password)
        self.user_repository.save(user)
        
        # 4. Отправка приветственного email
        self.email_service.send_welcome_email(user.email)
        
        # 5. Логирование
        self.logger.info(f"User created: {user.email}")
        
        # 6. Возврат DTO
        return UserDTO.from_entity(user)
    
    def _is_valid_email(self, email: str) -> bool:
        return "@" in email and "." in email.split("@")[1]

🔹 GetUserUseCase

class GetUserUseCase:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    def execute(self, user_id: int) -> UserDTO:
        # 1. Получение пользователя
        user = self.user_repository.find_by_id(user_id)
        if not user:
            raise ValueError("User not found")
        
        # 2. Возврат DTO
        return UserDTO.from_entity(user)

# С дополнительной логикой
class GetUserWithOrdersUseCase:
    def __init__(
        self,
        user_repository: UserRepository,
        order_repository: OrderRepository
    ):
        self.user_repository = user_repository
        self.order_repository = order_repository
    
    def execute(self, user_id: int) -> UserDetailDTO:
        # 1. Получение пользователя
        user = self.user_repository.find_by_id(user_id)
        if not user:
            raise ValueError("User not found")
        
        # 2. Получение заказов
        orders = self.order_repository.find_by_user_id(user_id)
        
        # 3. Возврат DTO с заказами
        return UserDetailDTO.from_entities(user, orders)

🔹 PlaceOrderUseCase

class PlaceOrderUseCase:
    def __init__(
        self,
        user_repository: UserRepository,
        order_repository: OrderRepository,
        product_repository: ProductRepository,
        inventory_service: InventoryService,
        payment_service: PaymentService,
        event_bus: EventBus
    ):
        self.user_repository = user_repository
        self.order_repository = order_repository
        self.product_repository = product_repository
        self.inventory_service = inventory_service
        self.payment_service = payment_service
        self.event_bus = event_bus
    
    def execute(self, request: PlaceOrderRequest) -> OrderDTO:
        # 1. Проверка пользователя
        user = self.user_repository.find_by_id(request.user_id)
        if not user:
            raise ValueError("User not found")
        
        # 2. Проверка наличия товаров
        for item in request.items:
            product = self.product_repository.find_by_id(item.product_id)
            if not product:
                raise ValueError(f"Product {item.product_id} not found")
            if not self.inventory_service.is_available(item.product_id, item.quantity):
                raise ValueError(f"Not enough inventory for product {item.product_id}")
        
        # 3. Создание заказа
        order = Order.create(request.user_id, request.items)
        self.order_repository.save(order)
        
        # 4. Резервирование товаров
        for item in request.items:
            self.inventory_service.reserve(item.product_id, item.quantity)
        
        # 5. Обработка платежа
        payment = self.payment_service.process_payment(
            order.total, request.payment_method
        )
        order.confirm_payment(payment.id)
        self.order_repository.save(order)
        
        # 6. Публикация события
        self.event_bus.publish("order_placed", {
            "order_id": order.id,
            "user_id": order.user_id,
            "total": order.total
        })
        
        # 7. Возврат DTO
        return OrderDTO.from_entity(order)

Тестирование Use Cases

🔹 Unit тесты

from unittest.mock import Mock

def test_create_user_use_case():
    # Arrange
    user_repository = Mock(spec=UserRepository)
    user_repository.find_by_email.return_value = None
    user_repository.save = Mock()
    
    email_service = Mock(spec=EmailService)
    email_service.send_welcome_email = Mock()
    
    password_hasher = Mock(spec=PasswordHasher)
    password_hasher.hash.return_value = "hashed_password"
    
    logger = Mock(spec=Logger)
    
    use_case = CreateUserUseCase(
        user_repository, email_service, password_hasher, logger
    )
    
    request = CreateUserRequest(
        name="Alice",
        email="alice@example.com",
        password="password123"
    )
    
    # Act
    result = use_case.execute(request)
    
    # Assert
    assert result.name == "Alice"
    assert result.email == "alice@example.com"
    user_repository.save.assert_called_once()
    email_service.send_welcome_email.assert_called_once_with("alice@example.com")

def test_create_user_use_case_user_exists():
    # Arrange
    user_repository = Mock(spec=UserRepository)
    user_repository.find_by_email.return_value = User(id=1, name="Alice", email="alice@example.com")
    
    use_case = CreateUserUseCase(user_repository, Mock(), Mock(), Mock())
    request = CreateUserRequest(name="Alice", email="alice@example.com", password="password123")
    
    # Act & Assert
    with pytest.raises(ValueError, match="already exists"):
        use_case.execute(request)

Преимущества Use Cases

  • Разделение ответственности — каждый Use Case делает одну вещь
  • Переиспользование — Use Case можно использовать из разных мест
  • Тестируемость — легко тестировать изолированно
  • Читаемость — код понятен и структурирован
  • Независимость — Use Case не зависит от UI
  • Документация — Use Case описывает бизнес-процесс

Недостатки Use Cases

  • Дополнительный слой — больше кода
  • Сложность — для простых операций может быть избыточно
  • Overhead — нужно создавать классы для каждого Use Case

💡 Когда использовать Use Cases?

Сценарий Рекомендация
Сложная бизнес-логика ✅ Обязательно
Clean Architecture ✅ Обязательно
Много шагов в процессе ✅ Рекомендуется
Простое CRUD ⚠️ Может быть избыточно
Быстрый прототип ⚠️ Можно пропустить

🎯 Ключевые принципы Use Cases

  1. Одна ответственность — каждый Use Case решает одну задачу
  2. Независимость от UI — Use Case не знает о представлении
  3. Явные зависимости — все зависимости передаются в конструктор
  4. Валидация — валидация входных данных в Use Case
  5. Обработка ошибок — явная обработка ошибок
"Use Case — это не про код. Это про бизнес-процессы.
Use Case описывает, что система должна делать, а не как она это делает."

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

application/
  ├── usecases/
  │   ├── user/
  │   │   ├── create_user_use_case.py
  │   │   ├── get_user_use_case.py
  │   │   └── update_user_use_case.py
  │   ├── order/
  │   │   ├── place_order_use_case.py
  │   │   └── cancel_order_use_case.py
  │   └── __init__.py
  ├── dto/
  │   ├── user_dto.py
  │   └── order_dto.py
  └── __init__.py