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.
This post is part of my journey to learn Python. You find the code for this post in my PythonFriday repository on GitHub.
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:
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.
3 thoughts on “Python Friday #220: Manage To-Dos With FastAPI”