Relationships & Performance Checkpoint
25 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)Fix the mistakes
The slips that cost marks in this module. Spot each one, then check the fix.
A report page lists students with their courses. A careless version sends a query for every student. This checkpoint designs the links, loads them well and proves the result by counting statements.
Design the links first
Find the cardinality of each link. One-to-many puts the foreign key on the many side. Many-to-many needs an association table. Decide what happens to children when the parent is deleted.
enrolment = Table( "enrolment", Base.metadata, Column("student_id", ForeignKey("student.id"), primary_key=True), Column("course_id", ForeignKey("course.id"), primary_key=True),)The two keys together are the primary key, so one student cannot join the same course twice.
Choose the loader for the page
Ask what the page shows. A list of parents with their lists needs selectinload. A list of children with one parent each can use joinedload. A page that shows only totals needs an aggregate query.
stmt = ( select(Student) .options(selectinload(Student.courses)) .order_by(Student.id))students = session.scalars(stmt).all()Example 1 · Three versions of the same report
from contextlib import contextmanagerfrom sqlalchemy import event, create_engine, select, func, String, ForeignKey, Table, Columnfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session, selectinload, joinedload
class Base(DeclarativeBase): pass
enrolment = Table( "enrolment", Base.metadata, Column("student_id", ForeignKey("student.id"), primary_key=True), Column("course_id", ForeignKey("course.id"), primary_key=True),)
class Student(Base): __tablename__ = "student" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50)) courses: Mapped[list["Course"]] = relationship(secondary=enrolment, back_populates="students")
class Course(Base): __tablename__ = "course" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(60)) students: Mapped[list["Student"]] = relationship(secondary=enrolment, back_populates="courses")
engine = create_engine("sqlite://")Base.metadata.create_all(engine)
def seed(): with Session(engine) as s: py, sql, web = Course(title="Python"), Course(title="SQL"), Course(title="Web") s.add_all([ Student(name="Asha", courses=[py, sql]), Student(name="Ravi", courses=[sql, web]), Student(name="Meena", courses=[py]), ]) s.commit()
seed()
@contextmanagerdef count_queries(engine): statements = []
def before(conn, cursor, statement, parameters, context, executemany): statements.append(statement)
event.listen(engine, "before_cursor_execute", before) try: yield statements finally: event.remove(engine, "before_cursor_execute", before)
def run(options, unique=False): with Session(engine) as s, count_queries(engine) as q: stmt = select(Student).options(*options).order_by(Student.id) result = s.scalars(stmt) if unique: result = result.unique() data = [(st.name, len(st.courses)) for st in result] return data, len(q)
print(run([]))print(run([selectinload(Student.courses)])[1])print(run([joinedload(Student.courses)], unique=True)[1])([('Asha', 2), ('Meena', 1), ('Ravi', 2)], 4)
2
1What this code does
- The lazy version sent 1 + 3 = 4 queries and gave the data.
selectinloadsent 2 queries. A joined list needsunique()and sent 1.
The students come first, and all their courses come in one more query, however many students there are.
Prove it with a count
A claim about speed needs a number. Wrap the code in the statement counter and print or assert the count. Keep the assert in a test.
with count_queries(engine) as statements: names = [(s.name, len(s.courses)) for s in students]assert len(statements) == 0Example 2 · One aggregate query for a count page
from contextlib import contextmanagerfrom sqlalchemy import event, create_engine, select, func, String, ForeignKey, Table, Columnfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session, selectinload, joinedload
class Base(DeclarativeBase): pass
enrolment = Table( "enrolment", Base.metadata, Column("student_id", ForeignKey("student.id"), primary_key=True), Column("course_id", ForeignKey("course.id"), primary_key=True),)
class Student(Base): __tablename__ = "student" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50)) courses: Mapped[list["Course"]] = relationship(secondary=enrolment, back_populates="students")
class Course(Base): __tablename__ = "course" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(60)) students: Mapped[list["Student"]] = relationship(secondary=enrolment, back_populates="courses")
engine = create_engine("sqlite://")Base.metadata.create_all(engine)
def seed(): with Session(engine) as s: py, sql, web = Course(title="Python"), Course(title="SQL"), Course(title="Web") s.add_all([ Student(name="Asha", courses=[py, sql]), Student(name="Ravi", courses=[sql, web]), Student(name="Meena", courses=[py]), ]) s.commit()
seed()
@contextmanagerdef count_queries(engine): statements = []
def before(conn, cursor, statement, parameters, context, executemany): statements.append(statement)
event.listen(engine, "before_cursor_execute", before) try: yield statements finally: event.remove(engine, "before_cursor_execute", before)
with Session(engine) as s, count_queries(engine) as q: stmt = ( select(Course.title, func.count(Student.id)) .join(Course.students) .group_by(Course.id, Course.title) .order_by(Course.title) ) print([tuple(r) for r in s.execute(stmt)])print(len(q))[('Python', 2), ('SQL', 2), ('Web', 1)]
1What this code does
- The database counted the students of each course.
- One statement, and no
StudentorCourseobjects were built.
If the courses were loaded up front, reading them runs no statement. A lazy version would run one for every student.
Finish up
Mark complete & continue
Next: Creating Migrations