Создание Pydantic-моделей на основе SQLAlchemy, режим from_attributes, DTO-паттерн.
Pydantic и SQLAlchemy — мощная комбинация. Изучим, как связать ORM-модели с Pydantic-валидацией
SQLAlchemy ORM-модели не совместимы с Pydantic напрямую. Нужно использовать from_attributes:
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase
from pydantic import BaseModel, ConfigDict
# SQLAlchemy модель
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
# Pydantic модель
class UserSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
email: str
# Использование
user_orm = User(id=1, name="Alice", email="alice@example.com")
user_schema = UserSchema.model_validate(user_orm)
print(user_schema.name) # "Alice"from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr
# SQLAlchemy ORM модель
class Base(DeclarativeBase):
pass
class UserORM(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(unique=True)
email: Mapped[str]
password_hash: Mapped[str]
created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
is_active: Mapped[bool] = mapped_column(default=True)
# Pydantic модели для разных операций
class UserCreate(BaseModel):
"""Для создания пользователя"""
username: str
email: EmailStr
password: str
class UserUpdate(BaseModel):
"""Для обновления — все поля optional"""
username: str | None = None
email: EmailStr | None = None
password: str | None = None
is_active: bool | None = None
class UserResponse(BaseModel):
"""Для ответа — без password_hash"""
model_config = ConfigDict(from_attributes=True)
id: int
username: str
email: str
created_at: datetime
is_active: bool
class UserInDB(UserResponse):
"""Для внутреннего использования — с password_hash"""
password_hash: strfrom typing import Optional, List
from pydantic import BaseModel, ConfigDict
# SQLAlchemy модели
class PostORM(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
content: Mapped[str]
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped["UserORM"] = relationship("UserORM")
comments: Mapped[List["CommentORM"]] = relationship("CommentORM")
# Pydantic DTO
class PostBase(BaseModel):
title: str
content: str
class PostCreate(PostBase):
author_id: int
class PostResponse(PostBase):
model_config = ConfigDict(from_attributes=True)
id: int
author_id: int
created_at: datetime
updated_at: datetime | None
class PostWithAuthor(PostResponse):
author: "UserResponse"
class PostWithComments(PostResponse):
comments: list["CommentResponse"]from sqlalchemy.orm import Session
def create_user(db: Session, user: UserCreate) -> UserORM:
"""Создание пользователя с валидацией"""
# Проверка на дубликат email
existing = db.query(UserORM).filter(UserORM.email == user.email).first()
if existing:
raise ValueError("Email already registered")
# Проверка на дубликат username
existing = db.query(UserORM).filter(UserORM.username == user.username).first()
if existing:
raise ValueError("Username already taken")
# Создание
hashed_password = hash_password(user.password)
db_user = UserORM(
username=user.username,
email=user.email,
password_hash=hashed_password
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
# Использование
user_create = UserCreate(username="alice", email="alice@example.com", password="secret123")
user_orm = create_user(db, user_create)
user_response = UserResponse.model_validate(user_orm)def update_user(db: Session, user_id: int, user_update: UserUpdate) -> UserORM:
"""Обновление пользователя с валидацией"""
db_user = db.query(UserORM).filter(UserORM.id == user_id).first()
if not db_user:
raise ValueError("User not found")
# Обновление только переданных полей
update_data = user_update.model_dump(exclude_unset=True)
# Если пароль обновляется — хэшируем
if "password" in update_data:
update_data["password_hash"] = hash_password(update_data.pop("password"))
for key, value in update_data.items():
setattr(db_user, key, value)
db.commit()
db.refresh(db_user)
return db_user
# Использование
user_update = UserUpdate(email="new@example.com", is_active=False)
updated_user = update_user(db, 1, user_update)from pydantic import BaseModel, ConfigDict
from typing import List, Optional
class CommentResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
post_id: int
content: str
created_at: datetime
class PostResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
title: str
content: str
author_id: int
comments: List[CommentResponse] = []
# Исключаем comments из рекурсивной сериализации
class Config:
from_attributes = True
# Загрузка с relationship
post_orm = db.query(PostORM).options(
joinedload(PostORM.comments)
).first()
post_response = PostResponse.model_validate(post_orm)
print(post_response.comments) # [CommentResponse(...), ...]from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum
class UserStatus(str, Enum):
active = "active"
inactive = "inactive"
all = "all"
class UserFilter(BaseModel):
"""Фильтр для поиска пользователей"""
status: UserStatus = UserStatus.active
min_id: int | None = Field(None, gt=0)
max_id: int | None = Field(None, gt=0)
email_domain: str | None = None
limit: int = Field(ge=1, le=100, default=10)
offset: int = Field(ge=0, default=0)
def filter_users(db: Session, filter: UserFilter) -> List[UserORM]:
"""Применение фильтра к query"""
query = db.query(UserORM)
if filter.status != UserStatus.all:
query = query.filter(UserORM.is_active == (filter.status == UserStatus.active))
if filter.min_id:
query = query.filter(UserORM.id >= filter.min_id)
if filter.max_id:
query = query.filter(UserORM.id <= filter.max_id)
if filter.email_domain:
query = query.filter(UserORM.email.like(f"%@{filter.email_domain}"))
return query.offset(filter.offset).limit(filter.limit).all()from pydantic import BaseModel, ConfigDict, computed_field
class UserSummary(BaseModel):
"""Лёгкая версия для списков"""
model_config = ConfigDict(from_attributes=True)
id: int
username: str
@computed_field
@property
def display_name(self) -> str:
return f"@{self.username}"
class UserDetail(BaseModel):
"""Полная версия для детального просмотра"""
model_config = ConfigDict(from_attributes=True)
id: int
username: str
email: str
created_at: datetime
is_active: bool
@computed_field
@property
def display_name(self) -> str:
return f"@{self.username}"
@computed_field
@property
def account_age_days(self) -> int:
return (datetime.now() - self.created_at).daysСоздайте полный набор моделей для блога:
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase, Mapped, mapped_column
from pydantic import BaseModel, ConfigDict, Field, EmailStr
from datetime import datetime
from typing import List, Optional
# SQLAlchemy модели
class Base(DeclarativeBase):
pass
class Author(Base):
__tablename__ = "authors"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
email: Mapped[str]
bio: Mapped[str | None]
posts: Mapped[List["Post"]] = relationship("Post", back_populates="author")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
content: Mapped[str]
author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
updated_at: Mapped[datetime | None]
author: Mapped["Author"] = relationship("Author", back_populates="posts")
# Pydantic модели:
# AuthorCreate, AuthorUpdate, AuthorResponse
# PostCreate, PostUpdate, PostResponse, PostWithAuthor
# Функции:
# create_author(db, author: AuthorCreate) -> Author
# get_author(db, author_id: int) -> AuthorResponse
# create_post(db, post: PostCreate) -> Post
# get_posts_by_author(db, author_id: int) -> List[PostResponse]Далее: Продвинутая сериализация: сериализаторы и контекст