from __future__ import annotations from datetime import datetime from fastapi import APIRouter, Depends, Path from fastapi.responses import JSONResponse from app.api.dto.tasks import ErrorDTO, TaskBoardDTO, TaskBoardItemDTO, TaskDetailsDTO from app.api.tasks import get_task_or_error, get_task_repository from app.storage import JsonFileTaskRepository, StoredTask router = APIRouter(prefix="/ui_data", tags=["ui_data"]) def get_task_status(task: StoredTask) -> str: if task.done_at is not None: return "done" if task.started_at is not None: return "in_progress" return "backlog" def format_duration(task: StoredTask) -> str: if task.started_at is None or task.done_at is None: return "—" delta = task.done_at - task.started_at total_seconds = int(delta.total_seconds()) hours, remainder = divmod(total_seconds, 3600) minutes, seconds = divmod(remainder, 60) return f"{hours}h {minutes}m {seconds}s" def to_board_item_dto(task: StoredTask) -> TaskBoardItemDTO: status = get_task_status(task) display_date_label: str display_date_value: datetime | None if status == "backlog": display_date_label = "Created" display_date_value = task.created_at elif status == "in_progress": display_date_label = "Started" display_date_value = task.started_at else: display_date_label = "Done" display_date_value = task.done_at return TaskBoardItemDTO( id=task.id, title=task.title, status=status, display_date_label=display_date_label, display_date_value=display_date_value, ) def to_task_details_dto(task: StoredTask) -> TaskDetailsDTO: return TaskDetailsDTO( id=task.id, title=task.title, status=get_task_status(task), created_at=task.created_at, started_at=task.started_at, done_at=task.done_at, cycle_time=format_duration(task), ) @router.get( "/stats/board", response_model=TaskBoardDTO, responses={ 400: {"model": ErrorDTO}, }, include_in_schema=False, ) def stats_board( repo: JsonFileTaskRepository = Depends(get_task_repository), ) -> TaskBoardDTO: tasks = repo.list_tasks() backlog_tasks: list[TaskBoardItemDTO] = [] in_progress_tasks: list[TaskBoardItemDTO] = [] done_tasks: list[TaskBoardItemDTO] = [] for task in tasks: item = to_board_item_dto(task) if item.status == "backlog": backlog_tasks.append(item) elif item.status == "in_progress": in_progress_tasks.append(item) else: done_tasks.append(item) return TaskBoardDTO( backlog_tasks=backlog_tasks, in_progress_tasks=in_progress_tasks, done_tasks=done_tasks, ) @router.get( "/tasks/{task_id}", response_model=TaskDetailsDTO, responses={ 400: {"model": ErrorDTO}, 404: {"model": ErrorDTO}, }, include_in_schema=False, ) def task_details( task_id: str = Path(...), repo: JsonFileTaskRepository = Depends(get_task_repository), ) -> TaskDetailsDTO | JSONResponse: task, error = get_task_or_error(repo=repo, task_id=task_id) if error is not None: return error return to_task_details_dto(task)