Архитектурные паттерны для production: DTO, разделение моделей, версионирование, миграции схем.
Паттерны и практики для надёжных, поддерживаемых проектов с Pydantic
Разделяйте модели для разных слоёв приложения:
from pydantic import BaseModel, Field, EmailStr, ConfigDict
from datetime import datetime
from typing import Optional
# === Domain Layer (бизнес-логика) ===
class User(BaseModel):
"""Доменная модель пользователя"""
id: int
email: EmailStr
username: str
is_active: bool = True
# === API Layer (внешний интерфейс) ===
class UserCreateDTO(BaseModel):
"""DTO для создания пользователя"""
email: EmailStr
username: str = Field(min_length=3, max_length=20)
password: str = Field(min_length=8)
class UserUpdateDTO(BaseModel):
"""DTO для обновления пользователя"""
email: EmailStr | None = None
username: str | None = Field(None, min_length=3, max_length=20)
password: str | None = Field(None, min_length=8)
is_active: bool | None = None
class UserResponseDTO(BaseModel):
"""DTO для ответа — без чувствительных данных"""
model_config = ConfigDict(from_attributes=True)
id: int
email: EmailStr
username: str
is_active: bool
created_at: datetime
class UserInDB(UserResponseDTO):
"""DTO для работы с БД — с password_hash"""
password_hash: str
# === Mapper ===
def to_domain(user_dto: UserCreateDTO, user_id: int) -> User:
"""Преобразование DTO в доменную модель"""
return User(
id=user_id,
email=user_dto.email,
username=user_dto.username
)
def to_response(user: User) -> UserResponseDTO:
"""Преобразование доменной модели в response DTO"""
return UserResponseDTO(
id=user.id,
email=user.email,
username=user.username,
is_active=user.is_active,
created_at=datetime.now()
)from pydantic import BaseModel, Field, model_validator
from typing import Literal
# Версия 1
class UserV1(BaseModel):
name: str
email: str
# Версия 2 — добавлено поле
class UserV2(BaseModel):
name: str
email: str
phone: str | None = None
# Версия 3 — изменение типа
class UserV3(BaseModel):
name: str
email: str
phone: str | None = None
role: Literal["user", "admin", "moderator"] = "user"
# Адаптер для миграции между версиями
class UserMigrator:
@staticmethod
def v1_to_v2(v1_data: dict) -> dict:
v2_data = v1_data.copy()
v2_data['phone'] = None # default
return v2_data
@staticmethod
def v2_to_v3(v2_data: dict) -> dict:
v3_data = v2_data.copy()
v3_data['role'] = 'user' # default
return v3_data
@classmethod
def migrate_to_latest(cls, data: dict, from_version: int) -> dict:
"""Миграция данных к последней версии"""
if from_version == 1:
data = cls.v1_to_v2(data)
from_version = 2
if from_version == 2:
data = cls.v2_to_v3(data)
from_version = 3
return data
# Использование
old_data = {"name": "Alice", "email": "alice@example.com"}
migrated = UserMigrator.migrate_to_latest(old_data, from_version=1)
user_v3 = UserV3(**migrated)from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, SecretStr, field_validator
from typing import List, Literal
class DatabaseSettings(BaseSettings):
url: SecretStr
pool_size: int = Field(ge=1, le=100, default=10)
echo: bool = False
model_config = SettingsConfigDict(
env_prefix='DB_',
env_nested_delimiter='__'
)
class CacheSettings(BaseSettings):
backend: Literal["redis", "memcached", "memory"] = "redis"
url: str = "redis://localhost:6379"
ttl: int = Field(ge=0, le=86400, default=3600) # до 24 часов
model_config = SettingsConfigDict(
env_prefix='CACHE_'
)
class AppSettings(BaseSettings):
name: str = "MyApp"
environment: Literal["development", "staging", "production"] = "development"
debug: bool = False
secret_key: SecretStr
database: DatabaseSettings
cache: CacheSettings
allowed_hosts: List[str] = ["*"]
model_config = SettingsConfigDict(
env_file='.env',
env_prefix='APP_'
)
@field_validator('allowed_hosts')
@classmethod
def validate_hosts(cls, v: List[str]) -> List[str]:
if cls.environment == 'production' and '*' in v:
raise ValueError('Wildcard hosts not allowed in production')
return v
# Загрузка настроек
settings = AppSettings(
database=DatabaseSettings(),
cache=CacheSettings()
)import pytest
from pydantic import ValidationError
def test_user_create_valid():
user = UserCreateDTO(
email="test@example.com",
username="testuser",
password="SecurePass123"
)
assert user.email == "test@example.com"
assert user.username == "testuser"
def test_user_create_invalid_email():
with pytest.raises(ValidationError) as exc_info:
UserCreateDTO(
email="invalid",
username="testuser",
password="SecurePass123"
)
assert 'email' in str(exc_info.value)
def test_user_create_short_username():
with pytest.raises(ValidationError) as exc_info:
UserCreateDTO(
email="test@example.com",
username="ab", # слишком короткий
password="SecurePass123"
)
errors = exc_info.value.errors()
assert any(e['type'] == 'string_too_short' for e in errors)
def test_user_update_partial():
update = UserUpdateDTO(username="newname")
assert update.username == "newname"
assert update.email is None
assert update.password is None
# Фикстуры для тестов
@pytest.fixture
def valid_user_data():
return {
"email": "test@example.com",
"username": "testuser",
"password": "SecurePass123"
}
def test_user_from_fixture(valid_user_data):
user = UserCreateDTO(**valid_user_data)
assert user is not Noneimport logging
from pydantic import BaseModel, ValidationError
from typing import Dict, Any
logger = logging.getLogger(__name__)
class ValidationLog(BaseModel):
"""Модель для логирования валидации"""
model_name: str
success: bool
error_count: int = 0
errors: Dict[str, str] = {}
input_data: Dict[str, Any] = {}
def log(self, level: str = 'info'):
"""Логирование записи"""
log_func = getattr(logger, level)
log_func(
f"Validation {'success' if self.success else 'failed'}: {self.model_name}",
extra={
'validation': {
'model': self.model_name,
'success': self.success,
'error_count': self.error_count,
'errors': self.errors
}
}
)
def validate_with_logging(model_class, data: dict) -> BaseModel:
"""Валидация с логированием"""
try:
instance = model_class(**data)
ValidationLog(
model_name=model_class.__name__,
success=True,
input_data=data
).log('debug')
return instance
except ValidationError as e:
ValidationLog(
model_name=model_class.__name__,
success=False,
error_count=e.error_count(),
errors={
'.'.join(str(loc) for loc in err['loc']): err['msg']
for err in e.errors()
},
input_data=data
).log('warning')
raise
# Использование
user = validate_with_logging(UserCreateDTO, {
"email": "test@example.com",
"username": "testuser",
"password": "SecurePass123"
})from pydantic import BaseModel, Field, field_validator
import re
class SafeInput(BaseModel):
"""Модель с защитой от инъекций"""
username: str = Field(min_length=3, max_length=50)
comment: str = Field(max_length=1000)
@field_validator('username')
@classmethod
def validate_username(cls, v: str) -> str:
# Только буквы, цифры, подчёркивание
if not re.match(r'^[a-zA-Z0-9_]+$', v):
raise ValueError('Username содержит недопустимые символы')
return v
@field_validator('comment')
@classmethod
def validate_comment(cls, v: str) -> str:
# Проверка на SQL-инъекции
sql_patterns = [
r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|UNION|ALTER)\b)",
r"(--)",
r"(;)",
r"(\b(OR|AND)\b\s+\d+\s*=\s*\d+)"
]
for pattern in sql_patterns:
if re.search(pattern, v, re.IGNORECASE):
raise ValueError('Comment содержит подозрительные паттерны')
return v
# Использование
safe = SafeInput(username="user123", comment="Hello!")
unsafe = SafeInput(username="user; DROP TABLE", comment="...") # ValidationErrorfrom pydantic import BaseModel, Field, ConfigDict
from typing import Optional
class UserCreate(BaseModel):
"""
Модель для создания нового пользователя.
Используется в endpoint POST /api/v1/users
Attributes:
email: Email пользователя в формате user@domain.com
username: Имя пользователя (3-20 символов, только латиница)
password: Пароль (минимум 8 символов)
Example:
>>> user = UserCreate(
... email="alice@example.com",
... username="alice_dev",
... password="SecurePass123"
... )
"""
model_config = ConfigDict(
json_schema_extra={
"examples": [
{
"email": "alice@example.com",
"username": "alice_dev",
"password": "SecurePass123"
}
]
}
)
email: str = Field(
...,
description="Email пользователя",
json_schema_extra={"example": "user@example.com"}
)
username: str = Field(
...,
min_length=3,
max_length=20,
description="Имя пользователя",
pattern=r'^[a-zA-Z0-9_]+$',
json_schema_extra={"example": "user_dev"}
)
password: str = Field(
...,
min_length=8,
description="Пароль пользователя (минимум 8 символов)",
json_schema_extra={"example": "SecurePass123"}
)Создайте полную систему с DTO, валидацией и настройками:
from pydantic import BaseModel, Field, EmailStr, SecretStr, ConfigDict
from pydantic_settings import BaseSettings, SettingsConfigDict
from datetime import datetime
from typing import Optional, List, Literal
from enum import Enum
# === Settings ===
class Environment(str, Enum):
development = "development"
production = "production"
class Settings(BaseSettings):
environment: Environment = Environment.development
secret_key: SecretStr
database_url: SecretStr
model_config = SettingsConfigDict(
env_file='.env',
env_prefix='APP_'
)
# === DTOs ===
class UserCreateDTO(BaseModel):
# email, username, password
pass
class UserUpdateDTO(BaseModel):
# optional fields
pass
class UserResponseDTO(BaseModel):
model_config = ConfigDict(from_attributes=True)
# id, email, username, created_at
pass
# === Domain ===
class User(BaseModel):
# id, email, username, is_active, created_at
pass
# === Mapper ===
class UserMapper:
@staticmethod
def dto_to_domain(dto: UserCreateDTO, user_id: int) -> User:
pass
@staticmethod
def domain_to_response(domain: User) -> UserResponseDTO:
pass
# === Service ===
class UserService:
def __init__(self, settings: Settings):
self.settings = settings
def create_user(self, dto: UserCreateDTO) -> User:
pass
def get_user(self, user_id: int) -> UserResponseDTO:
pass
def update_user(self, user_id: int, dto: UserUpdateDTO) -> UserResponseDTO:
passВы прошли полный курс по Pydantic v2! Теперь вы умеете:
Следующие шаги: