Декларативные модели, сессии, базовые CRUD операции
ORM (Object-Relational Mapping) позволяет работать с базой данных как с объектами Python. SQLAlchemy 2.0 представляет современный подход с полной поддержкой type hints.
В SQLAlchemy 2.0 рекомендуется использовать новый стиль декларативного базового класса:
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
"""Базовый класс для всех ORM-моделей."""
passfrom sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime
class User(Base):
__tablename__ = 'users'
# SQLAlchemy 2.0 стиль с type hints
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Связь с другими моделями
posts: Mapped[list["Post"]] = relationship(back_populates="author", lazy="selectin")
def __repr__(self):
return f"<User(id={self.id}, email='{self.email}')>"
class Post(Base):
__tablename__ = 'posts'
id: Mapped[int] = mapped_column(Integer, primary_key=True)
title: Mapped[str] = mapped_column(String(200), nullable=False)
content: Mapped[str | None] = mapped_column(String, nullable=True)
user_id: Mapped[int] = mapped_column(Integer, ForeignKey('users.id'), nullable=False)
published: Mapped[bool] = mapped_column(Boolean, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Обратная связь
author: Mapped["User"] = relationship(back_populates="posts")
def __repr__(self):
return f"<Post(id={self.id}, title='{self.title}')>"# SQLAlchemy 1.x стиль (всё ещё работает, но не рекомендуется)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
email = Column(String(255), unique=True, nullable=False)
# ...from sqlalchemy.orm import sessionmaker, Session
# Фабрика сессий
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
# Создание сессии
session = SessionLocal()
# Или напрямую
with Session(engine) as session:
# работа с сессией
passfrom sqlalchemy.orm import Session
with Session(engine) as session:
try:
# Работа с БД
user = User(email='test@example.com', name='Test')
session.add(user)
session.commit()
except Exception:
session.rollback()
raise# dependency для FastAPI
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Использование в endpoint
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
return db.get(User, user_id)# Создание одного объекта
user = User(email='alice@example.com', name='Alice')
session.add(user)
session.commit()
session.refresh(user) # Загрузить данные из БД (id, created_at, etc.)
# Или через add_all для нескольких
users = [
User(email='a@example.com', name='A'),
User(email='b@example.com', name='B'),
]
session.add_all(users)
session.commit()
# После commit id автоматически доступен
print(user.id) # 1# Получить по первичному ключу
user = session.get(User, 1) # None если не найдено
# Получить все
users = session.query(User).all()
# SQLAlchemy 2.0 стиль
from sqlalchemy import select
stmt = select(User)
users = session.execute(stmt).scalars().all()
# Фильтрация
stmt = select(User).where(User.email == 'alice@example.com')
user = session.execute(stmt).scalars().first()
# Фильтрация через filter (старый стиль, но удобный)
user = session.query(User).filter_by(email='alice@example.com').first()
user = session.query(User).filter(User.id > 1).all()# Обновление одного объекта
user = session.get(User, 1)
if user:
user.name = 'Alice Updated'
session.commit()
# Массовое обновление (Core стиль)
from sqlalchemy import update
stmt = update(User).where(User.published == False).values(published=True)
session.execute(stmt)
session.commit()
# Обновление через query (старый стиль)
session.query(User).filter(User.id == 1).update({User.name: 'Updated'})
session.commit()# Удаление объекта
user = session.get(User, 1)
if user:
session.delete(user)
session.commit()
# Массовое удаление
from sqlalchemy import delete
stmt = delete(User).where(User.created_at < datetime(2023, 1, 1))
session.execute(stmt)
session.commit()
# Через query
session.query(User).filter(User.id == 1).delete()
session.commit()from sqlalchemy import select, func
# Все записи
stmt = select(User)
users = session.execute(stmt).scalars().all()
# С фильтрацией
stmt = select(User).where(User.email.like('%@gmail.com'))
users = session.execute(stmt).scalars().all()
# Сортировка
stmt = select(User).order_by(User.name.asc(), User.created_at.desc())
# LIMIT и OFFSET
stmt = select(User).limit(10).offset(20)
# Агрегации
stmt = select(func.count(User.id))
count = session.execute(stmt).scalar()
# GROUP BY
stmt = select(User.user_id, func.count(Post.id)).join(Post).group_by(User.user_id)# Цепочка методов (очень выразительно)
users = (
session.query(User)
.filter(User.email.like('%@gmail.com'))
.order_by(User.name)
.limit(10)
.all()
)
# count
count = session.query(User).count()
# first
user = session.query(User).filter_by(email='test@example.com').first()class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
posts: Mapped[list["Post"]] = relationship(
back_populates="author",
lazy="selectin", # Стратегия загрузки
cascade="all, delete-orphan" # Каскадное удаление
)
class Post(Base):
__tablename__ = 'posts'
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
author: Mapped["User"] = relationship(back_populates="posts")# Ассоциативная таблица
user_tags = Table(
'user_tags', Base.metadata,
Column('user_id', ForeignKey('users.id'), primary_key=True),
Column('tag_id', ForeignKey('tags.id'), primary_key=True),
)
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
tags: Mapped[list["Tag"]] = relationship(
secondary=user_tags,
back_populates="users"
)
class Tag(Base):
__tablename__ = 'tags'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
users: Mapped[list["User"]] = relationship(back_populates="tags")# lazy='select' (по умолчанию) - загрузка при доступе (N+1!)
posts: Mapped[list["Post"]] = relationship(lazy='select')
# lazy='selectin' - отдельный SELECT с IN (рекомендуется)
posts: Mapped[list["Post"]] = relationship(lazy='selectin')
# lazy='joined' - JOIN (хорошо для one-to-one)
author: Mapped["User"] = relationship(lazy='joined')
# lazy='raise' - ошибка при ленивой загрузке (для отладки)
posts: Mapped[list["Post"]] = relationship(lazy='raise')
# Явное указание в запросе
from sqlalchemy.orm import joinedload, selectinload
stmt = select(User).options(joinedload(User.posts))
users = session.execute(stmt).scalars().unique().all()
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).scalars().all()Session реализует паттерн Unit of Work:
with Session(engine) as session:
# Все изменения отслеживаются
user = session.get(User, 1)
user.name = 'Updated' # Отслеживается
new_post = Post(title='New', user_id=1)
session.add(new_post) # Отслеживается
session.delete(some_post) # Отслеживается
# При commit все изменения синхронизируются с БД
session.commit() # Один transaction
# При ошибке — автоматический rollback
# session.rollback()from sqlalchemy import inspect
inspector = inspect(user)
# Состояние объекта
inspector.transient # True если не в сессии
inspector.pending # True если добавлен но не закоммичен
inspector.persistent # True если в БД
inspector.detached # True если сессия закрыта
# Изменённые атрибуты
inspector.modified # Список изменённых полей# Используйте context manager для сессий
with Session(engine) as session:
session.commit()
# Используйте type hints для моделей
class User(Base):
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
# Используйте __repr__ для отладки
def __repr__(self):
return f"<User(id={self.id})>"
# Используйте cascade для связанных объектов
posts = relationship("Post", cascade="all, delete-orphan", back_populates="author")
# Используйте unique() при joinedload
users = session.execute(
select(User).options(joinedload(User.posts))
).unique().scalars().all()# Не создавайте сессию без context manager
session = Session(engine) # ПЛОХО: может не закрыться
# Не забывайте commit
session.add(user) # ПЛОХО: нет session.commit()
# Не используйте N+1 запросы
for user in users:
print(user.posts) # ПЛОХО: N+1 запросов
# Используйте eager loading
users = session.execute(
select(User).options(selectinload(User.posts))
).scalars().all()Далее: Relationships и связи