Extended api (#18)

* small readme corrections

* extended the api

* fix import sorting

* fix & improve tests

* fix formatting

* reduce code duplication

* format fix

* fix typing issue

Author: brandstaetter

.isort.cfg

@@ -20,6 +20,10 @@ Backend REST API for the Task Management application.
 ```powershell
 (Invoke-WebRequest -Uri https://install.python-poetry.org -UseBasicParsing).Content | py -
 ```
+or
+```bash
+curl -sSL https://install.python-poetry.org | python3 -
+```
 
 2. Install dependencies:
 ```powershell
@@ -81,7 +85,7 @@ poetry run pytest --cov
 
 ### USB Thermal Printer
 - Supports ESC/POS compatible printers
-- Configure vendor_id and product_id in `config/printers.ini`
+- Configure USB vendor_id and product_id in `config/printers.ini`
 - Prints task details with QR code for quick access
 - Default configuration for ZJ-5870 printer included
 

README.md

@@ -66,7 +66,7 @@ force_grid_wrap = 0
 use_parentheses = true
 ensure_newline_before_comments = true
 line_length = 88
-skip = ["venv", ".venv", ".git", "__pycache__"]
+skip = ["venv", ".venv", ".git", "__pycache__", "output", "dist", "build", ".pytest_cache", ".coverage", "htmlcov"]
 known_first_party = ["taskmanagement_app"]
 sections = ["FUTURE", "STDLIB", "THIRDPARTY", "FIRSTPARTY", "LOCALFOLDER"]
 

openapi.json

@@ -1,11 +1,11 @@
 import logging
 from datetime import datetime, timezone
-from typing import List, Optional
+from typing import Any, List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from sqlalchemy.orm import Session
 
-from taskmanagement_app.core.exceptions import TaskStatusError
+from taskmanagement_app.core.exceptions import TaskNotFoundError, TaskStatusError
 from taskmanagement_app.core.printing.printer_factory import PrinterFactory
 from taskmanagement_app.crud.task import archive_task
 from taskmanagement_app.crud.task import complete_task as complete_task_crud
@@ -14,6 +14,7 @@
     get_task,
     get_tasks,
     read_random_task,
+    reset_task_to_todo,
 )
 from taskmanagement_app.crud.task import start_task as start_task_crud
 from taskmanagement_app.db.models.task import TaskModel, TaskState
@@ -250,3 +251,20 @@ async def trigger_maintenance(db: Session = Depends(get_db)) -> dict:
             status_code=500,
             detail=f"Error running maintenance job: {str(e)}",
         )
+
+
+@router.patch("/{task_id}/reset-to-todo", response_model=Task)
+def reset_task_to_todo_endpoint(task_id: int, db: Session = Depends(get_db)) -> Any:
+    """Reset a task to todo state and clear its progress timestamps."""
+    try:
+        return reset_task_to_todo(db=db, task_id=task_id)
+    except TaskNotFoundError:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Task with id {task_id} not found",
+        )
+    except TaskStatusError as e:
+        raise HTTPException(
+            status_code=400,
+            detail=str(e),
+        )

poetry.lock

@@ -25,3 +25,16 @@ class TaskStatusError(TaskManagementError):
     transition is encountered."""
 
     pass
+
+
+class TaskNotFoundError(TaskManagementError):
+    """Exception raised when a requested task cannot be found.
+
+    This exception is used when:
+    - Task ID does not exist in the database
+    - Task was deleted or is otherwise inaccessible
+    """
+
+    def __init__(self, task_id: int):
+        self.task_id = task_id
+        super().__init__(f"Task with id {task_id} not found")

pyproject.toml

@@ -5,7 +5,7 @@
 
 from sqlalchemy.orm import Session
 
-from taskmanagement_app.core.exceptions import TaskStatusError
+from taskmanagement_app.core.exceptions import TaskNotFoundError, TaskStatusError
 from taskmanagement_app.db.models.task import TaskModel, TaskState
 from taskmanagement_app.schemas.task import TaskCreate, TaskUpdate
 
@@ -249,10 +249,10 @@ def archive_task(db: Session, task_id: int) -> Optional[TaskModel]:
     if task:
         if task.state == TaskState.archived:
             raise TaskStatusError("Task is already archived")
-        elif task.state != TaskState.done:
+        elif task.state not in [TaskState.done, TaskState.todo]:
             raise TaskStatusError(
                 f"Cannot archive task in state {task.state}. "
-                "Task must be in 'done' state to be archived."
+                "Task must be in 'done' or 'todo' state to be archived."
             )
         task.state = TaskState.archived
         db.commit()
@@ -317,3 +317,29 @@ def read_random_task(db: Session) -> Optional[TaskModel]:
 
     # Use random.choices which allows weights
     return random.choices(tasks, weights=weights, k=1)[0]
+
+
+def reset_task_to_todo(db: Session, task_id: int) -> TaskModel:
+    """Reset a task to todo state and clear its progress timestamps.
+
+    Args:
+        db: Database session
+        task_id: ID of the task to reset
+
+    Returns:
+        Updated task
+
+    Raises:
+        TaskNotFoundError: If task doesn't exist
+    """
+    task = get_task(db, task_id)
+    if not task:
+        raise TaskNotFoundError(task_id)
+
+    task.state = TaskState.todo
+    task.started_at = None
+    task.completed_at = None
+
+    db.commit()
+    db.refresh(task)
+    return task