Pydantic интеграция, валидация на уровне модели, constraints
Валидация данных критически важна для целостности приложения. В этой теме вы изучите интеграцию Pydantic с SQLAlchemy, валидацию на уровне модели и лучшие практики.
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))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'
)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'),
)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 agefrom 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_confirmfrom 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))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 совместимость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)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: datetimefrom 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)):
"""Полная схема ответа."""
passfrom 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 userfrom 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 usersfrom 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)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
)# Используйте 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) # Автоматический индексВ следующей теме вы изучите Events и Hooks — аудит изменений, триггеры на уровне ORM, кастомная логика через event listeners.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.