Skip to content

#220: Manage To-Dos With FastAPI

After covering the basics of FastAPI, it is now time to do something more elaborate. For that purpose, we create a minimalistic to-do task tracker that let us work against an in-memory data store. To save us the time to constantly click in the browser, we will do this exercise by first writing a test and then create the implementation using the test-first approach.

The Pydantic models

Four our little application we split the input models from those we send back to the API client. That allows us to be specific on what data we expect from the users and show the difference to what we keep inside the API.

Our input model has these four fields:

1
2
3
4
5
6
7
8
from datetime import date
from pydantic import BaseModel

class TaskInput(BaseModel):
    name: str
    priority: int
    due_date: date
    done: bool

What we send back to the client contains a bit more details:

1
2
3
4
5
6
7
class TaskOutput(BaseModel):
    id: int
    name: str
    priority: int
    due_date: date
    done: bool
    created_at: date

A failing test to start

We want to work with our API by sending a task to the API and get the data with some additional details back:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from datetime import date, timedelta
from fastapi.testclient import TestClient
from ..main import app

client = TestClient(app)


def test_create_task():
    data = {
        "name": "A first task",
        "priority": 5,
        "due_date": str(date.today() + timedelta(days=1)),
        "done": False
    }

    response = client.post("/api/todo/", json=data)
    assert response.status_code == 200
    result = response.json()
    assert result['id'] > 0
    assert result['done'] == False
    assert result['created_at'] == str(date.today())
    assert result['name'] == data['name']
    assert result['priority'] == data['priority']
    assert result['due_date'] == data['due_date']

Our test currently fails and will so for the next few steps that we need to get done so that we get enough code to run our API.

A new API application

Our minimalistic first attempt to make our test pass can look like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from fastapi import FastAPI
from .models.todo import *

app = FastAPI()


@app.post("/api/todo")
async def create_task(task: TaskInput):
    pos = 1
    result = TaskOutput(id=pos, 
                        name=task.name, 
                        priority=task.priority, 
                        due_date=task.due_date,
                        done=task.done, 
                        created_at=date.today())
    return result

Make sure that you create the empty __init__.py files inside the models and tests folder and put one next to main.py. Otherwise, you end up with an error message like this one:

ImportError while importing test module ‘….\tests\test_todo.py’.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
…
tests\test_todo.py:3: in
from ..main import app
E ImportError: attempted relative import with no known parent package

If we now run our test, we should see it passing.

A failing test to add more functionality

With our first test passing, we can go for the next feature: we want to get the details of a newly created task. With this test we can check if our API gives us the expected data:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
def test_show_task():
    data = {
        "name": "A second task",
        "priority": 4,
        "due_date": str(date.today() + timedelta(days=1))
    }

    prepare_response = client.post("/api/todo/", json=data)
    assert prepare_response.status_code == 200

    id = prepare_response.json()['id']
    response = client.get(f"/api/todo/{id}")
    assert response.status_code == 200
    details = response.json()
    assert details['name'] == data['name']

This test also fails at first, then currently there is no endpoint and no way to store the data between the requests to get the result we expect.

Add a mock data store

To make our second test pass, we need to keep the data around for two (or more) requests. We can do that with a variable in the API that works as a mock data store:

1
2
3
app = FastAPI()

db = []

We need to change the create_task() method to put the newly created task into our data store and then read from there when someone asks for the item in the show_task() method:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
@app.post("/api/todo")
async def create_task(task: TaskInput):
    pos = len(db) + 1
    result = TaskOutput(id=pos, 
                        name=task.name, 
                        priority=task.priority, 
                        due_date=task.due_date,
                        done=False, 
                        created_at=date.today())
    db.append(result)
    return result


@app.get("/api/todo/{id}")
async def show_task(id: int):
    result = [item for item in db if item.id == id]
    return result[0]

We can now run our two tests, and both will pass.

Add some error handling

