Coverage for apps/application_1003/api.py: 81%
181 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""1003 financials router: full CRUD per entity + qualification (mounted under /api/loans)."""
3from uuid import UUID
5from django.db import DataError, IntegrityError
6from django.db.models import ForeignKey
7from ninja import Router
9from apps.accounts.auth import ensure_staff_scope, requires
10from apps.application_1003.models import (
11 Asset,
12 Declaration,
13 Demographics,
14 DownPaymentSource,
15 Employer,
16 GiftGrant,
17 HousingExpense,
18 Income,
19 Liability,
20 LoanIdentifier,
21 RealEstate,
22 Residence,
23 SubjectProperty,
24)
25from apps.application_1003.schemas import (
26 AssetIn,
27 AssetOut,
28 AssetPatch,
29 DeclarationOut,
30 DeclarationPatch,
31 DemographicsOut,
32 DemographicsPatch,
33 DownPaymentSourceIn,
34 DownPaymentSourceOut,
35 DownPaymentSourcePatch,
36 EmployerIn,
37 EmployerOut,
38 EmployerPatch,
39 GiftGrantIn,
40 GiftGrantOut,
41 GiftGrantPatch,
42 HousingExpenseIn,
43 HousingExpenseOut,
44 HousingExpensePatch,
45 IncomeIn,
46 IncomeOut,
47 IncomePatch,
48 LiabilityIn,
49 LiabilityOut,
50 LiabilityPatch,
51 LoanIdentifierIn,
52 LoanIdentifierOut,
53 LoanIdentifierPatch,
54 QualificationOut,
55 RealEstateIn,
56 RealEstateOut,
57 RealEstatePatch,
58 ResidenceIn,
59 ResidenceOut,
60 ResidencePatch,
61 SubjectPropertyOut,
62 SubjectPropertyPatch,
63)
64from apps.core_los.models import Loan
65from domain.calc import monthly_pi
66from domain.qualification import HousingItem, IncomeItem, LiabilityItem, qualify
67from domain.result import AppError, DomainError, format_db_error
69router = Router(tags=["1003"])
72def _scoped_loan(request, guid: UUID) -> Loan:
73 loan = Loan.objects.filter(guid=guid).first()
74 if loan is None: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true
75 raise DomainError([AppError("NOT_FOUND", "Loan not found")])
76 ensure_staff_scope(request.principal, loan)
77 return loan
80def _item(model, loan, item_id):
81 obj = model.objects.filter(loan=loan, pk=item_id).first()
82 if obj is None: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 raise DomainError([AppError("NOT_FOUND", f"{model.__name__} not found")])
84 return obj
87def _normalize_fks(model, data: dict) -> None:
88 """ModelSchema emits FK fields by model-field name (e.g. ``borrower``) carrying the pk;
89 rename to the ``_id`` attname so Django's create/setattr accept the integer."""
90 for field in model._meta.get_fields():
91 if isinstance(field, ForeignKey) and field.name in data:
92 data[field.attname] = data.pop(field.name)
95def _register_crud(path, model, out_s, in_s, patch_s, m2m=()):
96 key = path.replace("-", "_")
98 def list_view(request, guid: UUID):
99 loan = _scoped_loan(request, guid)
100 return list(model.objects.filter(loan=loan))
102 list_view.__name__ = f"list_{key}"
103 router.get(f"/{{guid}}/{path}", response=list[out_s], auth=requires("LOAN_VIEW"))(list_view)
105 def create_view(request, guid: UUID, payload: in_s):
106 loan = _scoped_loan(request, guid)
107 data = payload.model_dump(exclude_unset=True) # type: ignore[attr-defined] # dynamic schema param
108 m2m_vals = {f: data.pop(f, None) for f in m2m}
109 _normalize_fks(model, data)
110 primary = loan.borrowers.filter(is_primary=True).first() or loan.borrowers.first()
111 if "borrower_id" in [f.attname for f in model._meta.fields] and not data.get("borrower_id"):
112 if primary: 112 ↛ 114line 112 didn't jump to line 114 because the condition on line 112 was always true
113 data["borrower_id"] = primary.id
114 if "borrowers" in m2m:
115 if not m2m_vals.get("borrowers") and primary:
116 m2m_vals["borrowers"] = [primary.id]
117 try:
118 obj = model.objects.create(loan=loan, **data)
119 except (DataError, IntegrityError) as exc:
120 raise DomainError(
121 [AppError("VALIDATION_ERROR", format_db_error(exc, model.__name__))]
122 ) from exc
123 for field, ids in m2m_vals.items():
124 if ids is not None: 124 ↛ 123line 124 didn't jump to line 123 because the condition on line 124 was always true
125 getattr(obj, field).set(ids)
126 return 201, obj
128 create_view.__name__ = f"create_{key}"
129 router.post(f"/{{guid}}/{path}", response={201: out_s}, auth=requires("LOAN_EDIT"))(create_view)
131 def update_view(request, guid: UUID, item_id: int, payload: patch_s):
132 loan = _scoped_loan(request, guid)
133 obj = _item(model, loan, item_id)
134 data = payload.model_dump(exclude_unset=True) # type: ignore[attr-defined] # dynamic schema param
135 m2m_vals = {f: data.pop(f) for f in m2m if f in data}
136 _normalize_fks(model, data)
137 try:
138 for attr, value in data.items():
139 setattr(obj, attr, value)
140 obj.save()
141 except (DataError, IntegrityError) as exc:
142 raise DomainError(
143 [AppError("VALIDATION_ERROR", format_db_error(exc, model.__name__))]
144 ) from exc
145 for field, ids in m2m_vals.items(): 145 ↛ 146line 145 didn't jump to line 146 because the loop on line 145 never started
146 getattr(obj, field).set(ids)
147 return obj
149 update_view.__name__ = f"update_{key}"
150 router.patch(f"/{{guid}}/{path}/{{item_id}}", response=out_s, auth=requires("LOAN_EDIT"))(
151 update_view
152 )
154 def delete_view(request, guid: UUID, item_id: int):
155 loan = _scoped_loan(request, guid)
156 _item(model, loan, item_id).delete()
157 return 204, None
159 delete_view.__name__ = f"delete_{key}"
160 router.delete(
161 f"/{{guid}}/{path}/{{item_id}}", response={204: None}, auth=requires("LOAN_EDIT")
162 )(delete_view)
165_register_crud("employers", Employer, EmployerOut, EmployerIn, EmployerPatch)
166_register_crud("incomes", Income, IncomeOut, IncomeIn, IncomePatch)
167_register_crud("assets", Asset, AssetOut, AssetIn, AssetPatch, m2m=["borrowers"])
168_register_crud(
169 "real-estate", RealEstate, RealEstateOut, RealEstateIn, RealEstatePatch, m2m=["borrowers"]
170)
171_register_crud(
172 "liabilities", Liability, LiabilityOut, LiabilityIn, LiabilityPatch, m2m=["borrowers"]
173)
174_register_crud(
175 "housing-expenses", HousingExpense, HousingExpenseOut, HousingExpenseIn, HousingExpensePatch
176)
179@router.get("/{guid}/qualification", response=QualificationOut, auth=requires("LOAN_VIEW"))
180def qualification(request, guid: UUID):
181 loan = _scoped_loan(request, guid)
182 incomes = [IncomeItem(i.monthly_amt, i.current_ind) for i in loan.incomes.all()]
183 liabilities = [
184 LiabilityItem(liab.monthly_payment_amt, liab.exclude_from_dti_ind, liab.omit_ind)
185 for liab in loan.liabilities.all()
186 ]
187 housing = [HousingItem(h.payment_amt, h.timing_type) for h in loan.housing_expenses.all()]
189 try:
190 txn = loan.transaction
191 except Loan.transaction.RelatedObjectDoesNotExist:
192 txn = None
193 fallback_pi = (
194 monthly_pi(txn.base_loan_amt, txn.note_rate_percent or 0, txn.term_months)
195 if txn is not None
196 else None
197 )
199 result = qualify(incomes, liabilities, housing, fallback_pi=fallback_pi)
200 return {
201 "monthly_income": result.monthly_income,
202 "monthly_debt": result.monthly_debt,
203 "housing_expense": result.housing_expense,
204 "housing_dti_pct": result.housing_dti_pct,
205 "total_dti_pct": result.total_dti_pct,
206 }
209# --- 1003 extension: loan-FK collections (details/08) --------------------------
210_register_crud(
211 "down-payment-sources", DownPaymentSource,
212 DownPaymentSourceOut, DownPaymentSourceIn, DownPaymentSourcePatch,
213)
214_register_crud("gift-grants", GiftGrant, GiftGrantOut, GiftGrantIn, GiftGrantPatch)
215_register_crud(
216 "loan-identifiers", LoanIdentifier, LoanIdentifierOut, LoanIdentifierIn, LoanIdentifierPatch
217)
220def _borrower(loan, borrower_id: int):
221 borrower = loan.borrowers.filter(pk=borrower_id).first()
222 if borrower is None: 222 ↛ 223line 222 didn't jump to line 223 because the condition on line 222 was never true
223 raise DomainError([AppError("NOT_FOUND", "Borrower not found on this loan")])
224 return borrower
227# --- 1003 extension: borrower residences (borrower-FK) -------------------------
228@router.get("/{guid}/residences", response=list[ResidenceOut], auth=requires("LOAN_VIEW"))
229def list_residences(request, guid: UUID):
230 loan = _scoped_loan(request, guid)
231 return list(Residence.objects.filter(borrower__loan=loan))
234@router.post("/{guid}/residences", response={201: ResidenceOut}, auth=requires("LOAN_EDIT"))
235def create_residence(request, guid: UUID, payload: ResidenceIn):
236 loan = _scoped_loan(request, guid)
237 borrower = _borrower(loan, payload.borrower_id)
238 data = payload.dict()
239 data.pop("borrower_id")
240 return 201, Residence.objects.create(borrower=borrower, **data)
243@router.patch("/{guid}/residences/{item_id}", response=ResidenceOut, auth=requires("LOAN_EDIT"))
244def update_residence(request, guid: UUID, item_id: int, payload: ResidencePatch):
245 loan = _scoped_loan(request, guid)
246 res = Residence.objects.filter(pk=item_id, borrower__loan=loan).first()
247 if res is None:
248 raise DomainError([AppError("NOT_FOUND", "Residence not found")])
249 for attr, value in payload.model_dump(exclude_unset=True).items():
250 setattr(res, attr, value)
251 res.save()
252 return res
255@router.delete("/{guid}/residences/{item_id}", response={204: None}, auth=requires("LOAN_EDIT"))
256def delete_residence(request, guid: UUID, item_id: int):
257 loan = _scoped_loan(request, guid)
258 res = Residence.objects.filter(pk=item_id, borrower__loan=loan).first()
259 if res is None:
260 raise DomainError([AppError("NOT_FOUND", "Residence not found")])
261 res.delete()
262 return 204, None
265# --- 1003 extension: per-borrower declarations + demographics (OneToOne) -------
266@router.get(
267 "/{guid}/declarations/{borrower_id}", response=DeclarationOut, auth=requires("LOAN_VIEW")
268)
269def get_declaration(request, guid: UUID, borrower_id: int):
270 loan = _scoped_loan(request, guid)
271 decl, _ = Declaration.objects.get_or_create(borrower=_borrower(loan, borrower_id))
272 return decl
275@router.patch(
276 "/{guid}/declarations/{borrower_id}", response=DeclarationOut, auth=requires("LOAN_EDIT")
277)
278def update_declaration(request, guid: UUID, borrower_id: int, payload: DeclarationPatch):
279 loan = _scoped_loan(request, guid)
280 decl, _ = Declaration.objects.get_or_create(borrower=_borrower(loan, borrower_id))
281 for attr, value in payload.model_dump(exclude_unset=True).items():
282 setattr(decl, attr, value)
283 decl.save()
284 return decl
287@router.get(
288 "/{guid}/demographics/{borrower_id}", response=DemographicsOut, auth=requires("LOAN_VIEW")
289)
290def get_demographics(request, guid: UUID, borrower_id: int):
291 loan = _scoped_loan(request, guid)
292 demo, _ = Demographics.objects.get_or_create(borrower=_borrower(loan, borrower_id))
293 return demo
296@router.patch(
297 "/{guid}/demographics/{borrower_id}", response=DemographicsOut, auth=requires("LOAN_EDIT")
298)
299def update_demographics(request, guid: UUID, borrower_id: int, payload: DemographicsPatch):
300 loan = _scoped_loan(request, guid)
301 demo, _ = Demographics.objects.get_or_create(borrower=_borrower(loan, borrower_id))
302 for attr, value in payload.model_dump(exclude_unset=True).items():
303 setattr(demo, attr, value)
304 demo.save()
305 return demo
308# --- 1003 extension: subject property (OneToOne loan) --------------------------
309@router.get("/{guid}/subject-property", response=SubjectPropertyOut, auth=requires("LOAN_VIEW"))
310def get_subject_property(request, guid: UUID):
311 loan = _scoped_loan(request, guid)
312 sp, _ = SubjectProperty.objects.get_or_create(loan=loan)
313 return sp
316@router.patch("/{guid}/subject-property", response=SubjectPropertyOut, auth=requires("LOAN_EDIT"))
317def update_subject_property(request, guid: UUID, payload: SubjectPropertyPatch):
318 loan = _scoped_loan(request, guid)
319 sp, _ = SubjectProperty.objects.get_or_create(loan=loan)
320 data = payload.model_dump(exclude_unset=True)
321 try:
322 for attr, value in data.items():
323 setattr(sp, attr, value)
324 sp.save()
325 if "county" in data and data["county"] is not None:
326 loan.county = sp.county
327 loan.save(update_fields=["county"])
328 except (DataError, IntegrityError) as exc:
329 raise DomainError(
330 [AppError("VALIDATION_ERROR", f"Invalid subject property data: {exc}")]
331 ) from exc
332 return sp