#222: Filter the Tasks in the FastAPI Application

The to-do application started nicely and with the refactoring we got the most obvious code smells fixed. While checking the API in uvicorn, I noticed three missing features that we are going to implement in this post.

Show all existing tasks

So far, we can create, read, update, and delete a single task. But we cannot get back all the tasks currently in our application. Once more we start with a failing test to drive our development:

def test_show_all_tasks():
    prepare_task("a first task")
    prepare_task("a second task")
    prepare_task("a third task")

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

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

With this test in place, we can create our endpoint:

@app.get("/api/todo")
async def show_all_tasks():
    result = db.all()
    return result

Filter the tasks

The longer we run our application, the more tasks it will collect. It would be nice if we could filter our tasks to only get back the ones that are not done yet. Our test for this new feature can look like this:

def test_show_all_tasks_that_are_not_done():
    prepare_task("a finished task", done=True)
    prepare_task("an open task", done=False)

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

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

Filters are a topic that usually starts small and then grows. We therefore better look at a solution that we can extend should the need arise. With the feature of Dependency Injection in FastAPI we can create a function that accepts our filter, wraps it into the right data type and then hands it as a dictionary to our endpoint:

from fastapi import Depends, FastAPI, HTTPException

async def filter_parameters(q: str | None = None, include_done: bool = True):
    return {"q": q, "include_done": include_done}


@app.get("/api/todo")
async def show_all_tasks(filter: Annotated[dict, Depends(filter_parameters)]):
    result = db.all()

    if not filter["include_done"]:
        result = [item for item in result if item.done == False ]

    return result

For the time being, we keep the filter logic in the endpoint. When we move to SQLAlchemy, we can use a filter plug-in that allows us to filter inside the database and so reduce the workload for our application.

With our first filter in place, another idea pops up. Would it not be nice to get back only tasks that are due up to a certain date? Let us write the test so that we can see what we expect:

def test_show_all_tasks_that_are_due_within_five_days():
    prepare_task("in 10 days", due_date=date.today() + timedelta(days=10))

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

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

We can add another parameter to our filter function and then use the filter inside our endpoint. Since Pydantic and FastAPI convert our due_before parameter into a date object, we do not need to convert it:

async def filter_parameters(q: str | None = None, 
                            include_done: bool = True, 
                            due_before: date = date.today() + timedelta(days=365)):
    return {"q": q, "include_done": include_done, "due_before": due_before }


@app.get("/api/todo")
async def show_all_tasks(filter: Annotated[dict, Depends(filter_parameters)]):
    result = db.all()

    if not filter["include_done"]:
        result = [item for item in result if item.done == False ]

    result = [item for item in result if item.due_date <= filter["due_before"] ]

    return result

Remove the error message at the / route

If we open our API in a browser without specifying a route, we get this error message:

While our API works, this is not a nice way to greet our users. Before we add another endpoint, we write a small test to drive the behaviour:

def test_main_page_shows_info_message():
    response = client.get("/")

    assert response.status_code == 200
    assert response.json()['message'] == "The minimalistic ToDo API"

We can write this show_root() method to make the test pass:

@app.get("/")
async def main():
    return {'message':'The minimalistic ToDo API'}

If we now open the API without a specific route, we get this message instead of an error:

Next

With the addition of the above features, we now have everything in place to work with our to-do’s. However, as always there is another thing that pops up as soon as a real user tries to work with the API: We do not have any useful data validation. Next week we improve our models to be more specific on what we accept.