Data Transfer Objects
Объекты для передачи данных между слоями
DTO — объект, который переносит данные между процессами или слоями приложения, не содержащий бизнес-логики.
# ❌ Возвращаем доменную модель напрямую
class UserController:
def get_user(self, user_id: int):
user = user_service.get_user(user_id)
return user # Доменная модель с внутренними деталями
# Проблемы:
# - Раскрываем внутреннюю структуру
# - Сложно изменить доменную модель
# - Нет контроля над данными
# - Могут быть циклические зависимости
# ✅ Используем DTO
class UserDTO:
def __init__(self, id: int, name: str, email: str):
self.id = id
self.name = name
self.email = email
def to_dict(self):
return {
"id": self.id,
"name": self.name,
"email": self.email
}
class UserController:
def get_user(self, user_id: int):
user = user_service.get_user(user_id)
return UserDTO(user.id, user.name, user.email)
# Преимущества:
# - Контроль над данными
# - Независимость от доменной модели
# - Легко изменить представление
from dataclasses import dataclass
from typing import Optional
@dataclass
class CreateUserRequest:
name: str
email: str
password: str
def validate(self):
if not self.name:
raise ValueError("Name is required")
if "@" not in self.email:
raise ValueError("Invalid email")
if len(self.password) < 8:
raise ValueError("Password must be at least 8 characters")
@dataclass
class UpdateUserRequest:
name: Optional[str] = None
email: Optional[str] = None
# Использование
@app.post("/users")
def create_user(request: CreateUserRequest):
request.validate()
user = user_service.create_user(request.name, request.email, request.password)
return UserDTO.from_entity(user)
@dataclass
class UserDTO:
id: int
name: str
email: str
created_at: str
@staticmethod
def from_entity(user: User) -> 'UserDTO':
return UserDTO(
id=user.id,
name=user.name,
email=user.email,
created_at=user.created_at.isoformat()
)
def to_dict(self):
return {
"id": self.id,
"name": self.name,
"email": self.email,
"created_at": self.created_at
}
@dataclass
class UserListDTO:
users: List[UserDTO]
total: int
page: int
page_size: int
def to_dict(self):
return {
"users": [user.to_dict() for user in self.users],
"total": self.total,
"page": self.page,
"page_size": self.page_size
}
# DTO для передачи данных между слоями приложения
@dataclass
class UserCreateCommand:
name: str
email: str
password: str
# Application Layer → Domain Layer
class CreateUserUseCase:
def execute(self, command: UserCreateCommand) -> UserDTO:
user = User.create(command.name, command.email, command.password)
user_repository.save(user)
return UserDTO.from_entity(user)
# Presentation Layer → Application Layer
class UserController:
def create_user(self, request: CreateUserRequest):
command = UserCreateCommand(
name=request.name,
email=request.email,
password=request.password
)
return create_user_use_case.execute(command)
from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional, List
from datetime import datetime
class UserCreateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
email: EmailStr
password: str = Field(..., min_length=8)
@validator('name')
def validate_name(cls, v):
if not v.strip():
raise ValueError('Name cannot be empty')
return v.strip()
class UserUpdateRequest(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=100)
email: Optional[EmailStr] = None
class UserResponse(BaseModel):
id: int
name: str
email: str
created_at: datetime
class Config:
orm_mode = True # Автоматическое преобразование из ORM модели
# Использование в FastAPI
@app.post("/users", response_model=UserResponse)
def create_user(user: UserCreateRequest):
user_entity = user_service.create_user(user.name, user.email, user.password)
return UserResponse.from_orm(user_entity)
from pydantic import BaseModel, validator
from typing import List
class OrderCreateRequest(BaseModel):
user_id: int
items: List[OrderItemRequest]
total: float
@validator('items')
def validate_items(cls, v):
if not v:
raise ValueError('Order must have at least one item')
return v
@validator('total')
def validate_total(cls, v, values):
if 'items' in values:
calculated_total = sum(item.price * item.quantity for item in values['items'])
if abs(v - calculated_total) > 0.01:
raise ValueError('Total does not match items')
return v
class OrderItemRequest(BaseModel):
product_id: int
quantity: int
price: float
@validator('quantity')
def validate_quantity(cls, v):
if v <= 0:
raise ValueError('Quantity must be positive')
return v
class UserDTO:
def __init__(self, id: int, name: str, email: str):
self.id = id
self.name = name
self.email = email
@staticmethod
def from_entity(user: User) -> 'UserDTO':
return UserDTO(
id=user.id,
name=user.name,
email=user.email
)
def to_entity(self) -> User:
return User(
id=self.id,
name=self.name,
email=self.email
)
# Использование
user = user_repository.find_by_id(1)
user_dto = UserDTO.from_entity(user)
user_entity = user_dto.to_entity()
from dataclasses import dataclass, asdict
@dataclass
class UserDTO:
id: int
name: str
email: str
@classmethod
def from_entity(cls, user: User):
return cls(
id=user.id,
name=user.name,
email=user.email
)
def to_dict(self):
return asdict(self)
# Или через маппер
class UserMapper:
@staticmethod
def to_dto(user: User) -> UserDTO:
return UserDTO(
id=user.id,
name=user.name,
email=user.email
)
@staticmethod
def to_entity(dto: UserDTO) -> User:
return User(
id=dto.id,
name=dto.name,
email=dto.email
)
class UserDetailDTO:
def __init__(self, user: User, orders: List[Order]):
self.id = user.id
self.name = user.name
self.email = user.email
self.orders = [OrderDTO.from_entity(order) for order in orders]
self.total_spent = sum(order.total for order in orders)
self.order_count = len(orders)
@staticmethod
def from_entities(user: User, orders: List[Order]) -> 'UserDetailDTO':
return UserDetailDTO(user, orders)
def to_dict(self):
return {
"id": self.id,
"name": self.name,
"email": self.email,
"orders": [order.to_dict() for order in self.orders],
"total_spent": self.total_spent,
"order_count": self.order_count
}
# Использование
user = user_repository.find_by_id(1)
orders = order_repository.find_by_user_id(user.id)
user_dto = UserDetailDTO.from_entities(user, orders)
| Сценарий | Рекомендация |
|---|---|
| API (REST, GraphQL) | ✅ Обязательно |
| Микросервисы | ✅ Обязательно |
| Разные слои приложения | ✅ Рекомендуется |
| Простое CRUD приложение | ⚠️ Может быть избыточно |
| Монолитное приложение | ⚠️ Зависит от сложности |
"DTO — это не про код. Это про границы.
DTO определяет контракт между слоями и защищает домен от изменений во внешнем мире."