Применение бизнес-логики
Application Layer в Clean Architecture
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
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
)
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) │
└─────────────────────────────────┘
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]
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)
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)
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)
| Сценарий | Рекомендация |
|---|---|
| Сложная бизнес-логика | ✅ Обязательно |
| Clean Architecture | ✅ Обязательно |
| Много шагов в процессе | ✅ Рекомендуется |
| Простое CRUD | ⚠️ Может быть избыточно |
| Быстрый прототип | ⚠️ Можно пропустить |
"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