Compare commits
No commits in common. "d92982479c27af674303f6bff3116b5af50cd96a" and "21fd769fa995448fe827510e5129983a5da60008" have entirely different histories.
d92982479c
...
21fd769fa9
@ -1,122 +0,0 @@
|
|||||||
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
|
|
@ -22,7 +22,6 @@ from quart_schema import QuartSchema, RequestSchemaValidationError
|
|||||||
from backend.blueprints.control import blueprint as control_blueprint
|
from backend.blueprints.control import blueprint as control_blueprint
|
||||||
from backend.blueprints.members import blueprint as members_blueprint
|
from backend.blueprints.members import blueprint as members_blueprint
|
||||||
from backend.blueprints.sessions import blueprint as sessions_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
|
# For making sure error responses are in JSON format
|
||||||
from backend.lib.api_error import APIError
|
from backend.lib.api_error import APIError
|
||||||
@ -43,7 +42,6 @@ logging.basicConfig(level=logging.INFO)
|
|||||||
app.register_blueprint(control_blueprint)
|
app.register_blueprint(control_blueprint)
|
||||||
app.register_blueprint(sessions_blueprint)
|
app.register_blueprint(sessions_blueprint)
|
||||||
app.register_blueprint(members_blueprint)
|
app.register_blueprint(members_blueprint)
|
||||||
app.register_blueprint(todos_blueprint)
|
|
||||||
|
|
||||||
|
|
||||||
# rate limiting
|
# rate limiting
|
||||||
|
@ -1,54 +0,0 @@
|
|||||||
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
|
|
Loading…
x
Reference in New Issue
Block a user