Одно-ко-многим, многие-ко-многим, loading strategies
Правильное проектирование отношений между моделями — ключ к эффективной работе с ORM. В этой теме вы изучите все типы связей и лучшие практики их использования.
Самый распространённый тип отношений.
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
# Один пользователь имеет много постов
posts: Mapped[list["Post"]] = relationship(
back_populates="author",
lazy="selectin", # Стратегия загрузки
)
class Post(Base):
__tablename__ = 'posts'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
# Каждый пост принадлежит одному автору
author: Mapped["User"] = relationship(back_populates="posts")class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
# Каскадное удаление: при удалении пользователя удаляются все посты
posts: Mapped[list["Post"]] = relationship(
back_populates="author",
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")| Каскад | Описание |
|---|---|
save-update | Добавление в сессию каскадируется |
merge | merge() каскадируется |
refresh-expire | refresh/expire каскадируются |
expunge | expunge() каскадируется |
delete | delete() каскадируется |
all | Все вышеперечисленные |
delete-orphan | Удаление осиротевших объектов |
# Пример orphan
user = session.get(User, 1)
user.posts = [] # Все посты стали orphan
session.commit() # Посты удалены из БД при delete-orphanfrom sqlalchemy import Table, Column, ForeignKey
# Ассоциативная таблица
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)
name: Mapped[str] = mapped_column(String(100))
# Многие-ко-многим с Tag
tags: Mapped[list["Tag"]] = relationship(
secondary=user_tags, # Ассоциативная таблица
back_populates="users",
lazy="selectin",
)
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")# Добавление тегов пользователю
user = session.get(User, 1)
tag1 = Tag(name='python')
tag2 = Tag(name='sqlalchemy')
user.tags.append(tag1)
user.tags.append(tag2)
session.commit()
# Получение всех пользователей с тегом 'python'
from sqlalchemy import select
stmt = select(User).join(User.tags).where(Tag.name == 'python')
users = session.execute(stmt).scalars().all()Когда нужно хранить метаданные в связи (например, дату добавления тега).
class UserTag(Base):
__tablename__ = 'user_tags'
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), primary_key=True)
tag_id: Mapped[int] = mapped_column(ForeignKey('tags.id'), primary_key=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
score: Mapped[int] = mapped_column(Integer, default=0) # Дополнительное поле
# Отношения
user: Mapped["User"] = relationship(back_populates="user_tags")
tag: Mapped["Tag"] = relationship(back_populates="user_tags")
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
user_tags: Mapped[list["UserTag"]] = relationship(
back_populates="user",
cascade="all, delete-orphan",
)
# Convenience property для доступа к тегам
@property
def tags(self) -> list["Tag"]:
return [ut.tag for ut in self.user_tags]
class Tag(Base):
__tablename__ = 'tags'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
user_tags: Mapped[list["UserTag"]] = relationship(back_populates="tag")user = session.get(User, 1)
tag = session.get(Tag, 1)
# Создание связи с метаданными
user_tag = UserTag(user=user, tag=tag, score=10)
session.add(user_tag)
session.commit()
# Доступ к метаданным
for ut in user.user_tags:
print(f"Tag: {ut.tag.name}, Score: {ut.score}, Added: {ut.created_at}")class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255))
# Один профиль на пользователя
profile: Mapped["Profile"] = relationship(
back_populates="user",
uselist=False, # Важно: один объект, не список
lazy="joined", # JOIN загрузка эффективна для one-to-one
)
class Profile(Base):
__tablename__ = 'profiles'
id: Mapped[int] = mapped_column(primary_key=True)
bio: Mapped[str] = mapped_column(String, nullable=True)
avatar_url: Mapped[str] = mapped_column(String, nullable=True)
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), unique=True)
user: Mapped["User"] = relationship(back_populates="profile")user = session.get(User, 1)
# Доступ к профилю
print(user.profile.bio)
# Создание профиля
profile = Profile(bio='Hello', user=user)
session.add(profile)
session.commit()Связь модели с самой собой (например, комментарии с ответами).
class Comment(Base):
__tablename__ = 'comments'
id: Mapped[int] = mapped_column(primary_key=True)
content: Mapped[str] = mapped_column(String)
parent_id: Mapped[int | None] = mapped_column(ForeignKey('comments.id'))
# Родительский комментарий
parent: Mapped["Comment | None"] = relationship(
back_populates="replies",
remote_side=[id], # Важно: указывает на "родительскую" сторону
)
# Дочерние комментарии
replies: Mapped[list["Comment"]] = relationship(
back_populates="parent",
lazy="selectin",
)# Создание дерева комментариев
root = Comment(content='Root comment')
reply1 = Comment(content='Reply 1', parent=root)
reply2 = Comment(content='Reply 2', parent=root)
sub_reply = Comment(content='Sub reply', parent=reply1)
session.add_all([root, reply1, reply2, sub_reply])
session.commit()
# Доступ к дереву
for reply in root.replies:
print(reply.content)
for sub in reply.replies:
print(f" - {sub.content}")# lazy='select' (по умолчанию) - ленивая загрузка при доступе
posts: Mapped[list["Post"]] = relationship(lazy='select')
# Проблема: N+1 запросов при итерации
# lazy='selectin' - отдельный SELECT с IN
posts: Mapped[list["Post"]] = relationship(lazy='selectin')
# Хорошо для one-to-many
# lazy='joined' - JOIN
author: Mapped["User"] = relationship(lazy='joined')
# Хорошо для one-to-one
# lazy='raise' - ошибка при ленивой загрузке
posts: Mapped[list["Post"]] = relationship(lazy='raise')
# Для отладки N+1
# lazy='noload' - никогда не загружать
posts: Mapped[list["Post"]] = relationship(lazy='noload')from sqlalchemy.orm import joinedload, selectinload, subqueryload, lazyload
# Joinedload (JOIN)
stmt = select(User).options(joinedload(User.posts))
users = session.execute(stmt).scalars().unique().all()
# Selectinload (отдельный SELECT с IN)
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).scalars().all()
# Subqueryload (подзапрос)
stmt = select(User).options(subqueryload(User.posts))
users = session.execute(stmt).scalars().all()
# Lazyload (принудительная ленивая загрузка)
stmt = select(User).options(lazyload(User.posts))
# Загрузка конкретной колонки
from sqlalchemy.orm import load_only
stmt = select(User).options(load_only(User.id, User.name))| Стратегия | Запросов | Когда использовать |
|---|---|---|
select (lazy) | N+1 | Только для редкого доступа |
joinedload | 1 (JOIN) | One-to-one, мало дочерних |
selectinload | 2 (SELECT + IN) | One-to-many, много дочерних |
subqueryload | 2 (SELECT + подзапрос) | Сложные фильтры |
class User(Base):
posts: Mapped[list["Post"]] = relationship(back_populates="author")
class Post(Base):
author: Mapped["User"] = relationship(back_populates="posts")Преимущества:
class User(Base):
posts: Mapped[list["Post"]] = relationship(
backref="author", # Автоматически создаёт author в Post
lazy="selectin"
)
class Post(Base):
__tablename__ = 'posts'
# author создаётся автоматическиНедостатки:
Удобный доступ к связанным объектам через proxy.
from sqlalchemy.ext.associationproxy import association_proxy
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
user_tags: Mapped[list["UserTag"]] = relationship(back_populates="user")
# Proxy для доступа к именам тегов напрямую
tag_names = association_proxy('user_tags', 'tag_name')
@property
def tags(self) -> list["Tag"]:
return [ut.tag for ut in self.user_tags]
class UserTag(Base):
__tablename__ = 'user_tags'
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), primary_key=True)
tag_id: Mapped[int] = mapped_column(ForeignKey('tags.id'), primary_key=True)
tag_name: Mapped[str] = mapped_column(String(50)) # Денормализация
user: Mapped["User"] = relationship(back_populates="user_tags")
tag: Mapped["Tag"] = relationship(back_populates="user_tags")user = session.get(User, 1)
# Доступ к proxy как к списку
print(user.tag_names) # ['python', 'sqlalchemy']
# Добавление через proxy
user.tag_names.append('fastapi')
session.commit()# Используйте back_populates вместо backref
posts = relationship("Post", back_populates="author")
# Указывайте lazy стратегию явно
posts = relationship("Post", lazy="selectin", back_populates="author")
# Используйте cascade для зависимых объектов
posts = relationship("Post", cascade="all, delete-orphan", back_populates="author")
# Используйте uselist=False для one-to-one
profile = relationship("Profile", uselist=False, back_populates="user")
# Используйте remote_side для self-referential
parent = relationship("Comment", back_populates="replies", remote_side=[id])
# Используйте unique() с joinedload
users = session.execute(
select(User).options(joinedload(User.posts))
).unique().scalars().all()# Не используйте lazy='select' без необходимости (N+1!)
posts = relationship("Post", lazy='select') # ПЛОХО
# Не забывайте back_populates (связь не будет работать)
class User(Base):
posts = relationship("Post") # ПЛОХО: нет back_populates
# Не используйте backref в новом коде
posts = relationship("Post", backref="author") # ПЛОХО: неявно
# Не забывайте про cascade если объекты зависимы
posts = relationship("Post") # ПЛОХО: при удалении user posts останутсяВ следующей теме вы изучите Query API — продвинутые техники построения запросов, оконные функции, CTE и оптимизацию сложных запросов.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.