Domain-Driven Design

Что такое DDD?

Domain-Driven Design — подход к разработке сложных систем, при котором основное внимание уделяется предметной области (домену), а не техническим деталям.

🔹 Проблема: разрыв между кодом и реальностью

# ❌ Код не отражает реальность
class UserTable:
    def __init__(self):
        self.id = None
        self.name = None
        self.email = None

# Что такое "UserTable"? Это таблица? Или пользователь?

Код говорит на языке БД, а не на языке бизнеса

✅ Решение: язык домена (Ubiquitous Language)

# ✅ Код отражает реальность
class User:
    def __init__(self, name: str, email: Email):
        self.name = name
        self.email = email
    
    def change_email(self, new_email: Email):
        if not new_email.is_valid():
            raise ValueError("Invalid email")
        self.email = new_email

Код говорит на языке бизнеса

Основные концепции DDD

• Entity — объект с уникальным идентификатором
• Value Object — объект без идентификатора
• Aggregate — группа связанных объектов
• Repository — абстракция хранилища
• Domain Service — бизнес-логика вне Entity
• Bounded Context — границы домена

Entity (Сущность)

Объект с уникальным идентификатором. Две сущности равны, если их ID равны.

class User:  # Entity
    def __init__(self, user_id: str, name: str):
        self.id = user_id  # Уникальный идентификатор
        self.name = name
    
    def __eq__(self, other):
        return isinstance(other, User) and self.id == other.id

# Два пользователя с одинаковым ID — одна сущность
user1 = User("123", "Alice")
user2 = User("123", "Bob")  # Другой name, но тот же ID
assert user1 == user2  # True!

Value Object (Объект-значение)

Объект без идентификатора. Равенство по значению всех полей.

class Email:  # Value Object
    def __init__(self, address: str):
        if "@" not in address:
            raise ValueError("Invalid email")
        self.address = address.lower()
    
    def __eq__(self, other):
        return isinstance(other, Email) and self.address == other.address

email1 = Email("user@example.com")
email2 = Email("USER@EXAMPLE.COM")  # Разный регистр
assert email1 == email2  # True! Значение одинаковое

💡 Value Objects неизменяемы (immutable)

Aggregate (Агрегат)

Группа связанных объектов, которые рассматриваются как единое целое.

class Order:  # Aggregate Root
    def __init__(self, order_id: str, customer: Customer):
        self.id = order_id
        self.customer = customer
        self.items = []  # List[OrderItem]
        self.status = "pending"
    
    def add_item(self, product: Product, quantity: int):
        if self.status != "pending":
            raise ValueError("Cannot modify completed order")
        item = OrderItem(product, quantity)
        self.items.append(item)
    
    def calculate_total(self):
        return sum(item.price * item.quantity for item in self.items)

# OrderItem — часть агрегата Order
class OrderItem:
    def __init__(self, product: Product, quantity: int):
        self.product = product
        self.quantity = quantity

🔹 Агрегат защищает свою целостность

Repository (Репозиторий)

Абстракция для работы с хранилищем. Скрывает детали БД.

from abc import ABC, abstractmethod

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> None:
        pass
    
    @abstractmethod
    def find_by_id(self, user_id: str) -> User:
        pass
    
    @abstractmethod
    def find_by_email(self, email: Email) -> User:
        pass

# Реализация для PostgreSQL
class PostgresUserRepository(UserRepository):
    def __init__(self, db):
        self.db = db
    
    def save(self, user: User):
        self.db.execute(
            "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
            (user.id, user.name, user.email.address)
        )
    
    def find_by_id(self, user_id: str) -> User:
        row = self.db.fetch_one("SELECT * FROM users WHERE id = ?", (user_id,))
        return User(row['id'], row['name'], Email(row['email']))

# Реализация для MongoDB
class MongoUserRepository(UserRepository):
    def save(self, user: User):
        self.collection.insert_one({
            "_id": user.id,
            "name": user.name,
            "email": user.email.address
        })

Domain Service (Доменный сервис)

Бизнес-логика, которая не принадлежит конкретной Entity.

class TransferService:  # Domain Service
    def __init__(self, account_repo: AccountRepository):
        self.account_repo = account_repo
    
    def transfer(self, from_id: str, to_id: str, amount: Money):
        from_account = self.account_repo.find_by_id(from_id)
        to_account = self.account_repo.find_by_id(to_id)
        
        # Логика перевода не принадлежит Account
        if from_account.balance < amount:
            raise ValueError("Insufficient funds")
        
        from_account.withdraw(amount)
        to_account.deposit(amount)
        
        self.account_repo.save(from_account)
        self.account_repo.save(to_account)

💡 Когда логика не "вписывается" в Entity → Domain Service

Bounded Context (Ограниченный контекст)

