"""Assignment Engine API views. Endpoints --------- GET /assignments/ list assignments (filterable) POST /assignments/ create assignment GET /assignments/my/ my active assignments GET /assignments/inbox/ all my assignments GET /assignments/sent/ assignments I created GET /assignments/overdue/ overdue assignments GET /assignments/{public_id}/ retrieve assignment detail PATCH /assignments/{public_id}/ update assignment DELETE /assignments/{public_id}/ delete assignment POST /assignments/{public_id}/complete/ complete assignment POST /assignments/{public_id}/reassign/ reassign assignment POST /assignments/{public_id}/cancel/ cancel assignment GET /assignments/tags/ list tags POST /assignments/tags/ create tag PATCH /assignments/tags/{tag_id}/ update tag DELETE /assignments/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.assignments import selectors as sel from simorgh.apps.assignments import services from simorgh.apps.assignments.models import Assignment, AssignmentTag 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 def _serialize_tag(tag: AssignmentTag) -> dict: return { "id": tag.pk, "public_id": str(tag.public_id), "name": tag.name, "color": tag.color, } def _serialize_assignment(assignment: Assignment, *, detail: bool = False) -> dict: assigned_to = assignment.assigned_to assigned_by = assignment.assigned_by completed_by = assignment.completed_by data: dict = { "id": assignment.pk, "public_id": str(assignment.public_id), "title": assignment.title, "assignment_type": assignment.assignment_type, "status": assignment.status, "priority": assignment.priority, "assigned_by_id": assignment.assigned_by_id, "assigned_by_name": ( assigned_by.get_full_name() or assigned_by.username ) if assigned_by else None, "assigned_to_id": assignment.assigned_to_id, "assigned_to_name": ( assigned_to.get_full_name() or assigned_to.username ) if assigned_to else None, "assigned_role_id": assignment.assigned_role_id, "assigned_org_unit_id": assignment.assigned_org_unit_id, "due_date": assignment.due_date.isoformat() if assignment.due_date else None, "completed_at": assignment.completed_at.isoformat() if assignment.completed_at else None, "completed_by_id": assignment.completed_by_id, "completed_by_name": ( completed_by.get_full_name() or completed_by.username ) if completed_by else None, "is_confidential": assignment.is_confidential, "content_type_id": assignment.content_type_id, "object_id": assignment.object_id, "tags": [_serialize_tag(t) for t in assignment.tags.all()], "created_at": assignment.created_at.isoformat(), "updated_at": assignment.updated_at.isoformat(), } if detail: data["description"] = assignment.description 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]) data = request.data try: tag = services.create_assignment_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(AssignmentTag, pk=tag_id, tenant_id=tenant_id) if request.method == "DELETE": services.delete_assignment_tag(tag) return Response(status=status.HTTP_204_NO_CONTENT) services.update_assignment_tag( tag, name=request.data.get("name"), color=request.data.get("color"), ) return Response(_serialize_tag(tag)) # ── Assignments ─────────────────────────────────────────────────────────── @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def assignment_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.assignments_for_tenant(tenant_id).prefetch_related("tags") assigned_to = request.query_params.get("assigned_to") assigned_by = request.query_params.get("assigned_by") assignment_type = request.query_params.get("type") s = request.query_params.get("status") priority = request.query_params.get("priority") overdue = request.query_params.get("overdue") content_type_id = request.query_params.get("content_type_id") object_id = request.query_params.get("object_id") if assigned_to: qs = qs.filter(assigned_to_id=assigned_to) if assigned_by: qs = qs.filter(assigned_by_id=assigned_by) if assignment_type: qs = qs.filter(assignment_type=assignment_type) if s: qs = qs.filter(status=s) if priority: qs = qs.filter(priority=priority) if overdue == "true": from django.utils import timezone qs = qs.filter(due_date__lt=timezone.now()).exclude(status__in=("completed", "cancelled")) if content_type_id and object_id: qs = qs.filter(content_type_id=content_type_id, object_id=object_id) return Response([_serialize_assignment(a) for a in qs]) # POST data = request.data try: from datetime import datetime as _datetime due_date_raw = data.get("due_date") due_date = _datetime.fromisoformat(due_date_raw) if isinstance(due_date_raw, str) else due_date_raw assignment = services.create_assignment( tenant_id=tenant_id, organization_node_id=_get_org_node_id(request) or 1, title=data["title"], assigned_by_id=request.user.pk, description=data.get("description", ""), assignment_type=data.get("type", "for_action"), priority=data.get("priority", "medium"), assigned_to_id=data.get("assigned_to_id"), assigned_role_id=data.get("assigned_role_id"), assigned_org_unit_id=data.get("assigned_org_unit_id"), due_date=due_date, is_confidential=data.get("is_confidential", False), content_type_id=data.get("content_type_id"), object_id=data.get("object_id"), tag_ids=data.get("tag_ids"), ) 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_assignment(assignment, detail=True), status=status.HTTP_201_CREATED) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def assignment_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.assignments_my(tenant_id, request.user.pk).prefetch_related("tags") return Response([_serialize_assignment(a) for a in qs]) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def assignment_inbox(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.assignments_inbox(tenant_id, request.user.pk).prefetch_related("tags") return Response([_serialize_assignment(a) for a in qs]) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def assignment_sent(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.assignments_sent(tenant_id, request.user.pk).prefetch_related("tags") return Response([_serialize_assignment(a) for a in qs]) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def assignment_overdue(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.assignments_overdue(tenant_id, request.user.pk).prefetch_related("tags") return Response([_serialize_assignment(a) for a in qs]) @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def assignment_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) assignment = get_object_or_404( Assignment.objects.prefetch_related("tags"), public_id=public_id, tenant_id=tenant_id, ) if request.method == "GET": return Response(_serialize_assignment(assignment, detail=True)) if request.method == "DELETE": services.delete_assignment(assignment) return Response(status=status.HTTP_204_NO_CONTENT) # PATCH data = request.data try: from datetime import datetime as _datetime due_date_raw = data.get("due_date", ...) if due_date_raw is not ... and isinstance(due_date_raw, str): due_date_raw = _datetime.fromisoformat(due_date_raw) kwargs: dict = {} for field in ("title", "description", "priority", "is_confidential"): if field in data: kwargs[field] = data[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_assignment(assignment, **kwargs) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) assignment.refresh_from_db() return Response(_serialize_assignment(assignment, detail=True)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def assignment_complete(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) assignment = get_object_or_404(Assignment, public_id=public_id, tenant_id=tenant_id) try: services.complete_assignment(assignment, completed_by_id=request.user.pk) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_assignment(assignment)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def assignment_reassign(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) assignment = get_object_or_404(Assignment, public_id=public_id, tenant_id=tenant_id) to_user_id = request.data.get("to_user_id") if not to_user_id: return Response({"detail": "to_user_id is required."}, status=status.HTTP_400_BAD_REQUEST) try: services.reassign_assignment(assignment, to_user_id=to_user_id, actor_id=request.user.pk) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_assignment(assignment)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def assignment_cancel(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) assignment = get_object_or_404(Assignment, public_id=public_id, tenant_id=tenant_id) try: services.cancel_assignment(assignment, actor_id=request.user.pk) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_assignment(assignment))