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.
This post is part of my journey to learn Python. You find the code for this post in my PythonFriday repository on GitHub.
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.
2 thoughts on “Python Friday #221: Refactor the FastAPI To-Do Application”