Relational Queries & Filtering

The task router exposes filters; crud.list_tasks is what does the querying.

We'll see new ways to represent SQL Queries with SQLModel:

• where predicates that are conditionally applied
• join to filter by a related table's column
• selectinload to efficiently load a relationships

Conditional where Clauses

Each filter is independent.

A request with no query parameters returns every task; a request with ?status=done&priority=high returns only the tasks that match both.

The pattern is to build the statement up one where at a time, adding only the parameters that actually got passed in.

src/release_tracker/crud.py

from datetime import UTC, datetime

from sqlalchemy.orm import selectinload
from sqlmodel import Session, select

from .models import Project, Task, TaskPriority, TaskStatus


def list_tasks(
    session: Session,
    *,
    project_id: int | None = None,
    project_slug: str | None = None,
    task_status: TaskStatus | None = None,
    task_priority: TaskPriority | None = None,
    overdue_only: bool = False,
) -> list[Task]:
    statement = select(Task).options(selectinload(Task.project))  # type: ignore[arg-type]

    if project_id is not None:
        statement = statement.where(Task.project_id == project_id)
    if project_slug is not None:
        statement = statement.join(Project).where(Project.slug == project_slug)
    if task_status is not None:
        statement = statement.where(Task.status == task_status)
    if task_priority is not None:
        statement = statement.where(Task.priority == task_priority)
    if overdue_only:
        statement = statement.where(
            Task.due_date != None,  # noqa: E711
            Task.due_date < datetime.now(UTC).date(),  # type: ignore[operator]
            Task.status != TaskStatus.done,
        )

    return list(session.exec(statement).all())

statement = statement.where(...):

where returns a new statement with the predicate appended, so reassigning the variable composes filters incrementally. SQLAlchemy turns multiple where calls into a single SQL WHERE with AND between them, so two where calls and one where with a tuple are equivalent.

Note

The # noqa: E711 on Task.due_date != None is intentional.

---

Joining Through a Relationship

project_slug is a column on projects, not tasks. To filter tasks by a project's slug, the SQL needs a join:

Python

if project_slug is not None:
    statement = statement.join(Project).where(Project.slug == project_slug)

statement.join(Project) adds an SQL JOIN projects ON tasks.project_id = projects.id.

---

The N+1 Problem and selectinload

TaskRead exposes project_name and project_slug. Those come from properties on Task:

src/release_tracker/models.py

class Task(TaskBase, table=True):
    # ... id, project_id, project, etc. ...

    @property
    def project_name(self) -> str:
        return self.project.name

    @property
    def project_slug(self) -> str:
        return self.project.slug

Each property reads self.project, which lazy-loads the related Project row from the database the first time it's accessed.

Without any other configuration, calling crud.list_tasks and serializing 50 tasks runs 51 queries: one for the tasks, plus one per task to fetch its project. That's the N+1 query problem.

selectinload is the fix. Adding it to the select() tells SQLAlchemy to fetch all the projects for the loaded tasks in one extra query, eagerly:

Python

statement = select(Task).options(selectinload(Task.project))  # type: ignore[arg-type]

Two queries total, regardless of how many tasks come back. The first query loads the tasks, then the second does an IN (...) lookup with the project_ids from the first result and stitches the projects onto the loaded Task objects.

By the time the response serializer accesses task.project.name, the data is already in memory.

get_task uses the same pattern, since TaskRead always wants project_name and project_slug:

src/release_tracker/crud.py

def get_task(session: Session, task_id: int) -> Task | None:
    statement = (
        select(Task)
        .options(selectinload(Task.project))  # type: ignore[arg-type]
        .where(Task.id == task_id)
    )
    return session.exec(statement).first()

---

What the Other CRUD Functions Look Like

create_task, update_task, and delete_task are short and don't need filtering:

src/release_tracker/crud.py

def create_task(session: Session, project_id: int, payload: TaskCreate) -> Task:
    task = Task.model_validate(payload, update={"project_id": project_id})
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


def update_task(session: Session, task: Task, payload: TaskUpdate) -> Task:
    task.sqlmodel_update(payload.model_dump(exclude_unset=True))
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


def delete_task(session: Session, task: Task) -> None:
    session.delete(task)
    session.commit()

---