⚡ Самые нужные конструкции
# 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, 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()
session.query(User).filter_by(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 (вместо class Config)
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.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.query(User).filter_by(name="Alice").first()
all_users = session.query(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 — альтернативный способ