رابطه‌ها و بارگذاری بهینه در SQLAlchemy

۱. کلید خارجی و رابطه

کلید خارجی، مثل Course.instructor_id، چیزی است که پایگاه داده ذخیره می‌کند. رابطه، مثل course.instructor، چیزی است که پایتون می‌بیند: صفتی که کلید خارجی را دنبال می‌کند و شیء مرتبط را برمی‌گرداند.

User 1 ──< Course                      instructor -> courses_taught
User 1 ──< Enrollment >── 1 Course     many-to-many with data on the link

class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True)
    full_name: Mapped[str] = mapped_column(String(120))
    role: Mapped[str] = mapped_column(String(20), default='student')

    courses_taught: Mapped[list['Course']] = relationship(back_populates='instructor')
    enrollments: Mapped[list['Enrollment']] = relationship(
        back_populates='student', cascade='all, delete-orphan'
    )

    def __repr__(self) -> str:
        return f'User({self.full_name!r})'


class Course(Base):
    __tablename__ = 'courses'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    capacity: Mapped[int] = mapped_column(default=20)
    instructor_id: Mapped[int] = mapped_column(ForeignKey('users.id'))

    instructor: Mapped['User'] = relationship(back_populates='courses_taught')
    enrollments: Mapped[list['Enrollment']] = relationship(
        back_populates='course', cascade='all, delete-orphan'
    )
    # A read-only shortcut THROUGH the enrollments table:
    students: Mapped[list['User']] = relationship(secondary='enrollments', viewonly=True)

    def __repr__(self) -> str:
        return f'Course({self.title!r})'


class Enrollment(Base):
    __tablename__ = 'enrollments'
    __table_args__ = (UniqueConstraint('student_id', 'course_id'),)

    id: Mapped[int] = mapped_column(primary_key=True)
    student_id: Mapped[int] = mapped_column(ForeignKey('users.id', ondelete='CASCADE'))
    course_id: Mapped[int] = mapped_column(ForeignKey('courses.id', ondelete='CASCADE'))
    status: Mapped[str] = mapped_column(String(20), default='active')
    created_at: Mapped[datetime] = mapped_column(default=lambda: datetime.now(UTC))

    student: Mapped['User'] = relationship(back_populates='enrollments')
    course: Mapped['Course'] = relationship(back_populates='enrollments')

    def __repr__(self) -> str:
        return f'Enrollment({self.student.full_name} -> {self.course.title}, {self.status})'

مدل‌ها با رابطه: 01_relationships/models.py

• back_populates دو سمت را به هم وصل می‌کند: یک سمت را تنظیم کنید، سمت دیگر خودش به‌روز می‌شود.
• Mapped[list['Course']] نوعی است که ویرایشگر می‌فهمد. نام کلاس در رشته آمده، چون هنوز تعریف نشده است.
• students با secondary='enrollments' و viewonly=True یک میان‌بر فقط‌خواندنی از روی جدول ثبت‌نام است.

۲. یک‌به‌چند: مدرس و دوره‌هایش

def main() -> None:
    reset_database()
    with Session(engine) as session:
        ali = session.scalar(select(User).where(User.full_name == 'Ali Teacher'))
        assert ali is not None
        print(ali, 'teaches', ali.courses_taught)

        course = session.get(Course, 3)
        assert course is not None
        print(course, 'is taught by', course.instructor)

        git_course = Course(title='Git', instructor=ali)
        print('before flush, instructor_id =', git_course.instructor_id)
        print('already in the list?', git_course in ali.courses_taught)
        session.add(git_course)
        session.flush()
        print('after flush, instructor_id  =', git_course.instructor_id)

        # Move a course to another teacher by changing the relationship
        mina = course.instructor
        git_course.instructor = mina
        session.commit()
        print('Mina now teaches', mina.courses_taught)

کار با رابطه از هر دو سمت: 01_one_to_many.py

User('Ali Teacher') teaches [Course('Python'), Course('FastAPI')]
before flush, instructor_id = None
already in the list? True
after flush, instructor_id  = 1

