Coverage for apps/underwriting/api.py: 90%

1"""Credit router: order credit, scores, liability import (mounted under /api/loans).""" 

2 

3from uuid import UUID 

4 

5from django.db import transaction as db_transaction 

6from ninja import Router 

7 

8from apps.accounts.auth import ensure_staff_scope, requires 

9from apps.application_1003.models import Liability 

10from apps.core_los.models import Loan 

11from apps.underwriting.models import CreditOrder, CreditScore 

12from apps.underwriting.providers import get_credit_provider 

13from apps.underwriting.schemas import ( 

14 CreditOrderCreate, 

15 CreditOrderOut, 

16 CreditScoreOut, 

17 ImportLiabilitiesIn, 

18 ImportLiabilitiesOut, 

19) 

20from domain.result import AppError, DomainError 

21 

22router = Router(tags=["credit"]) 

23 

24 

25def _scoped_loan(request, guid: UUID) -> Loan: 

26 loan = Loan.objects.filter(guid=guid).first() 

27 if loan is None: 27 ↛ 28line 27 didn't jump to line 28 because the condition on line 27 was never true

28 raise DomainError([AppError("NOT_FOUND", "Loan not found")]) 

29 ensure_staff_scope(request.principal, loan) 

30 return loan 

31 

32 

33def _order_payload(order: CreditOrder) -> dict: 

34 # Scalar order fields from model_dump; nested scores as ORM objects so Ninja 

35 # serializes them once via CreditScoreOut (double-dumping drops the FK alias). 

36 data = CreditOrderOut.from_orm(order).model_dump() 

37 data["provider_status"] = "live" 

38 data["scores"] = list(order.scores.select_related("borrower").all()) 

39 return data 

40 

41 

42@router.post("/{guid}/credit-order", response={201: CreditOrderOut}, auth=requires("LOAN_EDIT")) 

43def order_credit(request, guid: UUID, payload: CreditOrderCreate): 

44 loan = _scoped_loan(request, guid) 

45 borrowers = list(loan.borrowers.all()) 

46 provider = get_credit_provider(org=loan.org_unit) 

47 with db_transaction.atomic(): 

48 order = CreditOrder.objects.create( 

49 loan=loan, 

50 credit_company_id=payload.credit_company_id, 

51 request_type=payload.request_type, 

52 status="Ordered", 

53 ) 

54 result = provider.order(loan, borrowers, payload.request_type) 

55 order.report_id = result.report_id 

56 order.status = "Completed" 

57 order.save(update_fields=["report_id", "status"]) 

58 for score in result.scores: 

59 CreditScore.objects.create( 

60 order=order, 

61 borrower_id=score.borrower_id, 

62 equifax=score.equifax, 

63 experian=score.experian, 

64 transunion=score.transunion, 

65 mid=score.mid, 

66 ) 

67 for tl in result.tradelines: 

68 liability = Liability.objects.create( 

69 loan=loan, 

70 liability_type=tl.liability_type, 

71 creditor_name=tl.creditor_name, 

72 monthly_payment_amt=tl.monthly_payment_amt, 

73 unpaid_balance_amt=tl.unpaid_balance_amt, 

74 ) 

75 liability.borrowers.set([tl.borrower_id]) 

76 if not loan.credit_pulled: 76 ↛ 79line 76 didn't jump to line 79

77 loan.credit_pulled = True 

78 loan.save(update_fields=["credit_pulled"]) 

79 return 201, _order_payload(order) 

80 

81 

82@router.get("/{guid}/credit-order", response=list[CreditOrderOut], auth=requires("LOAN_VIEW")) 

83def list_credit_orders(request, guid: UUID): 

84 loan = _scoped_loan(request, guid) 

85 return [_order_payload(o) for o in loan.credit_orders.all().order_by("-created_at")] 

86 

87 

88@router.get("/{guid}/credit-score", response=list[CreditScoreOut], auth=requires("LOAN_VIEW")) 

89def list_credit_scores(request, guid: UUID): 

90 loan = _scoped_loan(request, guid) 

91 return list(CreditScore.objects.filter(order__loan=loan)) 

92 

93 

94@router.post( 

95 "/{guid}/import-liabilities", response=ImportLiabilitiesOut, auth=requires("LOAN_EDIT") 

96) 

97def import_liabilities(request, guid: UUID, payload: ImportLiabilitiesIn): 

98 loan = _scoped_loan(request, guid) 

99 count = loan.liabilities.count() 

100 return {"imported": count}