Как отделяешь бизнес логику от данных?

Отделение бизнес-логики от данных

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

1. Проблема: смешанная логика

Плохо — всё в одном месте:

class User:
    def __init__(self, name: str, email: str):
        self.name = name
        self.email = email
    
    def save(self):
        """Бизнес-логика и данные вместе."""
        # Валидация (бизнес-логика)
        if not self.email:
            raise ValueError("Email обязателен")
        
        # Доступ к БД (работа с данными)
        db.execute("INSERT INTO users (name, email) VALUES (?, ?)",
                   (self.name, self.email))
        
        # Отправка письма (бизнес-логика)
        send_email(self.email, "Welcome!")
    
    def is_premium(self):
        """Зависит от БД, но выглядит как бизнес-логика."""
        subscription = db.query("SELECT * FROM subscriptions WHERE user_id = ?",
                                (self.id,))
        return subscription is not None

Проблемы:

• Невозможно тестировать бизнес-логику без БД
• Изменение БД требует перестройки модели
• Логика разбросана по разным методам
• Сложно переиспользовать

2. Решение: Layered Architecture (многоуровневая архитектура)

Слои:

Presentation (API, CLI, Web)
    ↓ (зависимость)
Application (Use Cases, Commands)
    ↓
Domain (Entities, Value Objects, Business Logic)
    ↓
Infrastructure (Repositories, External Services)

3. Domain Layer — Чистая бизнес-логика

from dataclasses import dataclass
from typing import Optional
from datetime import datetime

@dataclass
class User:
    """Domain Entity — только данные и бизнес-логика, БЕЗ зависимостей."""
    id: str
    name: str
    email: str
    created_at: datetime
    
    def validate(self) -> None:
        """Валидация — чистая бизнес-логика."""
        if not self.name or len(self.name) < 2:
            raise ValueError("Имя должно быть не менее 2 символов")
        
        if "@" not in self.email:
            raise ValueError("Email неверный")
    
    def is_new_user(self, days_threshold: int = 30) -> bool:
        """Бизнес-логика: определение новых пользователей."""
        delta = datetime.utcnow() - self.created_at
        return delta.days <= days_threshold
    
    def is_premium(self, subscription: Optional[dict]) -> bool:
        """Бизнес-логика зависит от данных, но не от БД прямо."""
        return subscription is not None and subscription["status"] == "active"

4. Application Layer — Use Cases

from abc import ABC, abstractmethod
from typing import Protocol

class UserRepository(Protocol):
    """Интерфейс для доступа к данным (Abstract Repository)."""
    
    async def find_by_email(self, email: str) -> Optional[User]:
        ...
    
    async def save(self, user: User) -> User:
        ...

class EmailService(Protocol):
    """Интерфейс для внешних сервисов."""
    
    async def send(self, to: str, subject: str, body: str) -> bool:
        ...

class CreateUserUseCase:
    """Application Service — оркестрирует доменную логику."""
    
    def __init__(self, 
                 user_repository: UserRepository,
                 email_service: EmailService):
        self.user_repo = user_repository
        self.email_svc = email_service
    
    async def execute(self, name: str, email: str) -> User:
        # 1. Проверяем, что пользователя нет
        existing = await self.user_repo.find_by_email(email)
        if existing:
            raise ValueError("Пользователь уже существует")
        
        # 2. Создаём доменный объект
        user = User(
            id=generate_id(),
            name=name,
            email=email,
            created_at=datetime.utcnow()
        )
        
        # 3. Валидируем (доменная логика)
        user.validate()
        
        # 4. Сохраняем через репозиторий
        saved_user = await self.user_repo.save(user)
        
        # 5. Отправляем письмо
        await self.email_svc.send(
            to=user.email,
            subject="Welcome!",
            body=f"Hi {user.name}!"
        )
        
        return saved_user

5. Infrastructure Layer — Реализация доступа к данным

from sqlalchemy import Column, String, DateTime
from sqlalchemy.orm import Session

class UserORM:
    """ORM модель — только для работы с БД."""
    __tablename__ = "users"
    id: Column(String, primary_key=True)
    name: Column(String)
    email: Column(String, unique=True)
    created_at: Column(DateTime)

