Data Validation

Валидация данных критически важна для целостности приложения. В этой теме вы изучите интеграцию Pydantic с SQLAlchemy, валидацию на уровне модели и лучшие практики.

#Валидация на уровне БД

#Constraints

from sqlalchemy import CheckConstraint, UniqueConstraint

class User(Base):
    __tablename__ = 'users'
    __table_args__ = (
        # Unique constraint
        UniqueConstraint('email', name='uq_users_email'),
        
        # Check constraint
        CheckConstraint('age >= 0', name='check_age_positive'),
        
        # Composite unique
        UniqueConstraint('country', 'city', name='uq_location'),
    )
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    age: Mapped[int] = mapped_column(Integer, nullable=True)
    country: Mapped[str] = mapped_column(String(100))
    city: Mapped[str] = mapped_column(String(100))

#Nullable и default

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    
    # NOT NULL
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    
    # NULL allowed
    phone: Mapped[str | None] = mapped_column(String(20), nullable=True)
    
    # Default значение
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    
    # Server default (вычисляется в БД)
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        server_default=func.now()
    )
    
    # Server default с выражением
    status: Mapped[str] = mapped_column(
        String(50),
        server_default='pending'
    )

#Enum типы

from enum import Enum as PyEnum
from sqlalchemy import Enum

class UserRole(PyEnum):
    ADMIN = 'admin'
    USER = 'user'
    GUEST = 'guest'


class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    
    # SQLAlchemy Enum
    role: Mapped[UserRole] = mapped_column(
        Enum(UserRole, name='user_role'),
        default=UserRole.USER
    )
    
    # Native PostgreSQL ENUM (рекомендуется)
    status: Mapped[str] = mapped_column(
        Enum('active', 'inactive', 'banned', name='user_status'),
        default='active'
    )

#Длина строк

class User(Base):
    __tablename__ = 'users'
    
    # Ограничение длины
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    
    # TEXT без ограничения (для длинного текста)
    bio: Mapped[str | None] = mapped_column(Text, nullable=True)
    
    # VARCHAR с проверкой длины через CheckConstraint
    __table_args__ = (
        CheckConstraint('length(email) >= 5', name='check_email_length'),
        CheckConstraint('length(name) >= 2', name='check_name_length'),
    )

#Валидация на уровне ORM

#ValidatedSession

from sqlalchemy.orm import validates

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    age: Mapped[int | None] = mapped_column(Integer, nullable=True)
    
    @validates('email')
    def validate_email(self, key, email: str) -> str:
        """Валидация email."""
        if not email:
            raise ValueError('Email is required')
        
        if '@' not in email:
            raise ValueError('Email must contain @')
        
        if len(email) > 255:
            raise ValueError('Email too long')
        
        return email.lower()
    
    @validates('name')
    def validate_name(self, key, name: str) -> str:
        """Валидация имени."""
        if not name or len(name.strip()) < 2:
            raise ValueError('Name must be at least 2 characters')
        
        return name.strip()
    
    @validates('age')
    def validate_age(self, key, age: int | None) -> int | None:
        """Валидация возраста."""
        if age is not None:
            if age < 0 or age > 150:
                raise ValueError('Age must be between 0 and 150')
        
        return age

#Валидация нескольких полей

from sqlalchemy.orm import validates
from sqlalchemy.exc import ValidationError

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    password: Mapped[str] = mapped_column(String(255))
    password_confirm: Mapped[str | None] = mapped_column(String(255))
    
    @validates('password_confirm')
    def validate_passwords_match(self, key, password_confirm: str) -> str:
        """Проверка совпадения паролей."""
        if password_confirm and password_confirm != self.password:
            raise ValidationError('Passwords do not match')
        return password_confirm

#События валидации

from sqlalchemy import event
from sqlalchemy.orm import validates
from sqlalchemy.exc import ValidationError

@event.listens_for(User, 'before_insert')
def validate_before_insert(mapper, connection, target):
    """Валидация перед вставкой."""
    if target.email and not is_valid_email(target.email):
        raise ValidationError('Invalid email format')


@event.listens_for(User, 'before_update')
def validate_before_update(mapper, connection, target):
    """Валидация перед обновлением."""
    if target.age and target.age < 0:
        raise ValidationError('Age cannot be negative')


def is_valid_email(email: str) -> bool:
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

#Интеграция с Pydantic

#Базовая интеграция

from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime


class UserBase(BaseModel):
    """Базовая схема пользователя."""
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    age: int | None = Field(None, ge=0, le=150)
    
    @validator('email')
    def email_lowercase(cls, v):
        return v.lower()


