Feat(API): CRUD functionalities with todo items

- wrote tests for todo items workflow

backend/src/backend/blueprints/todos.py

@@ -0,0 +1,122 @@
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import cast
+
+from quart import Blueprint, ResponseReturnValue, g
+from quart_auth import current_user, login_required
+from quart_rate_limiter import rate_limit
+from quart_schema import validate_querystring, validate_request, validate_response
+
+from backend.lib.api_error import APIError
+from backend.models.todo import (
+    Todo,
+    delete_todo,
+    insert_todo,
+    select_todo,
+    select_todos,
+    update_todo,
+)
+
+blueprint = Blueprint("todos", __name__)
+
+
+@dataclass
+class TodoData:
+    complete: bool
+    due: datetime | None
+    task: str
+
+
+@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(
+        g.connection,
+        int(cast(str, current_user.auth_id)),
+        data.task,
+        data.complete,
+        data.due,
+    )
+    return todo, 201
+
+
+@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(g.connection, id, int(cast(str, current_user.auth_id)))
+    if todo is None:
+        raise APIError(404, "NOT_FOUND")
+    else:
+        return todo
+
+
+@dataclass
+class Todos:
+    todos: list[Todo]
+
+
+@dataclass
+class TodoFilter:
+    complete: bool | None = 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(
+        g.connection,
+        int(cast(str, current_user.auth_id)),
+        query_args.complete,
+    )
+    return Todos(todos=todos)
+
+
+@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(
+        g.connection,
+        id,
+        int(cast(str, current_user.auth_id)),
+        data.task,
+        data.complete,
+        data.due,
+    )
+    if todo is None:
+        raise APIError(404, "NOT_FOUND")
+    else:
+        return 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(g.connection, id, int(cast(str, current_user.auth_id)))
+    return "", 202

backend/src/backend/run.py

@@ -22,6 +22,7 @@ from quart_schema import QuartSchema, RequestSchemaValidationError
 from backend.blueprints.control import blueprint as control_blueprint
 from backend.blueprints.members import blueprint as members_blueprint
 from backend.blueprints.sessions import blueprint as sessions_blueprint
+from backend.blueprints.todos import blueprint as todos_blueprint
 
 # For making sure error responses are in JSON format
 from backend.lib.api_error import APIError
@@ -42,6 +43,7 @@ logging.basicConfig(level=logging.INFO)
 app.register_blueprint(control_blueprint)
 app.register_blueprint(sessions_blueprint)
 app.register_blueprint(members_blueprint)
+app.register_blueprint(todos_blueprint)
 
 
 # rate limiting

backend/tests/blueprints/test_todos.py

@@ -0,0 +1,54 @@
+from quart import Quart
+
+
+async def test_post_todo(app: Quart) -> None:
+    test_client = app.test_client()
+    async with test_client.authenticated("1"):  # type: ignore
+        response = await test_client.post(
+            "/todos/",
+            json={"complete": False, "due": None, "task": "Test task"},
+        )
+        assert response.status_code == 201
+        assert (await response.get_json())["id"] > 0
+
+
+async def test_get_todo(app: Quart) -> None:
+    test_client = app.test_client()
+    async with test_client.authenticated("1"):  # type: ignore
+        response = await test_client.get("/todos/1/")
+        assert response.status_code == 200
+        assert (await response.get_json())["task"] == "Test Task"
+
+
+async def test_put_todo(app: Quart) -> None:
+    test_client = app.test_client()
+    async with test_client.authenticated("1"):  # type: ignore
+        response = await test_client.post(
+            "/todos/",
+            json={"complete": False, "due": None, "task": "Test task"},
+        )
+        todo_id = (await response.get_json())["id"]
+
+        response = await test_client.put(
+            f"/todos/{todo_id}/",
+            json={"complete": False, "due": None, "task": "Updated"},
+        )
+        assert (await response.get_json())["task"] == "Updated"
+
+        response = await test_client.get(f"/todos/{todo_id}/")
+        assert (await response.get_json())["task"] == "Updated"
+
+
+async def test_delete_todo(app: Quart) -> None:
+    test_client = app.test_client()
+    async with test_client.authenticated("1"):  # type: ignore
+        response = await test_client.post(
+            "/todos/",
+            json={"complete": False, "due": None, "task": "Test task"},
+        )
+        todo_id = (await response.get_json())["id"]
+
+        await test_client.delete(f"/todos/{todo_id}/")
+
+        response = await test_client.get(f"/todos/{todo_id}/")
+        assert response.status_code == 404