143 lines
5.5 KiB
Python
143 lines
5.5 KiB
Python
"""Milestone routes nested under /api/projects/<project_id>/milestones."""
|
|
import logging
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
|
|
from fabledassistant.auth import login_required, get_current_user_id
|
|
from fabledassistant.routes.utils import not_found, parse_pagination
|
|
from fabledassistant.services.access import can_write_project
|
|
from fabledassistant.services.milestones import (
|
|
create_milestone,
|
|
delete_milestone,
|
|
get_milestone_in_project,
|
|
get_milestone_progress,
|
|
list_milestones,
|
|
update_milestone,
|
|
)
|
|
from fabledassistant.services.notes import list_notes
|
|
from fabledassistant.services.projects import get_project_for_user
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
milestones_bp = Blueprint("milestones", __name__, url_prefix="/api/projects")
|
|
|
|
|
|
async def _milestone_dict(m) -> dict:
|
|
d = m.to_dict()
|
|
d.update(await get_milestone_progress(m.id))
|
|
return d
|
|
|
|
|
|
@milestones_bp.route("/<int:project_id>/milestones", methods=["GET"])
|
|
@login_required
|
|
async def list_milestones_route(project_id: int):
|
|
uid = get_current_user_id()
|
|
result = await get_project_for_user(uid, project_id)
|
|
if result is None:
|
|
return not_found("Project")
|
|
project, _ = result
|
|
# List milestones using the project owner's uid so ownership filter matches
|
|
owner_uid = project.user_id or uid
|
|
status = request.args.get("status")
|
|
milestones = await list_milestones(owner_uid, project_id, status=status)
|
|
return jsonify({"milestones": [await _milestone_dict(m) for m in milestones]})
|
|
|
|
|
|
@milestones_bp.route("/<int:project_id>/milestones", methods=["POST"])
|
|
@login_required
|
|
async def create_milestone_route(project_id: int):
|
|
uid = get_current_user_id()
|
|
if await get_project_for_user(uid, project_id) is None:
|
|
return not_found("Project")
|
|
if not await can_write_project(uid, project_id):
|
|
return jsonify({"error": "Permission denied"}), 403
|
|
data = await request.get_json()
|
|
if not data.get("title"):
|
|
return jsonify({"error": "title is required"}), 400
|
|
status = data.get("status", "active")
|
|
if status not in ("active", "done"):
|
|
return jsonify({"error": "status must be 'active' or 'done'"}), 400
|
|
milestone = await create_milestone(
|
|
uid,
|
|
project_id,
|
|
title=data["title"],
|
|
description=data.get("description"),
|
|
order_index=data.get("order_index", 0),
|
|
status=status,
|
|
)
|
|
return jsonify(await _milestone_dict(milestone)), 201
|
|
|
|
|
|
@milestones_bp.route("/<int:project_id>/milestones/<int:milestone_id>", methods=["GET"])
|
|
@login_required
|
|
async def get_milestone_route(project_id: int, milestone_id: int):
|
|
uid = get_current_user_id()
|
|
if await get_project_for_user(uid, project_id) is None:
|
|
return not_found("Project")
|
|
milestone = await get_milestone_in_project(project_id, milestone_id)
|
|
if milestone is None:
|
|
return not_found("Milestone")
|
|
return jsonify(await _milestone_dict(milestone))
|
|
|
|
|
|
@milestones_bp.route("/<int:project_id>/milestones/<int:milestone_id>", methods=["PATCH"])
|
|
@login_required
|
|
async def update_milestone_route(project_id: int, milestone_id: int):
|
|
uid = get_current_user_id()
|
|
if await get_project_for_user(uid, project_id) is None:
|
|
return not_found("Project")
|
|
if not await can_write_project(uid, project_id):
|
|
return jsonify({"error": "Permission denied"}), 403
|
|
milestone = await get_milestone_in_project(project_id, milestone_id)
|
|
if milestone is None:
|
|
return not_found("Milestone")
|
|
data = await request.get_json()
|
|
allowed = {"title", "description", "status", "order_index"}
|
|
fields = {k: v for k, v in data.items() if k in allowed}
|
|
if "status" in fields and fields["status"] not in ("active", "done"):
|
|
return jsonify({"error": "status must be 'active' or 'done'"}), 400
|
|
updated = await update_milestone(milestone.user_id, milestone_id, **fields)
|
|
if updated is None:
|
|
return not_found("Milestone")
|
|
return jsonify(await _milestone_dict(updated))
|
|
|
|
|
|
@milestones_bp.route("/<int:project_id>/milestones/<int:milestone_id>", methods=["DELETE"])
|
|
@login_required
|
|
async def delete_milestone_route(project_id: int, milestone_id: int):
|
|
uid = get_current_user_id()
|
|
if await get_project_for_user(uid, project_id) is None:
|
|
return not_found("Project")
|
|
if not await can_write_project(uid, project_id):
|
|
return jsonify({"error": "Permission denied"}), 403
|
|
milestone = await get_milestone_in_project(project_id, milestone_id)
|
|
if milestone is None:
|
|
return not_found("Milestone")
|
|
from fabledassistant.services.trash import delete as trash_delete
|
|
batch = await trash_delete(milestone.user_id, "milestone", milestone_id)
|
|
if batch is None:
|
|
return not_found("Milestone")
|
|
return "", 204
|
|
|
|
|
|
@milestones_bp.route("/<int:project_id>/milestones/<int:milestone_id>/tasks", methods=["GET"])
|
|
@login_required
|
|
async def get_milestone_tasks_route(project_id: int, milestone_id: int):
|
|
uid = get_current_user_id()
|
|
if await get_project_for_user(uid, project_id) is None:
|
|
return not_found("Project")
|
|
milestone = await get_milestone_in_project(project_id, milestone_id)
|
|
if milestone is None:
|
|
return not_found("Milestone")
|
|
status_filter = request.args.get("status")
|
|
limit, offset = parse_pagination(default_limit=100)
|
|
notes, total = await list_notes(
|
|
milestone.user_id,
|
|
is_task=True,
|
|
status=status_filter,
|
|
milestone_id=milestone_id,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
return jsonify({"notes": [n.to_dict() for n in notes], "total": total})
|