class SQLAlchemyUserRepository:
    """Реализация репозитория для PostgreSQL."""
    
    def __init__(self, session: Session):
        self.session = session
    
    async def find_by_email(self, email: str) -> Optional[User]:
        """Доступ к БД скрыт от доменной логики."""
        orm_user = self.session.query(UserORM).filter_by(email=email).first()
        if not orm_user:
            return None
        return self._orm_to_domain(orm_user)
    
    async def save(self, user: User) -> User:
        """Преобразуем доменный объект в ORM."""
        orm_user = UserORM(
            id=user.id,
            name=user.name,
            email=user.email,
            created_at=user.created_at
        )
        self.session.add(orm_user)
        self.session.commit()
        return user
    
    @staticmethod
    def _orm_to_domain(orm_user: UserORM) -> User:
        """Маппинг ORM → Domain Entity."""
        return User(
            id=orm_user.id,
            name=orm_user.name,
            email=orm_user.email,
            created_at=orm_user.created_at
        )

6. Presentation Layer — API

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class CreateUserRequest(BaseModel):
    """DTO для входных данных."""
    name: str
    email: str

class UserResponse(BaseModel):
    """DTO для выходных данных."""
    id: str
    name: str
    email: str

@app.post("/users")
async def create_user(request: CreateUserRequest) -> UserResponse:
    """Presentation layer вызывает Use Case."""
    try:
        user = await create_user_use_case.execute(
            name=request.name,
            email=request.email
        )
        return UserResponse(
            id=user.id,
            name=user.name,
            email=user.email
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

7. Тестирование: мок-объекты вместо БД

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_create_user_success():
    # Mock репозитория
    mock_repo = AsyncMock(spec=UserRepository)
    mock_repo.find_by_email.return_value = None
    mock_repo.save.return_value = User(...)
    
    # Mock email сервиса
    mock_email = AsyncMock(spec=EmailService)
    mock_email.send.return_value = True
    
    # Создаём use case с мок-объектами
    use_case = CreateUserUseCase(mock_repo, mock_email)
    
    # Выполняем
    result = await use_case.execute("John", "john@example.com")
    
    # Проверяем
    assert result.email == "john@example.com"
    mock_repo.save.assert_called_once()
    mock_email.send.assert_called_once()

@pytest.mark.asyncio
async def test_create_user_already_exists():
    mock_repo = AsyncMock(spec=UserRepository)
    mock_repo.find_by_email.return_value = User(...)  # Уже существует
    
    use_case = CreateUserUseCase(mock_repo, AsyncMock())
    
    with pytest.raises(ValueError, match="уже существует"):
        await use_case.execute("John", "john@example.com")

8. Dependency Injection для связывания слоёв

from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    # Infrastructure
    db = providers.Singleton(DatabaseConnection)
    user_repo = providers.Factory(
        SQLAlchemyUserRepository,
        session=db.session
    )
    email_svc = providers.Factory(EmailService)
    
    # Application
    create_user_use_case = providers.Factory(
        CreateUserUseCase,
        user_repository=user_repo,
        email_service=email_svc
    )

# Использование
container = Container()
use_case = container.create_user_use_case()
await use_case.execute("John", "john@example.com")

9. Дизайн паттерны для отделения

Repository Pattern — скрывает детали доступа к БД Service/UseCase — оркестрирует бизнес-логику DTO — преобразует данные между слоями Value Objects — неизменяемые объекты для бизнес-концепций

from dataclasses import dataclass

@dataclass(frozen=True)  # Immutable
class Email:
    """Value Object для Email."""
    value: str
    
    def __post_init__(self):
        if "@" not in self.value:
            raise ValueError("Неверный email")

@dataclass(frozen=True)
class Money:
    """Value Object для денег."""
    amount: float
    currency: str = "USD"
    
    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Разные валюты")
        return Money(self.amount + other.amount, self.currency)

10. Плюсы правильного разделения

✅ Тестируемость — тестируешь логику без БД ✅ Переиспользуемость — один use case для разных интерфейсов ✅ Масштабируемость — меняешь БД без изменения логики ✅ Поддерживаемость — каждый слой отвечает за свою область ✅ Независимость — слои могут эволюционировать отдельно

Правило: Доменная логика (Entity, Value Object, Domain Service) не должна знать ничего о:

• БД (SQL, ORM, миграции)
• Внешних сервисах (API, очереди)
• Фреймворках (FastAPI, Django)
• Деталях реализации (как сохранять, где брать)

Бизнес-логика — это суть приложения, всё остальное — детали реализации.