"""Tasks API views. Endpoints --------- GET /tasks/ list tasks (filterable) POST /tasks/ create task GET /tasks/my/ tasks assigned to current user GET /tasks/{public_id}/ retrieve task detail PATCH /tasks/{public_id}/ update task DELETE /tasks/{public_id}/ delete task POST /tasks/{public_id}/status/ change task status POST /tasks/{public_id}/assign/ assign task to user GET /tasks/{public_id}/subtasks/ list subtasks GET /tasks/{public_id}/comments/ list comments POST /tasks/{public_id}/comments/ add comment PATCH /tasks/{public_id}/comments/{id}/ edit comment DELETE /tasks/{public_id}/comments/{id}/ delete comment POST /tasks/{public_id}/watch/ watch task DELETE /tasks/{public_id}/watch/ unwatch task GET /tasks/tags/ list tags POST /tasks/tags/ create tag PATCH /tasks/tags/{tag_id}/ update tag DELETE /tasks/tags/{tag_id}/ delete tag """ from __future__ import annotations from django.shortcuts import get_object_or_404 from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from simorgh.apps.tasks import selectors as sel from simorgh.apps.tasks import services from simorgh.apps.tasks.models import Task, TaskComment, TaskTag # --------------------------------------------------------------------------- # Tenant helpers (mirrors projects.views pattern) # --------------------------------------------------------------------------- def _get_tenant_id(request: Request) -> int | None: tenant = getattr(request, "tenant", None) return tenant.pk if tenant else None def _get_org_node_id(request: Request) -> int | None: tenant = getattr(request, "tenant", None) if tenant is None: return None try: from simorgh.apps.organizations.models import OrganizationNode node = OrganizationNode.objects.filter(tenant=tenant).order_by("lft").first() return node.pk if node else None except Exception: return None # --------------------------------------------------------------------------- # Serialization helpers # --------------------------------------------------------------------------- def _serialize_tag(tag: TaskTag) -> dict: return { "id": tag.pk, "public_id": str(tag.public_id), "name": tag.name, "color": tag.color, } def _serialize_comment(c: TaskComment) -> dict: user = c.author return { "id": c.pk, "author_id": user.pk if user else None, "author_name": getattr(user, "get_full_name", lambda: "")() or getattr(user, "username", "") if user else "", "body": c.body, "is_edited": c.is_edited, "created_at": c.created_at.isoformat(), "updated_at": c.updated_at.isoformat(), } def _serialize_task(task: Task, *, detail: bool = False) -> dict: assignee = task.assignee project = task.project data: dict = { "id": task.pk, "public_id": str(task.public_id), "title": task.title, "status": task.status, "priority": task.priority, "project_id": task.project_id, "project_name": project.name if project else None, "phase_id": task.phase_id, "assignee_id": task.assignee_id, "assignee_name": ( assignee.get_full_name() or assignee.username ) if assignee else None, "reporter_id": task.reporter_id, "due_date": task.due_date.isoformat() if task.due_date else None, "estimated_hours": str(task.estimated_hours) if task.estimated_hours else None, "parent_id": task.parent_id, "order": task.order, "tags": [_serialize_tag(t) for t in task.tags.all()], "created_at": task.created_at.isoformat(), "updated_at": task.updated_at.isoformat(), } if detail: data["description"] = task.description data["subtask_count"] = task.subtasks.count() data["comment_count"] = task.comments.count() data["attachment_count"] = task.attachments.count() data["watcher_count"] = task.watchers.count() data["status_history"] = [ { "from_status": sc.from_status, "to_status": sc.to_status, "changed_by": sc.changed_by_id, "changed_at": sc.changed_at.isoformat(), } for sc in task.status_changes.all() ] return data # --------------------------------------------------------------------------- # Tags # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def tag_list_create(request: Request) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) if request.method == "GET": tags = sel.tags_for_tenant(tenant_id) return Response([_serialize_tag(t) for t in tags]) # POST data = request.data try: tag = services.create_task_tag( tenant_id=tenant_id, name=data["name"], color=data.get("color", ""), ) except KeyError as exc: return Response({"detail": f"Missing field: {exc}"}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_tag(tag), status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def tag_detail(request: Request, tag_id: int) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) tag = get_object_or_404(TaskTag, pk=tag_id, tenant_id=tenant_id) if request.method == "DELETE": services.delete_task_tag(tag) return Response(status=status.HTTP_204_NO_CONTENT) # PATCH services.update_task_tag( tag, name=request.data.get("name"), color=request.data.get("color"), ) return Response(_serialize_tag(tag)) # --------------------------------------------------------------------------- # Tasks # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def task_list_create(request: Request) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) if request.method == "GET": qs = sel.tasks_for_tenant(tenant_id).prefetch_related("tags") # Filters project_id = request.query_params.get("project") phase_id = request.query_params.get("phase") s = request.query_params.get("status") priority = request.query_params.get("priority") assignee = request.query_params.get("assignee") overdue = request.query_params.get("overdue") unassigned = request.query_params.get("unassigned") due_from = request.query_params.get("due_from") due_to = request.query_params.get("due_to") if project_id: qs = qs.filter(project_id=project_id) if phase_id: qs = qs.filter(phase_id=phase_id) if s: qs = qs.filter(status=s) if priority: qs = qs.filter(priority=priority) if assignee: qs = qs.filter(assignee_id=assignee) if unassigned == "true": qs = qs.filter(assignee__isnull=True) if due_from: qs = qs.filter(due_date__gte=due_from) if due_to: qs = qs.filter(due_date__lte=due_to) if overdue == "true": from django.utils import timezone qs = qs.filter(due_date__lt=timezone.now().date()).exclude(status__in=("done", "cancelled")) return Response([_serialize_task(t) for t in qs]) # POST data = request.data try: from datetime import date as _date due_date_raw = data.get("due_date") due_date = _date.fromisoformat(due_date_raw) if isinstance(due_date_raw, str) else due_date_raw task = services.create_task( tenant_id=tenant_id, organization_node_id=_get_org_node_id(request) or 1, title=data["title"], reporter_id=request.user.pk, description=data.get("description", ""), status=data.get("status", "todo"), priority=data.get("priority", "none"), project_id=data.get("project_id"), phase_id=data.get("phase_id"), assignee_id=data.get("assignee_id"), due_date=due_date, parent_id=data.get("parent_id"), tag_ids=data.get("tag_ids"), order=float(data.get("order", 0.0)), ) except KeyError as exc: return Response({"detail": f"Missing field: {exc}"}, status=status.HTTP_400_BAD_REQUEST) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_task(task, detail=True), status=status.HTTP_201_CREATED) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def task_my(request: Request) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) qs = sel.tasks_my(tenant_id, request.user.pk).prefetch_related("tags") return Response([_serialize_task(t) for t in qs]) @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def task_detail(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) task = get_object_or_404( Task.objects.prefetch_related("tags", "status_changes"), public_id=public_id, tenant_id=tenant_id, ) if request.method == "GET": return Response(_serialize_task(task, detail=True)) if request.method == "DELETE": services.delete_task(task) return Response(status=status.HTTP_204_NO_CONTENT) # PATCH data = request.data try: from datetime import date as _date due_date_raw = data.get("due_date", ...) if due_date_raw is not ... and isinstance(due_date_raw, str): due_date_raw = _date.fromisoformat(due_date_raw) kwargs: dict = {} for field in ("title", "description", "priority", "order"): if field in data: kwargs[field] = data[field] for nullable_field in ("project_id", "phase_id", "assignee_id"): if nullable_field in data: kwargs[nullable_field] = data[nullable_field] if due_date_raw is not ...: kwargs["due_date"] = due_date_raw if "tag_ids" in data: kwargs["tag_ids"] = data["tag_ids"] services.update_task(task, actor_id=request.user.pk, **kwargs) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) task.refresh_from_db() return Response(_serialize_task(task, detail=True)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def task_change_status(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) task = get_object_or_404(Task, public_id=public_id, tenant_id=tenant_id) new_status = request.data.get("status") if not new_status: return Response({"detail": "status is required."}, status=status.HTTP_400_BAD_REQUEST) try: services.change_task_status(task, new_status=new_status, actor_id=request.user.pk) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_task(task)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def task_assign(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) task = get_object_or_404(Task, public_id=public_id, tenant_id=tenant_id) user_id = request.data.get("user_id") services.assign_task(task, user_id=user_id, actor_id=request.user.pk) return Response(_serialize_task(task)) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def task_subtasks(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) task = get_object_or_404(Task, public_id=public_id, tenant_id=tenant_id) qs = sel.subtasks_for_task(task.pk).prefetch_related("tags") return Response([_serialize_task(t) for t in qs]) # --------------------------------------------------------------------------- # Comments # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def task_comments(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) task = get_object_or_404(Task, public_id=public_id, tenant_id=tenant_id) if request.method == "GET": qs = sel.comments_for_task(task.pk) return Response([_serialize_comment(c) for c in qs]) # POST body = request.data.get("body", "") try: comment = services.add_comment(task, author_id=request.user.pk, body=body) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_comment(comment), status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def task_comment_detail(request: Request, public_id: str, comment_id: int) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) task = get_object_or_404(Task, public_id=public_id, tenant_id=tenant_id) from simorgh.apps.tasks.models import TaskComment comment = get_object_or_404(TaskComment, pk=comment_id, task=task) if request.method == "DELETE": services.delete_comment(comment) return Response(status=status.HTTP_204_NO_CONTENT) # PATCH body = request.data.get("body", "") try: services.update_comment(comment, body=body) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_comment(comment)) # --------------------------------------------------------------------------- # Watch / Unwatch # --------------------------------------------------------------------------- @api_view(["POST", "DELETE"]) @permission_classes([IsAuthenticated]) def task_watch(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant header required."}, status=status.HTTP_400_BAD_REQUEST) task = get_object_or_404(Task, public_id=public_id, tenant_id=tenant_id) if request.method == "POST": services.watch_task(task, user_id=request.user.pk) return Response({"watching": True}) services.unwatch_task(task, user_id=request.user.pk) return Response({"watching": False})