Границы, внутри которых модель имеет одно значение.

┌─────────────────────┐  ┌─────────────────────┐
│  E-Commerce Context │  │  Shipping Context   │
│                     │  │                     │
│  User: покупатель   │  │  User: получатель   │
│  - name             │  │  - address          │
│  - email            │  │  - phone            │
│  - cart             │  │  - delivery_prefs   │
└─────────────────────┘  └─────────────────────┘

Одна сущность "User" имеет разное значение 
в разных контекстах!

Слои DDD

┌─────────────────────────────────┐
│   Presentation Layer            │ ← UI, API
├─────────────────────────────────┤
│   Application Layer             │ ← Use Cases, координация
├─────────────────────────────────┤
│   Domain Layer                  │ ← Entity, Value Objects, Domain Services
├─────────────────────────────────┤
│   Infrastructure Layer          │ ← DB, HTTP, Email
└─────────────────────────────────┘

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

ecommerce/
  ├── domain/              # Domain Layer
  │   ├── entities/
  │   │   ├── user.py
  │   │   ├── order.py
  │   │   └── product.py
  │   ├── value_objects/
  │   │   ├── email.py
  │   │   ├── money.py
  │   │   └── address.py
  │   ├── services/
  │   │   └── transfer_service.py
  │   └── repositories/
  │       └── user_repository.py
  │
  ├── application/         # Application Layer
  │   ├── usecases/
  │   │   ├── create_user.py
  │   │   ├── place_order.py
  │   │   └── transfer_money.py
  │   └── dto/
  │       └── user_dto.py
  │
  ├── infrastructure/      # Infrastructure Layer
  │   ├── persistence/
  │   │   └── postgres_user_repository.py
  │   ├── email/
  │   │   └── smtp_email_service.py
  │   └── api/
  │       └── fastapi_app.py
  │
  └── presentation/        # Presentation Layer
      └── api/
          └── routes.py

🔹 Пример: Application Layer

# application/usecases/create_user.py
class CreateUserUseCase:
    def __init__(self, user_repo: UserRepository, email_service: EmailService):
        self.user_repo = user_repo
        self.email_service = email_service
    
    def execute(self, name: str, email: str) -> UserDTO:
        # 1. Валидация
        email_obj = Email(email)
        
        # 2. Проверка существования
        if self.user_repo.find_by_email(email_obj):
            raise ValueError("User already exists")
        
        # 3. Создание сущности
        user = User(id=generate_id(), name=name, email=email_obj)
        
        # 4. Сохранение
        self.user_repo.save(user)
        
        # 5. Отправка приветственного email
        self.email_service.send_welcome(user.email)
        
        # 6. Возврат DTO
        return UserDTO.from_entity(user)

Практический пример: Система обучения

Моделируем домен для ML-платформы

🔹 Domain: Entity

# domain/entities/training_job.py
class TrainingJob:  # Aggregate Root
    def __init__(self, job_id: str, dataset: Dataset, model_config: ModelConfig):
        self.id = job_id
        self.dataset = dataset
        self.config = model_config
        self.status = JobStatus.PENDING
        self.metrics = Metrics()
        self.created_at = datetime.now()
    
    def start(self):
        if self.status != JobStatus.PENDING:
            raise ValueError("Job already started")
        self.status = JobStatus.RUNNING
    
    def complete(self, metrics: Metrics):
        if self.status != JobStatus.RUNNING:
            raise ValueError("Job not running")
        self.status = JobStatus.COMPLETED
        self.metrics = metrics
    
    def fail(self, error: str):
        self.status = JobStatus.FAILED
        self.error_message = error

🔹 Domain: Value Object

# domain/value_objects/metrics.py
class Metrics:  # Value Object
    def __init__(self, accuracy: float, loss: float, f1_score: float):
        if not (0 <= accuracy <= 1):
            raise ValueError("Accuracy must be between 0 and 1")
        self.accuracy = accuracy
        self.loss = loss
        self.f1_score = f1_score
    
    def __eq__(self, other):
        return (
            isinstance(other, Metrics) and
            self.accuracy == other.accuracy and
            self.loss == other.loss and
            self.f1_score == other.f1_score
        )
    
    def is_better_than(self, other: Metrics) -> bool:
        return self.accuracy > other.accuracy

🔹 Domain: Repository

# domain/repositories/training_job_repository.py
from abc import ABC, abstractmethod

class TrainingJobRepository(ABC):
    @abstractmethod
    def save(self, job: TrainingJob) -> None:
        pass
    
    @abstractmethod
    def find_by_id(self, job_id: str) -> TrainingJob:
        pass
    
    @abstractmethod
    def find_by_status(self, status: JobStatus) -> List[TrainingJob]:
        pass

🔹 Application: Use Case

