Как построить архитектуру на основе DDD?

Архитектура на основе Domain-Driven Design (DDD)

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

Слои архитектуры DDD

┌─────────────────────────────────────────────┐
│         Presentation (FastAPI, Django)      │  Контроллеры, сериализация
├─────────────────────────────────────────────┤
│          Application (Use Cases)            │  Orchestration, транзакции
├─────────────────────────────────────────────┤
│     Domain (Business Logic, Entities)       │  Rules, algorithms (защищено)
├─────────────────────────────────────────────┤
│  Infrastructure (Database, External APIs)   │  Технические детали
└─────────────────────────────────────────────┘

Правило зависимостей: только от центра к периферии. Domain не знает о Database.

1. Domain слой (ядро приложения)

Entities (Сущности)

# domain/models/user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from enum import Enum

class UserStatus(Enum):
    ACTIVE = "active"
    BANNED = "banned"
    DELETED = "deleted"

@dataclass
class User:
    """Domain Entity - не зависит от БД"""
    id: str
    email: str
    password_hash: str
    status: UserStatus
    created_at: datetime
    
    def is_active(self) -> bool:
        return self.status == UserStatus.ACTIVE
    
    def ban_user(self, reason: str) -> None:
        if not self.is_active():
            raise ValueError(f"User already {self.status.value}")
        self.status = UserStatus.BANNED

@dataclass
class Order:
    """Domain Entity"""
    id: str
    user_id: str
    items: list['OrderItem']
    total_price: float
    created_at: datetime
    
    def add_item(self, item: 'OrderItem') -> None:
        if item.quantity <= 0:
            raise ValueError("Quantity must be positive")
        self.items.append(item)
    
    def calculate_total(self) -> float:
        return sum(item.price * item.quantity for item in self.items)

Value Objects (Значимые объекты)

# domain/models/value_objects.py
from dataclasses import dataclass
from typing import Pattern
import re

@dataclass(frozen=True)
class Email:
    """Value Object - неизменяем, сравнивается по значению"""
    value: str
    
    def __post_init__(self):
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, self.value):
            raise ValueError(f"Invalid email: {self.value}")