Course(title=..., instructor=ali) هنگام flush مقدار instructor_id را خودش پر می‌کند و به لطف back_populates دوره تازه همان لحظه در ali.courses_taught هست.

۳. چندبه‌چند با شیء واسط

یک دانشجو چند دوره می‌گیرد و یک دوره چند دانشجو دارد. جدول پیوند یعنی enrollments یک مدل کامل (Enrollment) است، چون خود پیوند داده دارد: وضعیت و تاریخ.

def main() -> None:
    reset_database()
    with Session(engine) as session:
        leila = session.scalar(select(User).where(User.full_name == 'Leila'))
        assert leila is not None
        print('Leila takes:', [e.course.title for e in leila.enrollments])

        fastapi = session.scalar(select(Course).where(Course.title == 'FastAPI'))
        assert fastapi is not None
        print('FastAPI students:', fastapi.students)

        # Cancel one enrollment (data on the link!)
        link = next(e for e in fastapi.enrollments if e.student is leila)
        link.status = 'cancelled'
        session.commit()

        active_students = session.scalars(
            select(User)
            .join(Enrollment, Enrollment.student_id == User.id)
            .where(Enrollment.course_id == fastapi.id, Enrollment.status == 'active')
            .order_by(User.full_name)
        ).all()
        print('active in FastAPI:', active_students)

        # Enroll a new student through the relationship
        omid = User(full_name='Omid')
        omid.enrollments.append(Enrollment(course=fastapi))
        session.add(omid)
        session.commit()
        print('after Omid:', fastapi.students)

ثبت‌نام، لغو و فهرست دانشجویان فعال: 02_many_to_many.py

۴. مسئله N+1: ببینید، بعد درست کنید

بارگذاری تنبل راحت است: course.enrollments اولین باری که به آن دست بزنید یک SELECT اجرا می‌کند. در حلقه روی N دوره این یعنی یک پرس‌وجو برای فهرست و N پرس‌وجو برای ثبت‌نام‌ها: N+1. با هزار دوره، هزار و یک رفت‌وبرگشت.

@event.listens_for(engine, 'before_cursor_execute')
def count_query(conn, cursor, statement, parameters, context, executemany) -> None:  # type: ignore[no-untyped-def]
    queries.append(statement)


def run(label: str, options: list, describe: Callable[[Course], str]) -> None:  # type: ignore[type-arg]
    queries.clear()
    with Session(engine) as session:
        courses = session.scalars(select(Course).options(*options)).all()
        lines = [describe(course) for course in courses]
    print(f'{label:<34} {len(queries)} queries   {lines[:2]} ...')


def main() -> None:
    reset_database()
    teacher = lambda c: f'{c.title} by {c.instructor.full_name}'  # noqa: E731
    size = lambda c: f'{c.title}: {len(c.enrollments)} students'  # noqa: E731

    run('instructor, lazy', [], teacher)
    run('instructor, joinedload', [joinedload(Course.instructor)], teacher)
    run('enrollments, lazy (N+1)', [], size)
    run('enrollments, selectinload', [selectinload(Course.enrollments)], size)

    # Two levels deep: every student -> enrollments -> course
    queries.clear()
    with Session(engine) as session:
        stmt = (
            select(User)
            .where(User.role == 'student')
            .options(selectinload(User.enrollments).selectinload(Enrollment.course))
        )
        for student in session.scalars(stmt):
            _ = [e.course.title for e in student.enrollments]
    print(f'{"students -> courses, nested":<34} {len(queries)} queries:')
    for sql in queries:
        print('   ', ' '.join(sql.split())[:90])


if __name__ == '__main__':
    main()

شمردن پرس‌وجوها با رویداد موتور: 03_n_plus_1.py

instructor, lazy                   3 queries   ['Python by Ali Teacher', 'FastAPI by Ali Teacher'] ...
instructor, joinedload             1 queries   ['Python by Ali Teacher', 'FastAPI by Ali Teacher'] ...
enrollments, lazy (N+1)            5 queries   ['Python: 4 students', 'FastAPI: 3 students'] ...
enrollments, selectinload          2 queries   ['Python: 4 students', 'FastAPI: 3 students'] ...
students -> courses, nested        3 queries:

