#241: Filters for SQLAlchemy and FastAPI

Our filter for retrieving the tasks we are interested in still fetches all data from the database before applying the actual filtering. It is time to fix that and use a more flexible solution that works with the asynchronous SQLAlchemy that we introduced in the last post.

FastAPI Filter

There are multiple approaches to filter our data we can choose from. For this post I use FastAPI Filter, because it offers us a lot of flexibility. We can use everything it offers or, as I prefer, start small and add only the keywords we need.

By default, FastAPI Filter supports the following operators that we can append to our fields (field__[operator]):

neq – not equals
gt - greater than
gte - greater than or equal to
in – in a list of values
isnull – is NULL
lt - less than
lte - less than or equal to
not/ne – not or not equal to a specified value
not_in/nin – not in a list of values
like/ilike – like with case sensitive or case insensitive

Installation

We can install the fastapi-filter package with this command to support SQLAlchemy:

pip install fastapi-filter[sqlalchemy]

Keep in mind that you also need an asynchronous driver, for example aiosqlite, to connect to your database:

pip install aiosqlite

New tests for our data store

Even when we want to filter in FastAPI, we must make sure that we can use the filter with our database. Therefore, we start the implementation at the bottom and go upwards through our application.

Here are the tests we need to verify that the first filters work:

from ..models.task_filter import TaskFilter

...

@pytest.mark.asyncio
async def test_filter_empty_filter_gives_all_entries(with_db):
    store = DataStoreDb(with_db)
    await store.add(TaskInput(name="counter", 
                              priority=1, 
                              due_date=date.today(), 
                              done=False))
    await store.add(TaskInput(name="A second entry", 
                              priority=2, 
                              due_date=date.today(), 
                              done=True))
    filter = TaskFilter()

    entries = await store.filter(filter)

    assert len(entries) >= 2


@pytest.mark.asyncio
async def test_filter_with_filter_for_done_gives_only_done_entries(with_db):
    store = DataStoreDb(with_db)
    await store.add(TaskInput(name="counter", 
                              priority=1, 
                              due_date=date.today(), 
                              done=False))
    await store.add(TaskInput(name="A second entry", 
                              priority=2, 
                              due_date=date.today(), 
                              done=True))
    filter = TaskFilter()
    filter.done = True

    entries = await store.filter(filter)

    assert len(entries) >= 1

    for entry in entries:
        assert entry.done == True


@pytest.mark.asyncio
async def test_filter_with_filter_for_name_gives_only_matching_entries(with_db):
    store = DataStoreDb(with_db)
    await store.add(TaskInput(name="counter", 
                              priority=1, 
                              due_date=date.today(), 
                              done=False))
    await store.add(TaskInput(name="Create a filter", 
                              priority=2, 
                              due_date=date.today(), 
                              done=True))
    filter = TaskFilter()
    filter.name = "Create a filter"

    entries = await store.filter(filter)

    assert len(entries) == 1
    assert entries[0].name == "Create a filter"


@pytest.mark.asyncio
async def test_filter_with_search_gives_only_matching_entries(with_db):
    store = DataStoreDb(with_db)
    await store.add(TaskInput(name="Search for item", 
                              priority=1, 
                              due_date=date.today(), 
                              done=False))
    await store.add(TaskInput(name="Create a Search", 
                              priority=2, 
                              due_date=date.today(), 
                              done=True))
    filter = TaskFilter()
    filter.search = "Search"

    entries = await store.filter(filter)

    assert len(entries) >= 2
    for entry in entries:
        assert "Search" in entry.name

The tests currently fail, then we have no TaskFilter class and no filter() method on our data store.

Create the TaskFilter

The main component of FastAPI Filter is our filter object. We need to specify what fields we want to filter on and how we want to order the results.

from datetime import date
from typing import Optional
from fastapi_filter.contrib.sqlalchemy import Filter

from ..data.entities import Task

class TaskFilter(Filter):
    id: Optional[int] = None
    name: Optional[str] = None
    priority: Optional[list[int]] = None
    due_date__lte: Optional[date] = None
    done: Optional[bool] = None
    order_by: list[str] = ["name"]
    search: Optional[str] = None

    class Constants(Filter.Constants):
        model = Task
        search_model_fields = ["name"]

If we specify our field as it is, we can filter for an exact match. If we want to use an operator from the list above, we can use our field name, add two _ and then use the operator as we did with due_date__lte.

