Coverage for apps/client_needs/api.py: 79%
111 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""Document reference + client-need router (mounted under /api/loans), plus the
2document classification catalog router (mounted at "")."""
4from uuid import UUID
6from ninja import Router
8from apps.accounts.auth import ensure_staff_scope, requires
9from apps.client_needs.models import ClientNeed, Document, DocumentClassification
10from apps.client_needs.schemas import (
11 ClientNeedCreate,
12 ClientNeedFulfill,
13 ClientNeedOut,
14 DocumentClassificationCreate,
15 DocumentClassificationOut,
16 DocumentClassificationPatch,
17 DocumentCreate,
18 DocumentOut,
19 DocumentPatch,
20 GeneratedNeedsOut,
21)
22from apps.client_needs.services import generate_standard_needs
23from apps.core_los.models import Loan
24from domain.result import AppError, DomainError
26router = Router(tags=["client-needs"])
27classifications_router = Router(tags=["client-needs"])
30def _get_classification(code: str) -> DocumentClassification:
31 classification = DocumentClassification.objects.filter(code=code).first()
32 if classification is None: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true
33 raise DomainError([AppError("NOT_FOUND", "Document classification not found")])
34 return classification
37@classifications_router.get(
38 "/document-classifications",
39 response=list[DocumentClassificationOut],
40 auth=requires("DOCUMENT_CLASSIFICATION_VIEW"),
41)
42def list_document_classifications(request):
43 return list(DocumentClassification.objects.filter(is_active=True))
46@classifications_router.post(
47 "/document-classifications",
48 response={201: DocumentClassificationOut},
49 auth=requires("DOCUMENT_CLASSIFICATION_EDIT"),
50)
51def create_document_classification(request, payload: DocumentClassificationCreate):
52 if DocumentClassification.objects.filter(code=payload.code).exists(): 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 raise DomainError(
54 [AppError("VALIDATION", f"Classification {payload.code} already exists", "code")]
55 )
56 classification = DocumentClassification.objects.create(
57 code=payload.code, name=payload.name, category=payload.category, system=False
58 )
59 return 201, classification
62@classifications_router.patch(
63 "/document-classifications/{code}",
64 response=DocumentClassificationOut,
65 auth=requires("DOCUMENT_CLASSIFICATION_EDIT"),
66)
67def update_document_classification(request, code: str, payload: DocumentClassificationPatch):
68 classification = _get_classification(code)
69 for field, value in payload.model_dump(exclude_unset=True).items():
70 setattr(classification, field, value)
71 classification.save()
72 return classification
75@classifications_router.delete(
76 "/document-classifications/{code}",
77 response=DocumentClassificationOut,
78 auth=requires("DOCUMENT_CLASSIFICATION_EDIT"),
79)
80def delete_document_classification(request, code: str):
81 classification = _get_classification(code)
82 if classification.system:
83 raise DomainError(
84 [AppError("CLASSIFICATION_PROTECTED", "Built-in classifications can't be deleted")]
85 )
86 classification.is_active = False
87 classification.save(update_fields=["is_active"])
88 return classification
91def _scoped_loan(request, guid: UUID) -> Loan:
92 loan = Loan.objects.filter(guid=guid).first()
93 if loan is None: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 raise DomainError([AppError("NOT_FOUND", "Loan not found")])
95 ensure_staff_scope(request.principal, loan)
96 return loan
99@router.get("/{guid}/documents", response=list[DocumentOut], auth=requires("LOAN_VIEW"))
100def list_documents(request, guid: UUID):
101 loan = _scoped_loan(request, guid)
102 return list(loan.documents.all().order_by("-created_at"))
105@router.post("/{guid}/documents", response={201: DocumentOut}, auth=requires("LOAN_EDIT"))
106def create_document(request, guid: UUID, payload: DocumentCreate):
107 loan = _scoped_loan(request, guid)
108 if not DocumentClassification.objects.filter(
109 code=payload.classification, is_active=True
110 ).exists():
111 raise DomainError(
112 [
113 AppError(
114 "VALIDATION",
115 f"Unknown classification: {payload.classification}",
116 "classification",
117 )
118 ]
119 )
120 document = Document.objects.create(loan=loan, **payload.model_dump())
121 return 201, document
124@router.patch("/{guid}/documents/{document_id}", response=DocumentOut, auth=requires("LOAN_EDIT"))
125def update_document(request, guid: UUID, document_id: int, payload: DocumentPatch):
126 loan = _scoped_loan(request, guid)
127 doc = loan.documents.filter(pk=document_id).first()
128 if doc is None: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 raise DomainError([AppError("NOT_FOUND", "Document not found on this loan")])
130 data = payload.model_dump(exclude_unset=True)
131 if "classification" in data and data["classification"]: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true
132 if not DocumentClassification.objects.filter(code=data["classification"], is_active=True).exists():
133 raise DomainError([AppError("VALIDATION", f"Unknown classification: {data['classification']}", "classification")])
134 doc.classification = data["classification"]
135 if "external_doc_id" in data: 135 ↛ 137line 135 didn't jump to line 137 because the condition on line 135 was always true
136 doc.external_doc_id = data["external_doc_id"]
137 doc.save()
138 return doc
141@router.delete("/{guid}/documents/{document_id}", response={204: None}, auth=requires("LOAN_EDIT"))
142def delete_document(request, guid: UUID, document_id: int):
143 loan = _scoped_loan(request, guid)
144 doc = loan.documents.filter(pk=document_id).first()
145 if doc is None: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 raise DomainError([AppError("NOT_FOUND", "Document not found on this loan")])
147 doc.delete()
148 return 204, None
151@router.get("/{guid}/needs", response=list[ClientNeedOut], auth=requires("LOAN_VIEW"))
152def list_needs(request, guid: UUID):
153 loan = _scoped_loan(request, guid)
154 return list(loan.needs.all().order_by("-created_at"))
157@router.post("/{guid}/needs", response={201: ClientNeedOut}, auth=requires("LOAN_EDIT"))
158def create_need(request, guid: UUID, payload: ClientNeedCreate):
159 loan = _scoped_loan(request, guid)
160 data = payload.model_dump()
161 borrower_ids = data.pop("borrower_ids", [])
162 if not borrower_ids:
163 primary = loan.borrowers.filter(is_primary=True).first() or loan.borrowers.first()
164 if primary: 164 ↛ 166line 164 didn't jump to line 166 because the condition on line 164 was always true
165 borrower_ids = [primary.id]
166 need = ClientNeed.objects.create(loan=loan, **data)
167 need.borrowers.set(borrower_ids)
168 return 201, need
171@router.patch("/{guid}/needs/{need_id}", response=ClientNeedOut, auth=requires("LOAN_EDIT"))
172def fulfill_need(request, guid: UUID, need_id: int, payload: ClientNeedFulfill):
173 loan = _scoped_loan(request, guid)
174 need = ClientNeed.objects.filter(pk=need_id, loan=loan).first()
175 if need is None: 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true
176 raise DomainError([AppError("NOT_FOUND", "Client need not found on this loan")])
177 document = Document.objects.filter(pk=payload.document_id, loan=loan).first()
178 if document is None: 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true
179 raise DomainError([AppError("NOT_FOUND", "Document not found on this loan")])
180 need.fulfill_with(document)
181 return need
184@router.post(
185 "/{guid}/generate-needs", response={201: GeneratedNeedsOut}, auth=requires("LOAN_EDIT")
186)
187def generate_needs(request, guid: UUID):
188 loan = _scoped_loan(request, guid)
189 created, skipped = generate_standard_needs(loan)
190 return 201, {"created": created, "skipped": skipped}