🔖 Справочник: Pydantic v2 + SQLAlchemy 2.x

Шпаргалка для работы на практикуме

⚡ Самые нужные конструкции

# Pydantic v2 — field_validator
from pydantic import BaseModel, field_validator, Field, EmailStr
from pydantic import ConfigDict
from typing import Annotated
from decimal import Decimal

@field_validator('field_name')
@classmethod
def validate_field(cls, v): ...

# Annotated + Field
amount: Annotated[Decimal, Field(gt=0)]
tx_type: Annotated[str, Field(pattern=r"^(debit|credit)$")]
currency: Annotated[str, Field(min_length=3, max_length=3)]
email: EmailStr
model_config = ConfigDict(str_strip_whitespace=True)

# SQLAlchemy 2.x — движок и модель
from sqlalchemy import create_engine, select, String, Integer, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

engine = create_engine('sqlite:///:memory:', echo=True)

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))

# Сессия
with Session(engine) as session:
    session.add(obj); session.commit()
    session.delete(obj); session.commit()
    users = session.scalars(
        select(User).where(User.name == "Alice")
    ).all()

1. Pydantic v2

Базовая модель

from pydantic import BaseModel, Field, EmailStr
from typing import Annotated
from decimal import Decimal

class MyModel(BaseModel):
    name: str                                      # обязательное поле
    age: int = 0                                   # с дефолтом
    password: str = Field(min_length=8)            # с ограничением
    email: EmailStr                                # валидация email
    amount: Annotated[Decimal, Field(gt=0)]        # Decimal > 0
    code: Annotated[str, Field(min_length=3,
                               max_length=3)]      # ровно 3 символа
    tx_type: Annotated[str,
             Field(pattern=r"^(debit|credit)$")]   # regex-паттерн

@field_validator (обязателен @classmethod)

from pydantic import BaseModel, field_validator
from datetime import datetime

class Event(BaseModel):
    title: str
    date: datetime
    location: str

    @field_validator('date')
    @classmethod
    def date_must_be_future(cls, v: datetime) -> datetime:
        if v < datetime.now():
            raise ValueError("Event date must be in the future")
        return v

    @field_validator('date')
    @classmethod
    def at_least_24h_ahead(cls, v: datetime) -> datetime:
        from datetime import timedelta
        if v < datetime.now() + timedelta(hours=24):
            raise ValueError("Must be at least 24 hours ahead")
        return v

ConfigDict

from pydantic import BaseModel, ConfigDict, Field

class UserProfile(BaseModel):
    username: str
    password: str = Field(min_length=8)

    model_config = ConfigDict(
        str_strip_whitespace=True,          # обрезать пробелы
        validate_assignment=True,            # проверять при присвоении
        json_schema_extra={
            "example": {
                "username": "john_doe",
                "password": "securePass123"
            }
        }
    )

Установка и импорты

pip install "pydantic[email]"    # EmailStr требует extra

from pydantic import (
    BaseModel,
    field_validator,
    Field,
    EmailStr,
    ConfigDict,
)
from typing import Annotated
from decimal import Decimal
from datetime import datetime, timedelta

2. SQLAlchemy 2.x

create_engine — строки подключения

from sqlalchemy import create_engine

# MySQL через pymysql
engine = create_engine(
    'mysql+pymysql://user:password@localhost/mydatabase'
)

# SQLite — файл
engine = create_engine('sqlite:///mydb.db')

# SQLite — в памяти (тесты, практикум)
engine = create_engine('sqlite:///:memory:', echo=True)

# echo=True — логировать все SQL-запросы в stdout

DeclarativeBase и модели

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, ForeignKey
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int] = mapped_column(Integer)
    posts: Mapped[list["Post"]] = relationship(
        "Post", back_populates="user"
    )

class Post(Base):
    __tablename__ = 'posts'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    user: Mapped["User"] = relationship("User", back_populates="posts")

Создание таблиц

Base.metadata.create_all(engine)    # создать все таблицы
Base.metadata.drop_all(engine)      # удалить все таблицы

Работа с Session

from sqlalchemy import select
from sqlalchemy.orm import Session

with Session(engine) as session:
    # Добавить объект
    new_user = User(name="Alice", age=30)
    session.add(new_user)
    session.commit()

    # Добавить несколько объектов
    session.add_all([obj1, obj2, obj3])
    session.commit()

    # Запрос
    user = session.scalars(select(User).where(User.name == "Alice")).first()
    all_users = session.scalars(select(User)).all()

    # Удалить объект
    session.delete(user)
    session.commit()

Логирование SQLAlchemy

import logging

logging.basicConfig(level=logging.INFO)
# или более точно:
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# echo=True в create_engine — альтернативный способ