Relationships и связи

Правильное проектирование отношений между моделями — ключ к эффективной работе с ORM. В этой теме вы изучите все типы связей и лучшие практики их использования.

#One-to-Many (Один-ко-многим)

Самый распространённый тип отношений.

#Базовая связь

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")

#Каскадные операции (Cascade)

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-orphan

#Many-to-Many (Многие-ко-многим)

#Через ассоциативную таблицу

from 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()

#Many-to-Many с дополнительными полями

Когда нужно хранить метаданные в связи (например, дату добавления тега).

#Ассоциативная модель

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}")

#One-to-One (Один-к-одному)

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()

#Self-Referential (Рекурсивная связь)

Связь модели с самой собой (например, комментарии с ответами).

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}")

#Стратегии загрузки (Loading Strategies)

#lazy параметр

# 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 + подзапрос)	Сложные фильтры

#Backref vs back_populates

#back_populates (рекомендуется)

class User(Base):
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    author: Mapped["User"] = relationship(back_populates="posts")

Преимущества:

Явно и понятно
Легче отлаживать
Рекомендуется в SQLAlchemy 2.0

#backref (короткая запись)

class User(Base):
    posts: Mapped[list["Post"]] = relationship(
        backref="author",  # Автоматически создаёт author в Post
        lazy="selectin"
    )

class Post(Base):
    __tablename__ = 'posts'
    # author создаётся автоматически

Недостатки:

Менее явно
Сложнее отлаживать
Не рекомендуется в новом коде

#Association Proxy

Удобный доступ к связанным объектам через 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")

#Использование proxy

user = session.get(User, 1)

# Доступ к proxy как к списку
print(user.tag_names)  # ['python', 'sqlalchemy']

# Добавление через proxy
user.tag_names.append('fastapi')
session.commit()

#Best Practices

#✅ Делайте

# Используйте 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 останутся

Каскад

Описание

save-update

Добавление в сессию каскадируется

merge

merge() каскадируется

refresh-expire

refresh/expire каскадируются

expunge

expunge() каскадируется

delete

delete() каскадируется

all

Все вышеперечисленные

delete-orphan

Удаление осиротевших объектов

Стратегия

Запросов

Когда использовать

select (lazy)

N+1

Только для редкого доступа

joinedload

1 (JOIN)

One-to-one, мало дочерних

selectinload

2 (SELECT + IN)

One-to-many, много дочерних

subqueryload

2 (SELECT + подзапрос)

Сложные фильтры