Todo CRUD API

The Todo resource API will need to provide a RESTFul CRUD interface for the frontend to Create, Read, Update, and Delete Todos. This API is best implemented as a Blueprint, placed in backend/src/backend/blueprints/todos.py, putting all the features e.g. authentication, rate-limiting, models etc... that we've setup together.

Creating the blueprint

The blueprint itself can be created with the following code in backend/src/backend/blueprints/todos.py,

from quart import Blueprint

blueprint = Blueprint("todos", __name__)

and activated by adding the following to backend/src/backend/run.py,

from backend.blueprints.todos import blueprint as todos_blueprint

def create_app() -> Quart:
    ...
    app.register_blueprint(todos_blueprint)

Creating a todo

For a RESTFul API the todo creation route should be POST, expecting the todo data and returning the complete todo with a 201 status code on success, we'll assume that a fast real user can create no more than 10 todos in 10 seconds on average. The following should be added to backend/src/backend/blueprints/todos.py,

from datetime import timedelta

from quart import current_app
from quart_auth import current_user, login_required
from quart_schema import validate_request, validate_response
from quart_rate_limiter import rate_limit

from backend.models.todo import insert_todo, Todo, TodoData

@blueprint.post("/todos/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_request(TodoData)
@validate_response(Todo, 201)
async def post_todo(data: TodoData) -> tuple[Todo, int]:
    """Create a new Todo.

    This allows todos to be created and stored.
    """
    todo = await insert_todo(current_app.db, data, int(current_user.auth_id))
    return todo, 201

Reading a todo

For a RESTFul API the read route should be GET, returning a todo on success (we'll use the same rate limit) and a 404 if the todo does not exist. The following should be added to backend/src/backend/blueprints/todos.py,

from backend.lib.api_error import APIError
from backend.models.todo import select_todo

@blueprint.get("/todos/<int:id>/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_response(Todo)
async def get_todo(id: int) -> Todo:
    """Get a todo.

    Fetch a Todo by its ID.
    """
    todo = await select_todo(current_app.db, id, int(current_user.auth_id))
    if todo is None:
        raise APIError(404, "NOT_FOUND")
    else:
        return todo

Reading the todos

For a RESTFul API the read route should be GET, returning a list of todos on success (we'll use the same rate limit). To demonstrate how to do server side filtering we'll provide a read route that can optionally filter on the complete todo attribute. The following should be added to backend/src/backend/blueprints/todos.py,

from dataclasses import dataclass
from typing import Optional

from quart_schema import validate_querystring

from models.todo import select_todos

@dataclass
class Todos:
    todos: list[Todo]

@dataclass
class TodoFilter:
    complete: Optional[bool] = None

@blueprint.get("/todos/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_response(Todos)
@validate_querystring(TodoFilter)
async def get_todos(query_args: TodoFilter) -> Todos:
    """Get the todos.

    Fetch all the Todos optionally based on the complete status.
    """
    todos = await select_todos(
        current_app.db, int(current_user.auth_id), query_args.complete
    )
    return Todos(todos=todos)

Updating a todo

For a RESTFul API the todo update route should be PUT, expecting the todo data and returning the complete todo on success (we'll use the same rate limit) and a 404 if the todo does not exist. The following should be added to backend/src/backend/blueprints/todos.py,

from models.todo import update_todo

@blueprint.put("/todos/<int:id>/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_request(TodoData)
@validate_response(Todo)
async def put_todo(id: int, data: TodoData) -> Todo:
    """Update the identified todo

    This allows the todo to be replaced with the request data.
    """
    todo = await update_todo(current_app.db, id, data, int(current_user.auth_id))
    if todo is None:
        raise APIError(404, "NOT_FOUND")
    else:
        return todo

Deleting a todo

For a RESTFul API the todo deletion route should be DELETE, returning 202 on success and if the todo does not exist (we'll use the same rate limit). The following should be added to backend/src/backend/blueprints/todos.py,

from quart import ResponseReturnValue

from models.todo import delete_todo

@blueprint.delete("/todos/<int:id>/")
@rate_limit(10, timedelta(seconds=10))
@login_required
async def todo_delete(id: int) -> ResponseReturnValue:
    """Delete the identified todo

    This will delete the todo.
    """
    await delete_todo(current_app.db, id, int(current_user.auth_id))
    return "", 202

Testing

We should test that these routes work as a user would expect, starting by testing that the get route returns the test data todo by adding the following to backend/tests/blueprints/test_todos.py,

import pytest
from quart import Quart

@pytest.mark.asyncio
async def test_get_todo(app: Quart) -> None:
    test_client = app.test_client()
    async with test_client.authenticated("1"):
        response = await test_client.get("/todos/1/")
        todo = await response.get_json()
        assert todo["task"] == "Task"

Next we can test creating, updating, retreiving and then deleting a todo i.e. the full user flow, by adding the following to backend/tests/blueprints/test_todos.py.

@pytest.mark.asyncio
async def test_todo_flow(app: Quart) -> None:
    test_client = app.test_client()
    async with test_client.authenticated("1"):
        response = await test_client.get("/todos/")
        todos = (await response.get_json())["todos"]
        response = await test_client.post(
            "/todos/",
            json={"complete": False, "due": None, "task": "New Todo"},
        )
        todo_id = (await response.get_json())["id"]
        response = await test_client.get("/todos/")
        new_todos = (await response.get_json())["todos"]
        assert len(new_todos) == len(todos) + 1
        await test_client.put(
            f"/todos/{todo_id}/",
            json={"complete": True, "due": None, "task": "New Todo"},
        )
        response = await test_client.get(f"/todos/{todo_id}/")
        todo = await response.get_json()
        assert todo["complete"]
        await test_client.delete(f"/todos/{todo_id}/")
        response = await test_client.get("/todos/")
        final_todos = (await response.get_json())["todos"]
        assert len(final_todos) == len(todos)