Skip to content

#221: Refactor the FastAPI To-Do Application

Our minimalistic to-do application has collected some code smells. In this post we go through them and refactor our code to keep the application in a maintainable state for the upcoming changes.

Remove the duplication in our tests

When it comes to creating tasks, we only care in the test_create_task() function about all the necessary details that this operation involves. For all the other tests, we create tasks to have something with which we can work. Instead of repeating the details in every test, we can extract the code and put it in its own prepare_task() function:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
def prepare_task(name, priority=4, due_date=None, done=False):
    if due_date == None:
        due_date = date.today() + timedelta(days=1)

    data = {
        "name": name,
        "priority": priority,
        "due_date": str(due_date),
        "done": done
    }

    prepare_response = client.post("/api/todo/", json=data)
    assert prepare_response.status_code == 200
    return prepare_response.json()['id']

This allows us to remove the duplication in our tests and replace it with a call to our new method:

1
2
3
4
5
6
7
8
def test_show_task():
    name = "A second task"
    id = prepare_task(name)

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

This is much smaller than what we had before:

 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']

Refactor the API

The data store is only a mock that we replace in the future with a real database. But even with that in mind it is a bit unlucky that we have the logic for the data access copied throughout our endpoints.

If we encapsulate all the behaviour that interacts with our mock storage into its own class, we not only have smaller endpoints, but we can make sure that we do not reassign already used Ids for free. But before we can profit from the new possibilities, we must do the work.

To shorten this post, here are all the tests we need for our data store. You would write one test, write the implementation, and then repeat with the next test.

  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
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
from ..data.datastore import DataStore
from ..models.todo import TaskOutput, TaskInput
from datetime import date, timedelta
import pytest

def test_created_store_is_empty():
    store = DataStore()

    data = store.all()

    assert data == []

def test_can_add_entry():
    entry = TaskInput(name="a simple task", 
                      priority=1, 
                      due_date=date.today(), 
                      done=False)
    store = DataStore()

    data = store.add(entry)

    assert data.name == "a simple task"
    assert data.priority == 1
    assert data.due_date == date.today()
    assert data.done == False
    assert data.created_at == date.today()
    assert data.id == 1


def test_can_add_multiple_entries():
    entry_a = TaskInput(name="a simple task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    entry_b = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    store = DataStore()

    data_a = store.add(entry_a)
    data_b = store.add(entry_b)

    assert data_a.id < data_b.id


def test_can_get_specific_entry_back():
    entry_a = TaskInput(name="a simple task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    entry_b = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    store = DataStore()
    store.add(entry_a)
    store.add(entry_b)

    entry = store.get(2)

    assert entry.name == "b simple task"


def test_missing_entry_gets_None_back():
    store = DataStore()

    entry = store.get(2)

    assert entry == None


def test_can_get_all_entrries_back():
    entry_a = TaskInput(name="a simple task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    entry_b = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    entry_c = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    store = DataStore()
    store.add(entry_a)
    store.add(entry_b)
    store.add(entry_c)

    entries = store.all()

    assert len(entries) == 3


def test_can_delete_entry():
    entry_a = TaskInput(name="a simple task", 
                        priority=1, 
                        due_date=date.today(), 
                        done=False)
    entry_b = TaskInput(name="b simple task", 
                        priority=2, 
                        due_date=date.today(), 
                        done=False)
    store = DataStore()
    store.add(entry_a)
    store.add(entry_b)

    store.delete(2)

    entries = store.all()
    assert len(entries) == 1
    assert entries[0].id == 1


def test_can_update_entry():
    old = TaskInput(name="a simple task", 
                    priority=1, 
                    due_date=date.today(), 
                    done=False)
    store = DataStore()
    store.add(old)

    new = TaskInput(name="b simple task", 
                    priority=2, 
                    due_date=date.today() + timedelta(days=2), 
                    done=True)
    store.update(1, new)

    entry = store.get(1)
    assert entry.name == "b simple task"
    assert entry.priority == 2
    assert entry.due_date == date.today() + timedelta(days=2)
    assert entry.done == True


def test_non_existing_entry_cannot_be_updated():
    store = DataStore()

    new = TaskInput(name="b simple task", 
                    priority=2, 
                    due_date=date.today() + timedelta(days=2), 
                    done=True)
    with pytest.raises(ValueError) as e_info:
        store.update(123, new)
    assert str(e_info.value) == "no taks known with id '123'"

With our tests we can come up with a class like this DataStore inside the data folder:

 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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from ..models.todo import TaskInput, TaskOutput
from datetime import date

class DataStore:
    def __init__(self):
        self._data = []
        self._id_next = 1


    def add(self, entry: TaskInput): 
        extended_entry = TaskOutput(id=self._id_next, 
                                    created_at=date.today(), 
                                    **dict(entry))
        self._data.append(extended_entry)
        self._id_next += 1
        return extended_entry


    def all(self):
        return self._data


    def get(self, id) -> TaskOutput:
        result = [item for item in self._data if item.id == id]
        if len(result) > 0:
            return result[0]
        else:
            return None


    def delete(self, id):
        entry = self.get(id)
        if entry:
            self._data.remove(entry)


    def update(self, id: int, update: TaskInput):
        entry = self.get(id)
        if entry: 
            entry.name = update.name
            entry.priority = update.priority
            entry.due_date = update.due_date
            entry.done = update.done
            return entry
        else: 
            raise ValueError(f"no taks known with id '{id}'")

Again, do not forget to create an empty __init__.py inside the data folder next to the datastore.py.

We can now rewrite our FastAPI application and then run the tests to check that everything still works as expected:

 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
27
28
29
30
31
32
33
34
35
36
37
from fastapi import FastAPI, HTTPException
from .models.todo import *
from .data.datastore import DataStore

app = FastAPI()

db = DataStore()

@app.post("/api/todo")
async def create_task(task: TaskInput):
    result = db.add(task)
    return result


@app.get("/api/todo/{id}")
async def show_task(id: int):
    result = db.get(id)

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



@app.put("/api/todo/{id}")
async def update_task(id: int, task: TaskInput):
    try:
        result = db.update(id, task)
        return result
    except ValueError:
        raise HTTPException(status_code=404, detail="Task not found")


@app.delete("/api/todo/{id}")
async def delete_task(id: int):
    db.delete(id)

After extracting all the logic to store and retrieve data, our API is reduced to a thin layer over our data store. The more logic we can take out of our API, the better it is. Then it is much easier to reuse Python classes than API endpoints.

Next

The refactorings help us to keep our code in a better state. We will notice the improvement as soon as we need to add new features, and that usually happens sooner than later.

While checking if the Swagger documentation still works, I noticed two missing features that I like to add next week so that we can put our new structure to a test.