💻 Примеры: Pydantic v2 + SQLAlchemy 2.x

Примеры Pydantic v2

Пример 1: Event — валидация даты через field_validator

Что демонстрирует: как написать кастомный валидатор в Pydantic v2. Метод должен быть @classmethod, первым аргументом принимает cls, вторым — значение поля.

from pydantic import BaseModel, field_validator
from datetime import datetime, timedelta

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

# Корректное создание
future_event = Event(
    title="New Year Party",
    date=datetime.now() + timedelta(days=30),
    location="New York"
)
print(future_event)
# title='New Year Party' date=... location='New York'

# Некорректное создание — выбрасывает ValidationError
try:
    past_event = Event(
        title="Old Party",
        date=datetime(2020, 1, 1),
        location="London"
    )
except Exception as e:
    print(e)  # ValidationError: date — Event date must be in the future

Пример 2: UserProfile — Field с ограничениями, EmailStr, ConfigDict

Что демонстрирует: Field с min_length, тип EmailStr для автоматической валидации email, model_config вместо class Config.

from pydantic import BaseModel, EmailStr, Field, ConfigDict

class UserProfile(BaseModel):
    username: str
    password: str = Field(
        min_length=8,
        description="Password must be at least 8 characters long"
    )
    email: EmailStr

    model_config = ConfigDict(
        str_strip_whitespace=True,
        json_schema_extra={
            "example": {
                "username": "john_doe",
                "password": "securePassword123",
                "email": "john.doe@example.com"
            }
        }
    )

# Корректное создание
user = UserProfile(
    username="john_doe",
    password="securePassword123",
    email="john.doe@example.com"
)
print(user)

# Некорректный email — ValidationError
try:
    bad = UserProfile(username="alice", password="password123", email="aliceexample.com")
except Exception as e:
    print(e)  # value is not a valid email address

# Короткий пароль — ValidationError
try:
    short = UserProfile(username="bob", password="123", email="bob@example.com")
except Exception as e:
    print(e)  # String should have at least 8 characters

Пример 3: Transaction — Annotated[Decimal, Field] и str_strip_whitespace

Что демонстрирует: использование Annotated для ограничений типа Decimal и строки с regex-паттерном. str_strip_whitespace=True автоматически обрезает пробелы.

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

class Transaction(BaseModel):
    amount: Annotated[Decimal, Field(gt=0)]
    transaction_type: Annotated[str, Field(pattern=r"^(debit|credit)$")]
    currency: Annotated[str, Field(min_length=3, max_length=3)]

    model_config = ConfigDict(str_strip_whitespace=True)

# Корректная транзакция
tx = Transaction(
    amount=Decimal("150.50"),
    transaction_type="debit",
    currency="USD"
)
print(tx)
# amount=Decimal('150.50') transaction_type='debit' currency='USD'

# Сумма <= 0 — ValidationError
try:
    bad_tx = Transaction(amount=Decimal("-10"), transaction_type="debit", currency="USD")
except Exception as e:
    print(e)  # Input should be greater than 0

# Неверный тип транзакции — ValidationError
try:
    bad_type = Transaction(amount=Decimal("100"), transaction_type="transfer", currency="EUR")
except Exception as e:
    print(e)  # String should match pattern '^(debit|credit)$'

Пример 4: Appointment — поле с ограничением ≥24ч вперёд

Что демонстрирует: field_validator с проверкой временного интервала (не менее 24 часов от текущего момента).

from pydantic import BaseModel, field_validator
from datetime import datetime, timedelta

class Appointment(BaseModel):
    patient_name: str
    appointment_date: datetime

    @field_validator('appointment_date')
    @classmethod
    def check_appointment_date(cls, v: datetime) -> datetime:
        if v < datetime.now() + timedelta(hours=24):
            raise ValueError(
                "Appointment must be scheduled at least 24 hours in advance."
            )
        return v

# Корректная запись (через 25 часов)
appt = Appointment(
    patient_name="Alice Smith",
    appointment_date=datetime.now() + timedelta(hours=25)
)
print(appt)

# Запись через 1 час — ValidationError
try:
    soon = Appointment(
        patient_name="Bob Jones",
        appointment_date=datetime.now() + timedelta(hours=1)
    )
except Exception as e:
    print(e)  # Appointment must be scheduled at least 24 hours in advance.

Примеры SQLAlchemy 2.x

Пример 5: User и Address — one-to-many с relationship

Что демонстрирует: полная модель со связью один-ко-многим на SQLAlchemy 2.x. back_populates синхронизирует обе стороны связи.

from sqlalchemy import create_engine, String, Integer, ForeignKey
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column,
    relationship, Session
)
from typing import List, Optional

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))
    age: Mapped[int] = mapped_column(Integer)
    addresses: Mapped[List["Address"]] = relationship(
        "Address", back_populates="user"
    )

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

# Создать таблицы
Base.metadata.create_all(engine)

# Добавить данные
with Session(engine) as session:
    new_user = User(name="John Doe", age=28)
    addr = Address(street="123 Elm Street", user=new_user)
    session.add(new_user)
    session.add(addr)
    session.commit()
    print(f"User ID: {new_user.id}, Address: {addr.street}")

# Удалить пользователя
with Session(engine) as session:
    user = session.query(User).filter_by(name="John Doe").first()
    if user:
        session.delete(user)
        session.commit()
    # Проверить что удалён
    result = session.query(User).filter_by(name="John Doe").first()
    print(f"User after delete: {result}")  # None

Пример 6: MySQL + PyMySQL engine

Что демонстрирует: строка подключения для MySQL с указанием диалекта mysql+pymysql.

from sqlalchemy import create_engine

# Подключение к MySQL через PyMySQL
engine = create_engine(
    'mysql+pymysql://user:password@localhost/mydatabase'
)

# Установка: pip install pymysql