from quart import Blueprint, jsonify, request from fabledassistant.auth import login_required, get_current_user_id from fabledassistant.models.note import TaskPriority, TaskStatus from fabledassistant.routes.utils import not_found, parse_iso_date from fabledassistant.services.notes import ( create_note, delete_note, get_note, list_notes, update_note, ) tasks_bp = Blueprint("tasks", __name__, url_prefix="/api/tasks") @tasks_bp.route("", methods=["GET"]) @login_required async def list_tasks_route(): uid = get_current_user_id() q = request.args.get("q") tag = request.args.getlist("tag") status = request.args.get("status") priority = request.args.get("priority") sort = request.args.get("sort", "updated_at") order = request.args.get("order", "desc") limit = min(request.args.get("limit", 50, type=int), 500) offset = request.args.get("offset", 0, type=int) due_before = parse_iso_date(request.args.get("due_before"), "due_before") if isinstance(due_before, tuple): return due_before due_after = parse_iso_date(request.args.get("due_after"), "due_after") if isinstance(due_after, tuple): return due_after tasks, total = await list_notes( uid, q=q, tags=tag or None, is_task=True, status=status, priority=priority, due_before=due_before, due_after=due_after, sort=sort, order=order, limit=limit, offset=offset, ) return jsonify({"tasks": [t.to_dict() for t in tasks], "total": total}) @tasks_bp.route("", methods=["POST"]) @login_required async def create_task_route(): uid = get_current_user_id() data = await request.get_json() body = data.get("body", "") or data.get("description", "") tags = data.get("tags", []) due_date = parse_iso_date(data.get("due_date"), "due_date") if isinstance(due_date, tuple): return due_date status = TaskStatus(data["status"]).value if "status" in data else TaskStatus.todo.value priority = ( TaskPriority(data["priority"]).value if "priority" in data else TaskPriority.none.value ) project_id = data.get("project_id") if project_id is None and data.get("project"): from fabledassistant.services.projects import get_project_by_title as _gpbt proj = await _gpbt(uid, data["project"]) if proj: project_id = proj.id task = await create_note( uid, title=data.get("title", ""), body=body, status=status, priority=priority, due_date=due_date, tags=tags, project_id=project_id, milestone_id=data.get("milestone_id"), parent_id=data.get("parent_id"), ) return jsonify(task.to_dict()), 201 @tasks_bp.route("/", methods=["GET"]) @login_required async def get_task_route(task_id: int): uid = get_current_user_id() task = await get_note(uid, task_id) if task is None: return not_found("Task") data = task.to_dict() if task.parent_id: parent = await get_note(uid, task.parent_id) data["parent_title"] = parent.title if parent else None return jsonify(data) @tasks_bp.route("/", methods=["PUT"]) @login_required async def update_task_route(task_id: int): uid = get_current_user_id() data = await request.get_json() fields = {} for key in ("title", "status", "priority"): if key in data: fields[key] = data[key] # Accept both "body" and "description" (prefer body) if "body" in data: fields["body"] = data["body"] elif "description" in data: fields["body"] = data["description"] if "due_date" in data: if data["due_date"]: result = parse_iso_date(data["due_date"], "due_date") if isinstance(result, tuple): return result fields["due_date"] = result else: fields["due_date"] = None if "tags" in data: fields["tags"] = data["tags"] for key in ("project_id", "milestone_id", "parent_id"): if key in data: fields[key] = data[key] task = await update_note(uid, task_id, **fields) if task is None: return not_found("Task") return jsonify(task.to_dict()) @tasks_bp.route("//status", methods=["PATCH"]) @login_required async def patch_task_status(task_id: int): uid = get_current_user_id() data = await request.get_json() status_val = data.get("status") if not status_val: return jsonify({"error": "status is required"}), 400 try: TaskStatus(status_val) except ValueError: return jsonify({"error": f"Invalid status: {status_val}"}), 400 task = await update_note(uid, task_id, status=status_val) if task is None: return not_found("Task") return jsonify(task.to_dict()) @tasks_bp.route("/", methods=["DELETE"]) @login_required async def delete_task_route(task_id: int): uid = get_current_user_id() deleted = await delete_note(uid, task_id) if not deleted: return not_found("Task") return "", 204