@dataclass(frozen=True)
class Money:
    """Value Object для денег"""
    amount: float
    currency: str
    
    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")
        if self.currency not in ['USD', 'EUR', 'RUB']:
            raise ValueError(f"Unsupported currency: {self.currency}")
    
    def add(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies")
        return Money(self.amount + other.amount, self.currency)

# Использование
email = Email("user@example.com")
price = Money(99.99, "USD")

Repository интерфейсы

# domain/repositories/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional, List
from domain.models import User

class IUserRepository(ABC):
    """Interface - не зависит от реализации"""
    
    @abstractmethod
    async def get_by_id(self, user_id: str) -> Optional[User]:
        """Получить пользователя по ID"""
        pass
    
    @abstractmethod
    async def get_by_email(self, email: str) -> Optional[User]:
        """Получить пользователя по email"""
        pass
    
    @abstractmethod
    async def save(self, user: User) -> None:
        """Сохранить пользователя"""
        pass
    
    @abstractmethod
    async def delete(self, user_id: str) -> None:
        """Удалить пользователя"""
        pass

Domain Services

# domain/services/user_service.py
from domain.repositories import IUserRepository
from domain.models import User, UserStatus
from datetime import datetime

class UserDomainService:
    """Pure бизнес-логика, без side effects"""
    
    def __init__(self, user_repo: IUserRepository):
        self.user_repo = user_repo
    
    async def create_user(self, email: str, password: str) -> User:
        """Создать пользователя с проверками"""
        # Проверяем уникальность
        existing = await self.user_repo.get_by_email(email)
        if existing:
            raise ValueError(f"User {email} already exists")
        
        # Создаём сущность с бизнес-правилами
        user = User(
            id=str(uuid4()),
            email=email,
            password_hash=hash_password(password),  # Безопасность
            status=UserStatus.ACTIVE,
            created_at=datetime.utcnow()
        )
        
        return user
    
    async def deactivate_user(self, user_id: str) -> None:
        """Деактивировать пользователя"""
        user = await self.user_repo.get_by_id(user_id)
        if not user:
            raise ValueError(f"User {user_id} not found")
        
        # Бизнес-правило
        if user.status == UserStatus.DELETED:
            raise ValueError("Cannot deactivate deleted user")
        
        user.ban_user("Deactivation")
        await self.user_repo.save(user)

2. Application слой (Use Cases)

# application/use_cases/create_order.py
from dataclasses import dataclass
from typing import List
from domain.models import Order, OrderItem
from domain.repositories import IOrderRepository, IUserRepository

@dataclass
class CreateOrderRequest:
    user_id: str
    item_ids: List[str]

@dataclass
class CreateOrderResponse:
    order_id: str
    total_price: float

class CreateOrderUseCase:
    """Orchestrator для бизнес-процесса"""
    
    def __init__(
        self,
        order_repo: IOrderRepository,
        user_repo: IUserRepository,
        product_service: 'IProductService'  # Еще одна абстракция
    ):
        self.order_repo = order_repo
        self.user_repo = user_repo
        self.product_service = product_service
    
    async def execute(self, request: CreateOrderRequest) -> CreateOrderResponse:
        # 1. Проверяем пользователя
        user = await self.user_repo.get_by_id(request.user_id)
        if not user or not user.is_active():
            raise ValueError("User not active")
        
        # 2. Получаем товары
        items = []
        for item_id in request.item_ids:
            product = await self.product_service.get_product(item_id)
            if not product:
                raise ValueError(f"Product {item_id} not found")
            items.append(OrderItem(product.id, product.price, 1))
        
        # 3. Создаём заказ (Domain Logic)
        order = Order(
            id=str(uuid4()),
            user_id=request.user_id,
            items=items,
            total_price=sum(i.price for i in items),
            created_at=datetime.utcnow()
        )
        
        # 4. Сохраняем (Infrastructure)
        await self.order_repo.save(order)
        
        # 5. Возвращаем ответ
        return CreateOrderResponse(
            order_id=order.id,
            total_price=order.total_price
        )

3. Infrastructure слой

# infrastructure/repositories/sqlalchemy_user_repository.py
from sqlalchemy.orm import Session
from sqlalchemy import select
from typing import Optional
from domain.models import User
from domain.repositories import IUserRepository
from infrastructure.database.models import UserORM

class SQLAlchemyUserRepository(IUserRepository):
    """Реализация Repository для PostgreSQL"""
    
    def __init__(self, session: Session):
        self.session = session
    
    async def get_by_id(self, user_id: str) -> Optional[User]:
        stmt = select(UserORM).where(UserORM.id == user_id)
        user_orm = self.session.execute(stmt).scalar()
        
        if not user_orm:
            return None
        
        # Маппинг ORM → Domain
        return self._to_domain(user_orm)
    
    async def save(self, user: User) -> None:
        # Маппинг Domain → ORM
        user_orm = self._to_orm(user)
        self.session.merge(user_orm)
        self.session.commit()
    
    def _to_domain(self, orm: UserORM) -> User:
        """ORM модель → Domain модель"""
        return User(
            id=orm.id,
            email=orm.email,
            password_hash=orm.password_hash,
            status=UserStatus(orm.status),
            created_at=orm.created_at
        )
    
    def _to_orm(self, user: User) -> UserORM:
        """Domain модель → ORM модель"""
        return UserORM(
            id=user.id,
            email=user.email,
            password_hash=user.password_hash,
            status=user.status.value,
            created_at=user.created_at
        )

4. Presentation слой (FastAPI)

# presentation/api/orders.py
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from application.use_cases import CreateOrderUseCase
from infrastructure.repositories import SQLAlchemyUserRepository, SQLAlchemyOrderRepository

router = APIRouter()

class CreateOrderRequest(BaseModel):
    item_ids: list[str]

class OrderResponse(BaseModel):
    order_id: str
    total_price: float

@router.post("/orders", response_model=OrderResponse)
async def create_order(
    request: CreateOrderRequest,
    user_id: str = Depends(get_current_user_id),
    use_case: CreateOrderUseCase = Depends(get_create_order_use_case)
):
    result = await use_case.execute(
        CreateOrderRequest(user_id, request.item_ids)
    )
    return OrderResponse(
        order_id=result.order_id,
        total_price=result.total_price
    )

5. Dependency Injection

# main.py или app.py
from fastapi import FastAPI
from sqlalchemy.orm import Session
from infrastructure.database import get_db
from infrastructure.repositories import SQLAlchemyUserRepository
from application.use_cases import CreateOrderUseCase

app = FastAPI()

def get_create_order_use_case(db: Session = Depends(get_db)) -> CreateOrderUseCase:
    user_repo = SQLAlchemyUserRepository(db)
    order_repo = SQLAlchemyOrderRepository(db)
    product_service = ProductService()  # Может быть внешний сервис
    
    return CreateOrderUseCase(order_repo, user_repo, product_service)

# Все зависимости разрешаются фреймворком

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

myproject/
├── domain/                    # ← Чистая бизнес-логика
│   ├── models/
│   │   ├── user.py
│   │   ├── order.py
│   │   └── value_objects.py
│   ├── repositories/          # ← Интерфейсы (Abstract)
│   │   ├── user_repository.py
│   │   └── order_repository.py
│   └── services/              # ← Domain Services
│       └── user_service.py
│
├── application/               # ← Orchestration, Use Cases
│   └── use_cases/
│       ├── create_order.py
│       ├── get_user.py
│       └── update_profile.py
│
├── infrastructure/            # ← Технические детали
│   ├── repositories/          # ← Конкретные реализации
│   │   ├── sqlalchemy_user_repo.py
│   │   └── sqlalchemy_order_repo.py
│   ├── database/
│   │   ├── models.py          # ← ORM модели (SQLAlchemy)
│   │   └── session.py
│   └── external_services/
│       └── payment_gateway.py
│
├── presentation/              # ← API контроллеры
│   ├── api/
│   │   ├── orders.py
│   │   ├── users.py
│   │   └── products.py
│   └── schemas.py             # ← Pydantic models
│
└── main.py

Преимущества DDD

• Чистая архитектура — легко тестировать domain без БД
• Бизнес-логика защищена — domain не зависит от infrastructure
• Масштабируемость — легко добавлять новые use cases
• Сотрудничество — разработчики и бизнес говорят на одном языке
• Замена реализаций — смена БД или API провайдера не требует изменений domain

Когда DDD оправдан

• Сложная бизнес-логика (финтех, e-commerce, кор-системы)
• Большая команда (>5 разработчиков)
• Long-term проект (не MVP)
• Критичное качество и надежность

Когда DDD избыточен

• Простой CRUD (блог, каталог)
• One-man проект
• MVP с неясными требованиями

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