"""Approval Engine API views. Endpoints --------- GET /approvals/ list approval requests POST /approvals/ create approval request GET /approvals/my/ my pending approvals GET /approvals/sent/ approvals I created GET /approvals/overdue/ overdue approvals GET /approvals/{public_id}/ retrieve approval request detail PATCH /approvals/{public_id}/ update approval request DELETE /approvals/{public_id}/ delete approval request POST /approvals/{public_id}/cancel/ cancel approval request POST /approvals/{public_id}/decide/ decide on an approval step GET /approvals/matrices/ list approval matrices POST /approvals/matrices/ create approval matrix GET /approvals/matrices/{public_id}/ retrieve matrix detail PATCH /approvals/matrices/{public_id}/ update matrix DELETE /approvals/matrices/{public_id}/ delete matrix """ 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.approval_engine import selectors as sel from simorgh.apps.approval_engine import services from simorgh.apps.approval_engine.models import ( ApprovalRequest, ApprovalStep, ApprovalMatrix, ) 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_step(step: ApprovalStep) -> dict: decided_by = step.decided_by approver = step.approver return { "id": step.pk, "public_id": str(step.public_id), "request_id": step.request_id, "step_order": step.step_order, "approver_id": step.approver_id, "approver_name": ( approver.get_full_name() or approver.username ) if approver else None, "approver_role_id": step.approver_role_id, "approver_org_unit_id": step.approver_org_unit_id, "status": step.status, "decision": step.decision, "decided_by_id": step.decided_by_id, "decided_by_name": ( decided_by.get_full_name() or decided_by.username ) if decided_by else None, "decided_at": step.decided_at.isoformat() if step.decided_at else None, "note": step.note, "due_date": step.due_date.isoformat() if step.due_date else None, "is_parallel": step.is_parallel, "created_at": step.created_at.isoformat(), "updated_at": step.updated_at.isoformat(), } def _serialize_request(approval_request: ApprovalRequest, *, detail: bool = False) -> dict: requested_by = approval_request.requested_by data: dict = { "id": approval_request.pk, "public_id": str(approval_request.public_id), "title": approval_request.title, "approval_type": approval_request.approval_type, "status": approval_request.status, "priority": approval_request.priority, "requested_by_id": approval_request.requested_by_id, "requested_by_name": ( requested_by.get_full_name() or requested_by.username ) if requested_by else None, "due_date": approval_request.due_date.isoformat() if approval_request.due_date else None, "completed_at": approval_request.completed_at.isoformat() if approval_request.completed_at else None, "is_confidential": approval_request.is_confidential, "content_type_id": approval_request.content_type_id, "object_id": approval_request.object_id, "matrix_id": approval_request.matrix_id, "steps_count": approval_request.steps.count(), "created_at": approval_request.created_at.isoformat(), "updated_at": approval_request.updated_at.isoformat(), } if detail: data["description"] = approval_request.description data["steps"] = [_serialize_step(s) for s in approval_request.steps.all()] return data def _serialize_matrix(matrix: ApprovalMatrix, *, detail: bool = False) -> dict: data: dict = { "id": matrix.pk, "public_id": str(matrix.public_id), "tenant_id": matrix.tenant_id, "name": matrix.name, "content_type_id": matrix.content_type_id, "is_active": matrix.is_active, "created_at": matrix.created_at.isoformat(), "updated_at": matrix.updated_at.isoformat(), } if detail: data["description"] = matrix.description data["conditions"] = matrix.conditions data["approver_configs"] = [ { "step_order": cfg.step_order, "approver_id": cfg.approver_id, "approver_role_id": cfg.approver_role_id, "approver_org_unit_id": cfg.approver_org_unit_id, "is_parallel": cfg.is_parallel, "is_required": cfg.is_required, "quorum": cfg.quorum, "majority_voting": cfg.majority_voting, } for cfg in matrix.approver_configs.all() ] return data # ── Approval Matrices ─────────────────────────────────────────────────────── @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def matrix_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": matrices = sel.matrices_for_tenant(tenant_id) return Response([_serialize_matrix(m) for m in matrices]) data = request.data try: matrix = services.create_approval_matrix( tenant_id=tenant_id, name=data["name"], content_type_id=data["content_type_id"], description=data.get("description", ""), conditions=data.get("conditions", {}), is_active=data.get("is_active", True), approver_configs=data.get("approver_configs"), ) 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_matrix(matrix, detail=True), status=status.HTTP_201_CREATED) @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def matrix_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) matrix = get_object_or_404(ApprovalMatrix, public_id=public_id, tenant_id=tenant_id) if request.method == "GET": return Response(_serialize_matrix(matrix, detail=True)) if request.method == "DELETE": services.delete_approval_matrix(matrix) return Response(status=status.HTTP_204_NO_CONTENT) services.update_approval_matrix( matrix, name=request.data.get("name"), description=request.data.get("description"), conditions=request.data.get("conditions"), is_active=request.data.get("is_active"), ) return Response(_serialize_matrix(matrix, detail=True)) # ── Approval Requests ─────────────────────────────────────────────────────── @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def request_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.requests_for_tenant(tenant_id).prefetch_related("steps") requested_by = request.query_params.get("requested_by") s = request.query_params.get("status") approval_type = request.query_params.get("type") priority = request.query_params.get("priority") content_type_id = request.query_params.get("content_type_id") object_id = request.query_params.get("object_id") overdue = request.query_params.get("overdue") if requested_by: qs = qs.filter(requested_by_id=requested_by) if s: qs = qs.filter(status=s) if approval_type: qs = qs.filter(approval_type=approval_type) if priority: qs = qs.filter(priority=priority) if content_type_id and object_id: qs = qs.filter(content_type_id=content_type_id, object_id=object_id) if overdue == "true": from django.utils import timezone as _tz qs = qs.filter(due_date__lt=_tz.now()).exclude( status__in=("approved", "rejected", "cancelled", "expired") ) return Response([_serialize_request(r) for r 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 approval_request = services.create_approval_request( tenant_id=tenant_id, organization_node_id=_get_org_node_id(request) or 1, title=data["title"], requested_by_id=request.user.pk, description=data.get("description", ""), approval_type=data.get("type", "single"), priority=data.get("priority", "medium"), 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"), matrix_id=data.get("matrix_id"), steps_config=data.get("steps"), ) 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_request(approval_request, detail=True), status=status.HTTP_201_CREATED) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def request_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.requests_pending_my(tenant_id, request.user.pk).prefetch_related("steps") return Response([_serialize_request(r) for r in qs]) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def request_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.requests_my(tenant_id, request.user.pk).prefetch_related("steps") return Response([_serialize_request(r) for r in qs]) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def request_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.requests_overdue(tenant_id, request.user.pk).prefetch_related("steps") return Response([_serialize_request(r) for r in qs]) @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def request_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) approval_request = get_object_or_404( ApprovalRequest.objects.prefetch_related("steps"), public_id=public_id, tenant_id=tenant_id, ) if request.method == "GET": return Response(_serialize_request(approval_request, detail=True)) if request.method == "DELETE": services.delete_approval_request(approval_request) 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 services.update_approval_request(approval_request, **kwargs) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) approval_request.refresh_from_db() return Response(_serialize_request(approval_request, detail=True)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def request_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) approval_request = get_object_or_404(ApprovalRequest, public_id=public_id, tenant_id=tenant_id) try: services.cancel_approval_request(approval_request, actor_id=request.user.pk) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_request(approval_request)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def request_decide(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) approval_request = get_object_or_404( ApprovalRequest.objects.prefetch_related("steps"), public_id=public_id, tenant_id=tenant_id, ) step_id = request.data.get("step_id") if not step_id: return Response({"detail": "step_id is required."}, status=status.HTTP_400_BAD_REQUEST) step = get_object_or_404(ApprovalStep, public_id=step_id, request=approval_request) decision = request.data.get("decision") if not decision: return Response({"detail": "decision is required."}, status=status.HTTP_400_BAD_REQUEST) try: services.decide_step( step, decision=decision, decided_by_id=request.user.pk, note=request.data.get("note", ""), delegate_to_id=request.data.get("delegate_to_id"), ) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) approval_request.refresh_from_db() return Response(_serialize_request(approval_request, detail=True))