۵. cascade: وقتی والد تغییر می‌کند

cascade='all, delete-orphan' روی Course.enrollments یعنی حذف دوره ثبت‌نام‌هایش را هم حذف می‌کند و بیرون آوردن یک ثبت‌نام از فهرست، آن ردیف را حذف می‌کند. روی courses_taught مدرس هیچ cascadeی نیست، و این عمدی است:

enrollments: 10
after removing one from the list: 9
after deleting the course: 6
refused: NOT NULL constraint failed: courses.instructor_id

وقتی مدرسی که هنوز دوره دارد حذف می‌شود، SQLAlchemy سعی می‌کند instructor_id دوره‌هایش را NULL کند و ستون NOT NULL جلویش را می‌گیرد. مدرسِ دارای دوره نمی‌تواند ناپدید شود.

۶. یک تراکنش، چند ردیف

«این سه دوره را بگیر» باید یا کامل انجام شود یا اصلاً انجام نشود. اگر دوره سوم پر باشد، دو ثبت‌نام اول نباید باقی بمانند:

def enroll_in_many(student_id: int, course_ids: list[int]) -> None:
    with Session(engine) as session, session.begin():
        student = session.get(User, student_id)
        assert student is not None
        for course_id in course_ids:
            course = session.get(Course, course_id)
            assert course is not None
            taken = session.scalar(
                select(func.count()).where(
                    Enrollment.course_id == course_id, Enrollment.status == 'active'
                )
            )
            if (taken or 0) >= course.capacity:
                raise CourseFullError(f'{course.title} is full')
            session.add(Enrollment(student=student, course=course))
            session.flush()  # send it now, so the next count sees it

همه یا هیچ: 05_multi_record_transaction.py

flush() درون حلقه ثبت‌نام تازه را به پایگاه داده می‌فرستد تا شمارش بعدی آن را ببیند، بدون اینکه تراکنش تمام شود.

۷. پروژه: ثبت‌نام با رابطه‌ها

    def detail(self, course_id: int) -> CourseDetail:
        """The course AND its instructor in one query (joinedload), plus a count."""
        stmt = select(Course).options(joinedload(Course.instructor)).where(Course.id == course_id)
        course = self.db.scalar(stmt)
        if course is None:
            raise NotFoundError(f'Course {course_id} not found')
        active = EnrollmentService(self.db).active_count(course_id)
        return CourseDetail(
            **CourseRead.model_validate(course).model_dump(),
            instructor=InstructorSummary.model_validate(course.instructor),
            active_students=active,
        )

دوره و مدرسش در یک پرس‌وجو: 02_project/app/services.py

    def for_student(self, student_id: int) -> list[Enrollment]:
        """Enrollments with their course: 2 queries in total, not 1 + N."""
        UserService(self.db).get(student_id)
        stmt = (
            select(Enrollment)
            .options(selectinload(Enrollment.course))
            .where(Enrollment.student_id == student_id)
            .order_by(Enrollment.id)
        )
        return list(self.db.scalars(stmt))

    def students_of(self, course_id: int, status: str | None = None) -> list[CourseStudent]:
        """One query with a JOIN, returning (enrollment, user) pairs."""
        CourseService(self.db).get(course_id)
        stmt = (
            select(Enrollment, User)
            .join(Enrollment.student)
            .where(Enrollment.course_id == course_id)
            .order_by(User.full_name)
        )
        if status:
            stmt = stmt.where(Enrollment.status == status)
        return [
            CourseStudent(
                student_id=user.id,
                full_name=user.full_name,
                email=user.email,
                status=enrollment.status,
                enrolled_at=enrollment.created_at,
            )
            for enrollment, user in self.db.execute(stmt).tuples()
        ]

دوره‌های یک دانشجو و دانشجویان یک دوره: 02_project/app/services.py

۸. خلاصه فصل