# application/usecases/start_training.py
class StartTrainingUseCase:
    def __init__(
        self,
        job_repo: TrainingJobRepository,
        dataset_repo: DatasetRepository,
        trainer: ModelTrainer
    ):
        self.job_repo = job_repo
        self.dataset_repo = dataset_repo
        self.trainer = trainer
    
    def execute(self, job_id: str):
        # 1. Получаем job
        job = self.job_repo.find_by_id(job_id)
        
        # 2. Запускаем job
        job.start()
        self.job_repo.save(job)
        
        # 3. Запускаем обучение (async)
        self.trainer.train_async(job, self._on_complete)
    
    def _on_complete(self, job_id: str, metrics: Metrics):
        job = self.job_repo.find_by_id(job_id)
        job.complete(metrics)
        self.job_repo.save(job)

🔹 Infrastructure: Repository Implementation

# infrastructure/persistence/postgres_training_job_repository.py
class PostgresTrainingJobRepository(TrainingJobRepository):
    def __init__(self, db):
        self.db = db
    
    def save(self, job: TrainingJob):
        self.db.execute("""
            INSERT INTO training_jobs 
            (id, dataset_id, config, status, metrics, created_at)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT (id) DO UPDATE SET
                status = EXCLUDED.status,
                metrics = EXCLUDED.metrics
        """, (
            job.id,
            job.dataset.id,
            json.dumps(job.config.to_dict()),
            job.status.value,
            json.dumps(job.metrics.to_dict()) if job.metrics else None,
            job.created_at
        ))
    
    def find_by_id(self, job_id: str) -> TrainingJob:
        row = self.db.fetch_one(
            "SELECT * FROM training_jobs WHERE id = ?",
            (job_id,)
        )
        return self._to_entity(row)
    
    def _to_entity(self, row) -> TrainingJob:
        # Преобразование из БД в Entity
        dataset = self.dataset_repo.find_by_id(row['dataset_id'])
        config = ModelConfig.from_dict(json.loads(row['config']))
        job = TrainingJob(row['id'], dataset, config)
        job.status = JobStatus(row['status'])
        if row['metrics']:
            job.metrics = Metrics.from_dict(json.loads(row['metrics']))
        return job

🔹 Presentation: API

# presentation/api/routes.py
from fastapi import FastAPI, Depends
from application.usecases.start_training import StartTrainingUseCase

app = FastAPI()

@app.post("/training-jobs/{job_id}/start")
def start_training(
    job_id: str,
    use_case: StartTrainingUseCase = Depends(get_start_training_use_case)
):
    try:
        use_case.execute(job_id)
        return {"status": "started"}
    except ValueError as e:
        return {"error": str(e)}, 400

Когда использовать DDD?

✅ Хорошо подходит для:

• Сложных бизнес-доменов (банки, e-commerce, медицина)
• Долгосрочных проектов (код живет годами)
• Больших команд (нужна общая модель)
• Часто меняющихся требований (домен эволюционирует)

❌ Не подходит для:

• Простых CRUD-приложений (избыточно)
• Прототипов/MVP (слишком сложно)
• Технических доменов (где нет бизнес-логики)
• Маленьких команд (overhead)

💡 Альтернативы DDD

Антипаттерны DDD

❌ Anemic Domain Model

# ❌ Модель без поведения
class User:
    def __init__(self):
        self.id = None
        self.name = None
        self.email = None

# Вся логика в сервисах
class UserService:
    def validate_user(self, user: User):
        return "@" in user.email
    
    def change_email(self, user: User, new_email: str):
        user.email = new_email

Модель — просто "мешок с данными"

✅ Rich Domain Model

# ✅ Модель с поведением
class User:
    def __init__(self, user_id: str, name: str, email: Email):
        self.id = user_id
        self.name = name
        self.email = email
    
    def change_email(self, new_email: Email):
        if not new_email.is_valid():
            raise ValueError("Invalid email")
        self.email = new_email
    
    def is_valid(self) -> bool:
        return self.email.is_valid() and len(self.name) > 0

Логика рядом с данными

❌ Утечка инфраструктуры в домен

# ❌ Домен зависит от БД
class User:
    def save(self):
        db.execute("INSERT INTO users ...")  # Инфраструктура!

# ✅ Домен чистый
class User:
    def change_email(self, new_email: Email):
        self.email = new_email

# Репозиторий в инфраструктуре
class UserRepository:
    def save(self, user: User):
        db.execute("INSERT INTO users ...")

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

1. Ubiquitous Language — код говорит на языке бизнеса
2. Rich Domain Model — логика в домене, а не в сервисах
3. Bounded Context — границы домена
4. Layered Architecture — домен независим от инфраструктуры
5. Aggregates — целостность данных

"DDD — это не про код. Это про мышление.
Это про то, как понимать проблему и выражать решение."