class UserCreate(UserBase):
    """Схема для создания пользователя."""
    password: str = Field(..., min_length=8)
    password_confirm: str
    
    @validator('password_confirm')
    def passwords_match(cls, v, values):
        if 'password' in values and v != values['password']:
            raise ValueError('Passwords do not match')
        return v


class UserResponse(UserBase):
    """Схема ответа."""
    id: int
    created_at: datetime
    
    class Config:
        from_attributes = True  # SQLAlchemy совместимость

#Конвертация между SQLAlchemy и Pydantic

from sqlalchemy.orm import Session
from pydantic import BaseModel


class UserCreate(BaseModel):
    email: str
    name: str
    age: int | None = None
    
    def to_orm(self) -> User:
        """Конвертация Pydantic → SQLAlchemy."""
        return User(
            email=self.email,
            name=self.name,
            age=self.age
        )


class UserResponse(BaseModel):
    id: int
    email: str
    name: str
    age: int | None
    
    class Config:
        from_attributes = True
    
    @classmethod
    def from_orm(cls, user: User) -> 'UserResponse':
        """Конвертация SQLAlchemy → Pydantic."""
        return cls(
            id=user.id,
            email=user.email,
            name=user.name,
            age=user.age
        )


# Использование
def create_user(db: Session, user_data: UserCreate) -> UserResponse:
    user = user_data.to_orm()
    db.add(user)
    db.commit()
    db.refresh(user)
    return UserResponse.from_orm(user)

#Pydantic v2 с SQLAlchemy

from pydantic import BaseModel, EmailStr, ConfigDict
from datetime import datetime


class UserBase(BaseModel):
    """Pydantic v2 стиль."""
    email: EmailStr
    name: str
    age: int | None = None
    
    model_config = ConfigDict(
        from_attributes=True,  # Вместо orm_mode
        str_strip_whitespace=True,  # Автоматическая trim строк
    )


class UserCreate(UserBase):
    password: str
    
    @field_validator('password')
    @classmethod
    def password_strength(cls, v):
        if len(v) < 8:
            raise ValueError('Password too short')
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain uppercase')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        return v


class UserResponse(UserBase):
    id: int
    created_at: datetime

#Автоматическая генерация схем

from pydantic import BaseModel
from pydantic_sqlalchemy import sqlalchemy_to_pydantic
from myapp.models import User


class UserCreate(sqlalchemy_to_pydantic(User, exclude=['id', 'created_at'])):
    """Автоматически сгенерированная схема."""
    pass


class UserUpdate(sqlalchemy_to_pydantic(User, exclude=['id'], optional=True)):
    """Все поля опциональны для update."""
    pass


class UserResponse(sqlalchemy_to_pydantic(User)):
    """Полная схема ответа."""
    pass

#Валидация в API (FastAPI)

#Endpoint валидация

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2)
    age: int | None = Field(None, ge=0, le=150)


class UserResponse(BaseModel):
    id: int
    email: str
    name: str
    
    class Config:
        from_attributes = True


@app.post('/users', response_model=UserResponse)
def create_user(user_data: UserCreate, db: Session = Depends(get_db)):
    # Проверка на дубликат email
    existing = db.query(User).filter(User.email == user_data.email).first()
    if existing:
        raise HTTPException(status_code=400, detail='Email already registered')
    
    # Создание пользователя
    user = User(**user_data.model_dump())
    db.add(user)
    db.commit()
    db.refresh(user)
    
    return user


@app.get('/users/{user_id}', response_model=UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail='User not found')
    return user

#Валидация query параметров

from fastapi import Query
from typing import Optional


@app.get('/users')
def list_users(
    page: int = Query(1, ge=1, description='Page number'),
    per_page: int = Query(10, ge=1, le=100, description='Items per page'),
    search: Optional[str] = Query(None, min_length=2, max_length=100),
    db: Session = Depends(get_db)
):
    query = db.query(User)
    
    if search:
        query = query.filter(User.name.ilike(f'%{search}%'))
    
    offset = (page - 1) * per_page
    users = query.offset(offset).limit(per_page).all()
    
    return users

#Кастомные типы с валидацией

#TypeDecorator

from sqlalchemy.types import TypeDecorator, String
import re


class EmailString(TypeDecorator):
    """Тип для email с автоматической валидацией."""
    
    impl = String
    cache_ok = True
    
    def process_bind_param(self, value, dialect):
        if value is not None:
            value = value.lower().strip()
            
            # Валидация email
            pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            if not re.match(pattern, value):
                raise ValueError(f'Invalid email: {value}')
        
        return value
    
    def process_result_value(self, value, dialect):
        return value


