diff --git a/core/models.py b/core/models.py index 62587a6..dde8f65 100644 --- a/core/models.py +++ b/core/models.py @@ -266,12 +266,28 @@ def get_task(conn: sqlite3.Connection, id: str) -> dict | None: return _row_to_dict(row) +VALID_TASK_SORT_FIELDS = frozenset({ + "updated_at", "created_at", "priority", "status", "title", "id", +}) + + def list_tasks( conn: sqlite3.Connection, project_id: str | None = None, status: str | None = None, + limit: int | None = None, + sort: str = "updated_at", + sort_dir: str = "desc", ) -> list[dict]: - """List tasks with optional project/status filters.""" + """List tasks with optional project/status filters, limit, and sort. + + sort: column name (validated against VALID_TASK_SORT_FIELDS, default 'updated_at') + sort_dir: 'asc' or 'desc' (default 'desc') + """ + # Validate sort field to prevent SQL injection + sort_col = sort if sort in VALID_TASK_SORT_FIELDS else "updated_at" + sort_direction = "DESC" if sort_dir.lower() != "asc" else "ASC" + query = "SELECT * FROM tasks WHERE 1=1" params: list = [] if project_id: @@ -280,7 +296,10 @@ def list_tasks( if status: query += " AND status = ?" params.append(status) - query += " ORDER BY priority, created_at" + query += f" ORDER BY {sort_col} {sort_direction}" + if limit is not None: + query += " LIMIT ?" + params.append(limit) return _rows_to_list(conn.execute(query, params).fetchall()) diff --git a/web/api.py b/web/api.py index adc169d..c788acd 100644 --- a/web/api.py +++ b/web/api.py @@ -654,6 +654,28 @@ def start_project_phase(project_id: str): # Tasks # --------------------------------------------------------------------------- +@app.get("/api/tasks") +def list_tasks( + status: str | None = Query(default=None), + limit: int = Query(default=20, ge=1, le=500), + sort: str = Query(default="updated_at"), + project_id: str | None = Query(default=None), +): + """List tasks with optional filters. sort defaults to updated_at desc.""" + from core.models import VALID_TASK_SORT_FIELDS + conn = get_conn() + tasks = models.list_tasks( + conn, + project_id=project_id, + status=status, + limit=limit, + sort=sort if sort in VALID_TASK_SORT_FIELDS else "updated_at", + sort_dir="desc", + ) + conn.close() + return tasks + + @app.get("/api/tasks/{task_id}") def get_task(task_id: str): conn = get_conn()