Как сделать Many-to-Many связь в SQLAlchemy 2.0?

Как сделать Many-to-Many связь в SQLAlchemy 2.0

Простой случай: базовая Many-to-Many

Когда нужны только связи без дополнительных данных:

from sqlalchemy import Column, Integer, String, Table, ForeignKey, create_engine
from sqlalchemy.orm import declarative_base, relationship, Session

Base = declarative_base()

# Таблица связи (association table)
student_course = Table(
    'student_course',
    Base.metadata,
    Column('student_id', Integer, ForeignKey('students.id'), primary_key=True),
    Column('course_id', Integer, ForeignKey('courses.id'), primary_key=True)
)

class Student(Base):
    __tablename__ = 'students'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    
    # Связь many-to-many
    courses = relationship(
        'Course',
        secondary=student_course,
        back_populates='students'
    )

class Course(Base):
    __tablename__ = 'courses'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    
    students = relationship(
        'Student',
        secondary=student_course,
        back_populates='courses'
    )

# Использование
engine = create_engine('sqlite:///db.sqlite')
Base.metadata.create_all(engine)

with Session(engine) as session:
    # Создаём объекты
    student = Student(name='Alice')
    course = Course(name='Python Basics')
    
    # Связываем их
    student.courses.append(course)
    
    session.add(student)
    session.commit()
    
    # Получаем
    alice = session.query(Student).filter_by(name='Alice').first()
    print(alice.courses)  # [<Course Python Basics>]

Сложный случай: Many-to-Many с дополнительными данными

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

from datetime import datetime

# Вместо Table используем полноценный класс
class StudentCourse(Base):
    __tablename__ = 'student_course'
    
    student_id = Column(Integer, ForeignKey('students.id'), primary_key=True)
    course_id = Column(Integer, ForeignKey('courses.id'), primary_key=True)
    
    # Дополнительные поля в таблице связи
    enrolled_at = Column(DateTime, default=datetime.utcnow)
    grade = Column(String, nullable=True)
    
    # Связи к родительским таблицам
    student = relationship('Student', back_populates='course_registrations')
    course = relationship('Course', back_populates='student_registrations')

class Student(Base):
    __tablename__ = 'students'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    
    # Связь через association object
    course_registrations = relationship(
        'StudentCourse',
        back_populates='student',
        cascade='all, delete-orphan'
    )
    
    # Удобный доступ к курсам
    @property
    def courses(self):
        return [reg.course for reg in self.course_registrations]

class Course(Base):
    __tablename__ = 'courses'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    
    student_registrations = relationship(
        'StudentCourse',
        back_populates='course',
        cascade='all, delete-orphan'
    )
    
    @property
    def students(self):
        return [reg.student for reg in self.student_registrations]

# Использование
with Session(engine) as session:
    student = Student(name='Bob')
    course = Course(name='Advanced Python')
    
    # Создаём регистрацию с дополнительными данными
    registration = StudentCourse(
        student=student,
        course=course,
        grade='A'
    )
    
    session.add(registration)
    session.commit()
    
    # Получаем
    bob = session.query(Student).filter_by(name='Bob').first()
    for reg in bob.course_registrations:
        print(f"{reg.course.name}: {reg.grade}")

SQLAlchemy 2.0 синтаксис (new style)

В SQLAlchemy 2.0 рекомендуется новый синтаксис:

from sqlalchemy.orm import Mapped, mapped_column, relationship
from typing import List
from datetime import datetime

class Student(Base):
    __tablename__ = 'students'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    
    # Many-to-many через association table
    courses: Mapped[List['Course']] = relationship(
        secondary=student_course,
        back_populates='students'
    )

class Course(Base):
    __tablename__ = 'courses'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    
    students: Mapped[List['Student']] = relationship(
        secondary=student_course,
        back_populates='courses'
    )

# Или с association object
class StudentCourse(Base):
    __tablename__ = 'student_course'
    
    student_id: Mapped[int] = mapped_column(
        ForeignKey('students.id'),
        primary_key=True
    )
    course_id: Mapped[int] = mapped_column(
        ForeignKey('courses.id'),
        primary_key=True
    )
    enrolled_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    grade: Mapped[str | None] = mapped_column(default=None)
    
    # Relationships
    student: Mapped['Student'] = relationship(back_populates='course_registrations')
    course: Mapped['Course'] = relationship(back_populates='student_registrations')

class Student(Base):
    __tablename__ = 'students'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    
    course_registrations: Mapped[List['StudentCourse']] = relationship(
        back_populates='student',
        cascade='all, delete-orphan'
    )

class Course(Base):
    __tablename__ = 'courses'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    
    student_registrations: Mapped[List['StudentCourse']] = relationship(
        back_populates='course',
        cascade='all, delete-orphan'
    )

Запросы к Many-to-Many

from sqlalchemy import select

# SQLAlchemy 1.x стиль
session = Session(engine)

# Получить студента с его курсами
student = session.query(Student).filter_by(name='Alice').first()
print([c.name for c in student.courses])

# Найти все курсы студента
student = session.query(Student).options(
    selectinload(Student.courses)
).filter_by(name='Alice').first()

# SQLAlchemy 2.0 синтаксис
stmt = select(Student).where(Student.name == 'Alice')
student = session.scalar(stmt)

# Найти студентов, которые учат Python
stmt = select(Student).join(
    student_course
).join(Course).where(
    Course.name.contains('Python')
)
students = session.scalars(stmt).unique().all()

Добавление и удаление связей

with Session(engine) as session:
    student = session.query(Student).filter_by(id=1).first()
    course = session.query(Course).filter_by(id=1).first()
    
    # Добавить связь
    student.courses.append(course)
    session.commit()
    
    # Удалить связь
    student.courses.remove(course)
    session.commit()
    
    # Или с association object
    student = session.query(Student).filter_by(id=1).first()
    course = session.query(Course).filter_by(id=1).first()
    
    reg = StudentCourse(student=student, course=course, grade='B')
    session.add(reg)
    session.commit()
    
    # Удалить
    session.delete(reg)
    session.commit()

Best Practices

1. Используй association object если нужны доп. данные
2. Добавляй cascade для удаления - cascade='all, delete-orphan'
3. Используй selectinload для загрузки связей
4. Типизируй - Mapped[List['Course']]
5. Тестируй запросы - смотри SQL через echo=True

Мany-to-Many это мощный инструмент, но нужно правильно его использовать!