class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(EmailString(255), nullable=False)

#JSON с валидацией

from sqlalchemy.types import TypeDecorator, JSON
from pydantic import BaseModel, ValidationError


class PydanticJSON(TypeDecorator):
    """JSON тип с Pydantic валидацией."""
    
    impl = JSON
    cache_ok = True
    
    def __init__(self, pydantic_model: type[BaseModel]):
        super().__init__()
        self.pydantic_model = pydantic_model
    
    def process_bind_param(self, value, dialect):
        if value is not None:
            # Валидация через Pydantic
            try:
                validated = self.pydantic_model.model_validate(value)
                return validated.model_dump()
            except ValidationError as e:
                raise ValueError(f'Invalid JSON: {e}')
        return value
    
    def process_result_value(self, value, dialect):
        if value is not None:
            return self.pydantic_model.model_validate(value)
        return value


# Использование
class MetadataSchema(BaseModel):
    tags: list[str] = []
    priority: int = Field(ge=1, le=10)
    description: str | None = None


class Task(Base):
    __tablename__ = 'tasks'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    metadata: Mapped[MetadataSchema | None] = mapped_column(
        PydanticJSON(MetadataSchema),
        nullable=True
    )

#Best Practices

#✅ Делайте

# Используйте constraints на уровне БД
__table_args__ = (
    UniqueConstraint('email', name='uq_users_email'),
    CheckConstraint('age >= 0', name='check_age_positive'),
)

# Валидируйте в Pydantic для API
class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2)

# Используйте server_default для timestamps
created_at: Mapped[datetime] = mapped_column(
    server_default=func.now()
)

# Валидируйте email через TypeDecorator
email: Mapped[str] = mapped_column(EmailString(255))

#❌ Не делайте

# Не полагайтесь только на валидацию в Python
# Всегда используйте constraints в БД

# Не храните пароли в открытом виде
password: Mapped[str]  # ПЛОХО
# Используйте хеширование

# Не используйте float для денег
price: Mapped[float]  # ПЛОХО
price: Mapped[Decimal] = mapped_column(Numeric(10, 2))  # ХОРОШО

# Не забывайте про индексы для unique полей
email: Mapped[str] = mapped_column(String(255), unique=True)  # Автоматический индекс

Data Validation

Валидация данных критически важна для целостности приложения. В этой теме вы изучите интеграцию Pydantic с SQLAlchemy, валидацию на уровне модели и лучшие практики.

#Валидация на уровне БД

#Constraints

from sqlalchemy import CheckConstraint, UniqueConstraint

class User(Base):
    __tablename__ = 'users'
    __table_args__ = (
        # Unique constraint
        UniqueConstraint('email', name='uq_users_email'),
        
        # Check constraint
        CheckConstraint('age >= 0', name='check_age_positive'),
        
        # Composite unique
        UniqueConstraint('country', 'city', name='uq_location'),
    )
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    age: Mapped[int] = mapped_column(Integer, nullable=True)
    country: Mapped[str] = mapped_column(String(100))
    city: Mapped[str] = mapped_column(String(100))

#Nullable и default

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    
    # NOT NULL
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    
    # NULL allowed
    phone: Mapped[str | None] = mapped_column(String(20), nullable=True)
    
    # Default значение
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    
    # Server default (вычисляется в БД)
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        server_default=func.now()
    )
    
    # Server default с выражением
    status: Mapped[str] = mapped_column(
        String(50),
        server_default='pending'
    )

#Enum типы

from enum import Enum as PyEnum
from sqlalchemy import Enum

class UserRole(PyEnum):
    ADMIN = 'admin'
    USER = 'user'
    GUEST = 'guest'


class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    
    # SQLAlchemy Enum
    role: Mapped[UserRole] = mapped_column(
        Enum(UserRole, name='user_role'),
        default=UserRole.USER
    )
    
    # Native PostgreSQL ENUM (рекомендуется)
    status: Mapped[str] = mapped_column(
        Enum('active', 'inactive', 'banned', name='user_status'),
        default='active'
    )

#Длина строк

class User(Base):
    __tablename__ = 'users'
    
    # Ограничение длины
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    
    # TEXT без ограничения (для длинного текста)
    bio: Mapped[str | None] = mapped_column(Text, nullable=True)
    
    # VARCHAR с проверкой длины через CheckConstraint
    __table_args__ = (
        CheckConstraint('length(email) >= 5', name='check_email_length'),
        CheckConstraint('length(name) >= 2', name='check_name_length'),
    )

#Валидация на уровне ORM