The order_by uses the field name in ascending order if we do not overwrite it. That way we always get a sorted result.

The search field allows us to search for a part of the name. If we want to include other fields, we can add them to the list of search_model_fields.

As a final point, make sure that you set the default value to None. That way all the filters are optional. If you omit the None value, you turn them into required fields.

Extend the data store

In our data store, we add a new filter() method that uses the TaskFilter to expand the query with a WHERE and an ORDER BY clause:

from ..models.task_filter import TaskFilter

...

    async def filter(self, filter: TaskFilter):
        async with self.db() as session:
            query = select(Task)
            query = filter.filter(query)
            query = filter.sort(query)
            result = await session.execute(query)
            results = []

            for entry in result:
                results.append(self.__to_output(*entry))

            return results

I had to add an * in front of entry to convert it to our TaskOutput object. Otherwise, I got an error about a missing id attribute.

Our new tests for the data store should now work. If that is the case, we can go one layer up to FastAPI.

Change the endpoint tests

With our new FastAPI Filter package, the syntax to filter the tasks changes a bit. We can change the syntax of the existing tests that filter the tasks and add a few new tests to check that the order_by works as expected:

@pytest.mark.asyncio
async def test_show_all_tasks():
    await prepare_task("a first task")
    await prepare_task("a second task")
    await prepare_task("a third task")

    response = client.get("/api/todo")

    assert response.status_code == 200
    tasks = response.json()
    assert len(tasks) >= 3


@pytest.mark.asyncio
async def test_show_all_tasks_that_are_not_done():
    await prepare_task("a finished task", done=True)
    await prepare_task("an open task", done=False)

    response = client.get("/api/todo?done=false")

    assert response.status_code == 200
    tasks = response.json()
    done = [task for task in tasks if task['done'] == True]
    assert len(done) == 0


@pytest.mark.asyncio
async def test_show_all_tasks_that_are_due_within_five_days():
    await prepare_task("in 10 days", due_date=date.today() + timedelta(days=10))
    await prepare_task("in 4 days", due_date=date.today() + timedelta(days=4))

    response = client.get(f"/api/todo?done=false&due_date__lte={date.today() + timedelta(days=5)}")

    assert response.status_code == 200
    tasks = response.json()
    assert len(tasks) >= 1
    larger = [task for task in tasks if date.fromisoformat(task['due_date']) > date.today() + timedelta(days=5)]
    assert len(larger) == 0


@pytest.mark.asyncio
async def test_show_all_tasks_that_match_search_criteria_sorted_by_name():
    await prepare_task("485960 A", due_date=date.today())
    await prepare_task("485960 B", due_date=date.today())
    await prepare_task("485960 C", due_date=date.today())

    response = client.get(f"/api/todo?search=485960")

    assert response.status_code == 200
    tasks = response.json()
    assert len(tasks) == 3
    assert tasks[0]["name"] == "485960 A"
    assert tasks[1]["name"] == "485960 B"
    assert tasks[2]["name"] == "485960 C"


@pytest.mark.asyncio
async def test_show_all_tasks_that_match_search_criteria_sorted_by_name_descending():
    await prepare_task("5780383 A", due_date=date.today())
    await prepare_task("5780383 B", due_date=date.today())
    await prepare_task("5780383 C", due_date=date.today())

    response = client.get(f"/api/todo?search=5780383&order_by=-name")

    assert response.status_code == 200
    tasks = response.json()
    assert len(tasks) == 3
    assert tasks[0]["name"] == "5780383 C"
    assert tasks[1]["name"] == "5780383 B"
    assert tasks[2]["name"] == "5780383 A"

The endpoint tests that filter the tasks should all fail. To make them pass we need to wire-up the filters in our endpoint.

Change the /tasks endpoint

Our show_all_tasks() method needs to use our TaskFilter as a dependency and pass that filter to our data store. Our new Filter package comes with the FilterDepends keyword, that does all the magic to translate the query string to our TaskFilter object:

from fastapi_filter import FilterDepends
from ..models.task_filter import TaskFilter

@router.get("/")
async def show_all_tasks(filter: TaskFilter = FilterDepends(TaskFilter), 
                         db: DataStoreDb = Depends(get_db)) -> List[TaskOutput]:
    result = await db.filter(filter)
    return result

We can now run all our tests and they should pass.

Updated documentation

If we start our application and head to the /docs endpoint, we find the filter fields as part of the documentation:

Our filter fields are now part of the OpenAPI specification.