What happens if we ask for a task that does not exist? Will our application crash or will it produce a useful message? Let us write a test to find out what happens:

1
2
3
4
def test_show_task_where_task_is_unknown():
    response = client.get(f"/api/todo/-1")
    assert response.status_code == 404
    assert response.json()['detail'] == "Task not found"

Our current implementation will throw an exception because there is no element at position 0 in an empty list. We can make our test pass by adding this check:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from fastapi import FastAPI, HTTPException

@app.get("/api/todo/{id}")
async def show_task(id: int):
    result = [item for item in db if item.id == id]

    if len(result) > 0:
        return result[0]
    else:
        raise HTTPException(status_code=404, detail="Task not found")

The HTTPException allows us to specify a status code and set a message that we can send to the client. With that addition, our API can now handle the case where the requested task does not exist.

Update a task

We currently can create a task and get its details. When we want to update our task, we need a new test to drive our functionality:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def test_update_task():
    data = {
        "name": "A 2nd task",
        "priority": 4,
        "due_date": str(date.today() + timedelta(days=1)),
        "done": False
    }

    prepare_response = client.post("/api/todo/", json=data)
    assert prepare_response.status_code == 200

    id = prepare_response.json()['id']

    update = {
        "name": "An updated task",
        "priority": 5,
        "due_date": str(date.today() + timedelta(days=2)),
        "done": False
    }

    response = client.put(f"/api/todo/{id}", json=update)
    assert response.status_code == 200
    assert response.json()['name'] == "An updated task"

    check = client.get(f"/api/todo/{id}")
    assert check.json()['name'] == "An updated task"

In our API, we need to find the requested task, update the fields, and then return the changed task:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
@app.put("/api/todo/{id}")
async def update_task(id: int, task: TaskInput):
    result = [item for item in db if item.id == id]

    if len(result) == 1:
        current = result[0]
        current.name = task.name
        current.priority = task.priority
        current.due_date = task.due_date

        return current
    else:
        raise HTTPException(status_code=404, detail="Task not found")

Attention: Since we work on the element of the list and not on a copy, we do not need to update the list. That will need to change when we use a database.

We can now run our tests and they all should pass. We can spot an opportunity for refactoring when it comes to the duplicated code to create an event to have some data to work on in our tests. We write that down to a notepad but postpone that refactoring for a moment.

Delete a task

When we want to delete a task, we start once more with a failing test:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
def test_delete_task():
    data = {
        "name": "A 2nd task",
        "priority": 4,
        "due_date": str(date.today() + timedelta(days=1)),
        "done": False
    }

    prepare_response = client.post("/api/todo/", json=data)
    assert prepare_response.status_code == 200

    id = prepare_response.json()['id']
    response = client.delete(f"/api/todo/{id}")
    assert response.status_code == 200

    check = client.get(f"/api/todo/{id}")
    assert check.status_code == 404

We can now implement the delete_task() method with this code:

1
2
3
4
5
6
@app.delete("/api/todo/{id}")
async def delete_task(id: int):
    result = [item for item in db if item.id == id]

    if len(result) == 1:
        db.remove(result[0])

When we do not specify a response, we get the default response with status code 200. That currently is good enough for our API. We will return to status codes in a future post.

Run with uvcorn

If we run our to-do app like we did the minimalistic example, we run into this error:

File “…\minimal_todo\main.py”, line 3, in
from .models.todo import *
ImportError: attempted relative import with no known parent package

Instead, we need to go one folder up of our main.py and run uvicorn with this command:

1
uvicorn minimal_todo.main:app --reload

If we go to http://127.0.0.1:8000/docs, we should see our Swagger documentation:

Our To-Do API now runs with uvicorn and shows us the documentation for our endpoints

Next

With our test-first approach we got our ToDo API in a state that it supports the basic CRUD operations. The tests help us to make sure that the application keeps working when we add more features or change the behaviour. However, there is now a lot of redundancy in our tests that we can refactor next week.