#ValidatedSession

from sqlalchemy.orm import validates

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    age: Mapped[int | None] = mapped_column(Integer, nullable=True)
    
    @validates('email')
    def validate_email(self, key, email: str) -> str:
        """Валидация email."""
        if not email:
            raise ValueError('Email is required')
        
        if '@' not in email:
            raise ValueError('Email must contain @')
        
        if len(email) > 255:
            raise ValueError('Email too long')
        
        return email.lower()
    
    @validates('name')
    def validate_name(self, key, name: str) -> str:
        """Валидация имени."""
        if not name or len(name.strip()) < 2:
            raise ValueError('Name must be at least 2 characters')
        
        return name.strip()
    
    @validates('age')
    def validate_age(self, key, age: int | None) -> int | None:
        """Валидация возраста."""
        if age is not None:
            if age < 0 or age > 150:
                raise ValueError('Age must be between 0 and 150')
        
        return age

#Валидация нескольких полей

from sqlalchemy.orm import validates
from sqlalchemy.exc import ValidationError

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    password: Mapped[str] = mapped_column(String(255))
    password_confirm: Mapped[str | None] = mapped_column(String(255))
    
    @validates('password_confirm')
    def validate_passwords_match(self, key, password_confirm: str) -> str:
        """Проверка совпадения паролей."""
        if password_confirm and password_confirm != self.password:
            raise ValidationError('Passwords do not match')
        return password_confirm

#События валидации

from sqlalchemy import event
from sqlalchemy.orm import validates
from sqlalchemy.exc import ValidationError

@event.listens_for(User, 'before_insert')
def validate_before_insert(mapper, connection, target):
    """Валидация перед вставкой."""
    if target.email and not is_valid_email(target.email):
        raise ValidationError('Invalid email format')


@event.listens_for(User, 'before_update')
def validate_before_update(mapper, connection, target):
    """Валидация перед обновлением."""
    if target.age and target.age < 0:
        raise ValidationError('Age cannot be negative')


def is_valid_email(email: str) -> bool:
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

#Интеграция с Pydantic

#Базовая интеграция

from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime


class UserBase(BaseModel):
    """Базовая схема пользователя."""
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    age: int | None = Field(None, ge=0, le=150)
    
    @validator('email')
    def email_lowercase(cls, v):
        return v.lower()


class UserCreate(UserBase):
    """Схема для создания пользователя."""
    password: str = Field(..., min_length=8)
    password_confirm: str
    
    @validator('password_confirm')
    def passwords_match(cls, v, values):
        if 'password' in values and v != values['password']:
            raise ValueError('Passwords do not match')
        return v


class UserResponse(UserBase):
    """Схема ответа."""
    id: int
    created_at: datetime
    
    class Config:
        from_attributes = True  # SQLAlchemy совместимость

#Конвертация между SQLAlchemy и Pydantic

from sqlalchemy.orm import Session
from pydantic import BaseModel


class UserCreate(BaseModel):
    email: str
    name: str
    age: int | None = None
    
    def to_orm(self) -> User:
        """Конвертация Pydantic → SQLAlchemy."""
        return User(
            email=self.email,
            name=self.name,
            age=self.age
        )


class UserResponse(BaseModel):
    id: int
    email: str
    name: str
    age: int | None
    
    class Config:
        from_attributes = True
    
    @classmethod
    def from_orm(cls, user: User) -> 'UserResponse':
        """Конвертация SQLAlchemy → Pydantic."""
        return cls(
            id=user.id,
            email=user.email,
            name=user.name,
            age=user.age
        )


# Использование
def create_user(db: Session, user_data: UserCreate) -> UserResponse:
    user = user_data.to_orm()
    db.add(user)
    db.commit()
    db.refresh(user)
    return UserResponse.from_orm(user)

#Pydantic v2 с SQLAlchemy

from pydantic import BaseModel, EmailStr, ConfigDict
from datetime import datetime


class UserBase(BaseModel):
    """Pydantic v2 стиль."""
    email: EmailStr
    name: str
    age: int | None = None
    
    model_config = ConfigDict(
        from_attributes=True,  # Вместо orm_mode
        str_strip_whitespace=True,  # Автоматическая trim строк
    )


class UserCreate(UserBase):
    password: str
    
    @field_validator('password')
    @classmethod
    def password_strength(cls, v):
        if len(v) < 8:
            raise ValueError('Password too short')
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain uppercase')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        return v


class UserResponse(UserBase):
    id: int
    created_at: datetime

#Автоматическая генерация схем

from pydantic import BaseModel
from pydantic_sqlalchemy import sqlalchemy_to_pydantic
from myapp.models import User


