bvandeusen
22634aa0c9
- Add parse_pagination() to routes/utils.py; replace 6 duplicate limit/offset extractions in notes, tasks, chat, projects, milestones routes - Extract _enrich_shares() in sharing.py; eliminates identical 12-line loop in list_project_shares and list_note_shares - Extract _deduplicate_by_permission() in sharing.py; eliminates identical deduplication blocks in list_shared_with_me for projects and notes Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
120 lines
4.2 KiB
Python
120 lines
4.2 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.milestones import (
|
|
create_milestone,
|
|
delete_milestone,
|
|
get_milestone,
|
|
get_milestone_progress,
|
|
list_milestones,
|
|
update_milestone,
|
|
)
|
|
from fabledassistant.services.notes import list_notes
|
|
from fabledassistant.services.projects import get_project
|
|
|
|
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()
|
|
project = await get_project(uid, project_id)
|
|
if project is None:
|
|
return not_found("Project")
|
|
status = request.args.get("status")
|
|
milestones = await list_milestones(uid, project_id, status=status)
|
|
result = [await _milestone_dict(m) for m in milestones]
|
|
return jsonify({"milestones": result})
|
|
|
|
|
|
@milestones_bp.route("/<int:project_id>/milestones", methods=["POST"])
|
|
@login_required
|
|
async def create_milestone_route(project_id: int):
|
|
uid = get_current_user_id()
|
|
project = await get_project(uid, project_id)
|
|
if project is None:
|
|
return not_found("Project")
|
|
data = await request.get_json()
|
|
if not data.get("title"):
|
|
return jsonify({"error": "title is required"}), 400
|
|
milestone = await create_milestone(
|
|
uid,
|
|
project_id,
|
|
title=data["title"],
|
|
description=data.get("description"),
|
|
order_index=data.get("order_index", 0),
|
|
)
|
|
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()
|
|
milestone = await get_milestone(uid, milestone_id)
|
|
if milestone is None or milestone.project_id != project_id:
|
|
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()
|
|
milestone = await get_milestone(uid, milestone_id)
|
|
if milestone is None or milestone.project_id != project_id:
|
|
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}
|
|
updated = await update_milestone(uid, 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()
|
|
milestone = await get_milestone(uid, milestone_id)
|
|
if milestone is None or milestone.project_id != project_id:
|
|
return not_found("Milestone")
|
|
deleted = await delete_milestone(uid, milestone_id)
|
|
if not deleted:
|
|
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()
|
|
milestone = await get_milestone(uid, milestone_id)
|
|
if milestone is None or milestone.project_id != project_id:
|
|
return not_found("Milestone")
|
|
status_filter = request.args.get("status")
|
|
limit, offset = parse_pagination(default_limit=100)
|
|
notes, total = await list_notes(
|
|
uid,
|
|
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})
|