class UserCreate(sqlalchemy_to_pydantic(User, exclude=['id', 'created_at'])):
    """Автоматически сгенерированная схема."""
    pass


class UserUpdate(sqlalchemy_to_pydantic(User, exclude=['id'], optional=True)):
    """Все поля опциональны для update."""
    pass


class UserResponse(sqlalchemy_to_pydantic(User)):
    """Полная схема ответа."""
    pass

#Валидация в API (FastAPI)

#Endpoint валидация

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2)
    age: int | None = Field(None, ge=0, le=150)


class UserResponse(BaseModel):
    id: int
    email: str
    name: str
    
    class Config:
        from_attributes = True


@app.post('/users', response_model=UserResponse)
def create_user(user_data: UserCreate, db: Session = Depends(get_db)):
    # Проверка на дубликат email
    existing = db.query(User).filter(User.email == user_data.email).first()
    if existing:
        raise HTTPException(status_code=400, detail='Email already registered')
    
    # Создание пользователя
    user = User(**user_data.model_dump())
    db.add(user)
    db.commit()
    db.refresh(user)
    
    return user


@app.get('/users/{user_id}', response_model=UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail='User not found')
    return user

#Валидация query параметров

from fastapi import Query
from typing import Optional


@app.get('/users')
def list_users(
    page: int = Query(1, ge=1, description='Page number'),
    per_page: int = Query(10, ge=1, le=100, description='Items per page'),
    search: Optional[str] = Query(None, min_length=2, max_length=100),
    db: Session = Depends(get_db)
):
    query = db.query(User)
    
    if search:
        query = query.filter(User.name.ilike(f'%{search}%'))
    
    offset = (page - 1) * per_page
    users = query.offset(offset).limit(per_page).all()
    
    return users

#Кастомные типы с валидацией

#TypeDecorator

from sqlalchemy.types import TypeDecorator, String
import re


class EmailString(TypeDecorator):
    """Тип для email с автоматической валидацией."""
    
    impl = String
    cache_ok = True
    
    def process_bind_param(self, value, dialect):
        if value is not None:
            value = value.lower().strip()
            
            # Валидация email
            pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            if not re.match(pattern, value):
                raise ValueError(f'Invalid email: {value}')
        
        return value
    
    def process_result_value(self, value, dialect):
        return value


class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(EmailString(255), nullable=False)

#JSON с валидацией

from sqlalchemy.types import TypeDecorator, JSON
from pydantic import BaseModel, ValidationError


class PydanticJSON(TypeDecorator):
    """JSON тип с Pydantic валидацией."""
    
    impl = JSON
    cache_ok = True
    
    def __init__(self, pydantic_model: type[BaseModel]):
        super().__init__()
        self.pydantic_model = pydantic_model
    
    def process_bind_param(self, value, dialect):
        if value is not None:
            # Валидация через Pydantic
            try:
                validated = self.pydantic_model.model_validate(value)
                return validated.model_dump()
            except ValidationError as e:
                raise ValueError(f'Invalid JSON: {e}')
        return value
    
    def process_result_value(self, value, dialect):
        if value is not None:
            return self.pydantic_model.model_validate(value)
        return value


# Использование
class MetadataSchema(BaseModel):
    tags: list[str] = []
    priority: int = Field(ge=1, le=10)
    description: str | None = None


class Task(Base):
    __tablename__ = 'tasks'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    metadata: Mapped[MetadataSchema | None] = mapped_column(
        PydanticJSON(MetadataSchema),
        nullable=True
    )

#Best Practices

#✅ Делайте

# Используйте constraints на уровне БД
__table_args__ = (
    UniqueConstraint('email', name='uq_users_email'),
    CheckConstraint('age >= 0', name='check_age_positive'),
)

# Валидируйте в Pydantic для API
class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2)

# Используйте server_default для timestamps
created_at: Mapped[datetime] = mapped_column(
    server_default=func.now()
)

# Валидируйте email через TypeDecorator
email: Mapped[str] = mapped_column(EmailString(255))

#❌ Не делайте

# Не полагайтесь только на валидацию в Python
# Всегда используйте constraints в БД

# Не храните пароли в открытом виде
password: Mapped[str]  # ПЛОХО
# Используйте хеширование

# Не используйте float для денег
price: Mapped[float]  # ПЛОХО
price: Mapped[Decimal] = mapped_column(Numeric(10, 2))  # ХОРОШО

# Не забывайте про индексы для unique полей
email: Mapped[str] = mapped_column(String(255), unique=True)  # Автоматический индекс