Coverage for apps/core_los/public_api.py: 76%

1"""Public Partner API router (mounted at /api/public). 

2 

3Endpoints for third-party developer integration: 

4 POST /api/public/auth/token -> PublicTokenOut 

5 GET /api/public/loans -> list[PublicLoanSummaryOut] 

6 POST /api/public/loans -> 201 PublicLoanCreatedOut 

7""" 

8 

9from __future__ import annotations 

10 

11import random 

12from decimal import Decimal 

13from uuid import uuid4 

14 

15from django.db import transaction 

16from ninja import Router 

17from ninja.security import HttpBearer 

18 

19from apps.accounts.models import ApiClient, OrgUnit, User 

20from apps.application_1003.models import Borrower 

21from apps.core_los.models import Loan, LoanTransaction 

22from apps.core_los.public_schemas import ( 

23 PublicLoanCreatedOut, 

24 PublicLoanInput, 

25 PublicLoanSummaryOut, 

26 PublicTokenOut, 

27 PublicTokenRequest, 

28) 

29from apps.integrations.encryption import decrypt_secret 

30from domain.result import AppError, DomainError 

31 

32router = Router(tags=["developer-public-api"]) 

33 

34# In-memory mapping of partner bearer tokens -> org_unit_id 

35_PARTNER_TOKENS: dict[str, int] = {} 

36 

37 

38class PartnerAuth(HttpBearer): 

39 def authenticate(self, request, token: str) -> int | None: 

40 if token in _PARTNER_TOKENS: 40 ↛ 43line 40 didn't jump to line 43 because the condition on line 40 was always true

41 return _PARTNER_TOKENS[token] 

42 # In test/dev, also accept raw demo token if passed directly 

43 if token == "lat_demo": 

44 org = OrgUnit.objects.filter(is_headquarters=True).first() or OrgUnit.objects.first() 

45 if org: 

46 return org.id 

47 raise DomainError([AppError("UNAUTHORIZED", "Missing or invalid access token.")]) 

48 

49 

50def _resolve_default_org() -> OrgUnit: 

51 org = OrgUnit.objects.filter(is_headquarters=True).first() or OrgUnit.objects.first() 

52 if org is None: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 org = OrgUnit.objects.create(name="Headquarters", is_headquarters=True) 

54 return org 

55 

56 

57@router.post("/auth/token", response=PublicTokenOut) 

58def request_token(request, payload: PublicTokenRequest): 

59 """Exchange partner client credentials for a bearer token.""" 

60 client_id = payload.client_id.strip() 

61 client_secret = payload.client_secret.strip() 

62 

63 valid_org_id: int | None = None 

64 

65 # 1. Standard pre-seeded demo client (Arive lumen-demo / lumen-secret) 

66 if client_id == "lumen-demo" and client_secret == "lumen-secret": 

67 org = _resolve_default_org() 

68 valid_org_id = org.id 

69 else: 

70 # 2. Database ApiClient 

71 client = ApiClient.objects.filter(client_id=client_id).first() 

72 if client: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 decrypted = decrypt_secret(client.secret_enc) if client.secret_enc else "" 

74 if decrypted == client_secret or client.secret_enc == client_secret: 

75 valid_org_id = client.org_unit_id or _resolve_default_org().id 

76 

77 if valid_org_id is None: 

78 raise DomainError([AppError("UNAUTHORIZED", "Invalid client credentials.")]) 

79 

80 token = f"lat_{uuid4().hex}" 

81 _PARTNER_TOKENS[token] = valid_org_id 

82 return {"access_token": token, "expires_in": 3600} 

83 

84 

85@router.get("/loans", response=list[PublicLoanSummaryOut], auth=PartnerAuth()) 

86def list_partner_loans(request): 

87 """List loans scoped to the authenticated partner organization.""" 

88 org_id: int = request.auth 

89 loans = Loan.objects.filter(org_unit_id=org_id).order_by("-created_at") 

90 

91 out: list[dict] = [] 

92 for loan in loans: 

93 primary_b = loan.borrowers.filter(is_primary=True).first() or loan.borrowers.first() 

94 borrower_name = f"{primary_b.first_name} {primary_b.last_name}" if primary_b else "Unknown" 

95 

96 txn = getattr(loan, "transaction", None) 

97 amt = float(txn.base_loan_amt) if txn and txn.base_loan_amt else 0.0 

98 ltv = float(txn.ltv_ratio_percent) if txn and txn.ltv_ratio_percent else 80.0 

99 

100 rate_lock = getattr(loan, "rate_lock", None) 

101 rate = float(rate_lock.locked_rate) if rate_lock and rate_lock.locked_rate else None 

102 prod = rate_lock.product.name if rate_lock and rate_lock.product else None 

103 lock_status = "Locked" if rate_lock else "Not Locked" 

104 

105 out.append( 

106 { 

107 "arive_loan_no": loan.arive_loan_no, 

108 "guid": str(loan.guid), 

109 "borrower_name": borrower_name, 

110 "state_code": loan.state_code or "CA", 

111 "county": loan.county or None, 

112 "stage": loan.stage or "APPLICATION_INTAKE", 

113 "purpose": loan.purpose or "Purchase", 

114 "amount": amt, 

115 "ltv": ltv, 

116 "rate": rate, 

117 "product": prod, 

118 "lock": lock_status, 

119 "open_tasks": loan.tasks.filter(completed=False).count(), 

120 } 

121 ) 

122 return out 

123 

124 

125@router.post("/loans", response={201: PublicLoanCreatedOut}, auth=PartnerAuth()) 

126def create_partner_loan(request, payload: PublicLoanInput): 

127 """Create a new loan via partner API.""" 

128 org_id: int = request.auth 

129 

130 lo = User.objects.filter(pk=payload.loan_officer_id).first() 

131 if lo is None: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 lo = User.objects.filter(org_unit_id=org_id).first() or User.objects.first() 

133 if lo is None: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 raise DomainError([AppError("NOT_FOUND", "No valid loan officer found.")]) 

135 

136 # Generate next loan number 

137 rand_seq = random.randint(10000, 99999) 

138 loan_no = f"LMN-{rand_seq}" 

139 while Loan.objects.filter(arive_loan_no=loan_no).exists(): 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 rand_seq = random.randint(10000, 99999) 

141 loan_no = f"LMN-{rand_seq}" 

142 

143 with transaction.atomic(): 

144 loan = Loan.objects.create( 

145 arive_loan_no=loan_no, 

146 org_unit_id=org_id, 

147 loan_officer=lo, 

148 state_code=payload.state_code, 

149 occupancy=payload.occupancy, 

150 purpose=payload.purpose, 

151 stage="APPLICATION_INTAKE", 

152 ) 

153 Borrower.objects.create( 

154 loan=loan, 

155 first_name=payload.borrower.first_name, 

156 last_name=payload.borrower.last_name, 

157 email=payload.borrower.email, 

158 is_primary=True, 

159 ) 

160 LoanTransaction.objects.create( 

161 loan=loan, 

162 base_loan_amt=Decimal("400000.00"), 

163 note_amt=Decimal("400000.00"), 

164 purchase_price_amt=Decimal("500000.00"), 

165 appraised_value_amt=Decimal("500000.00"), 

166 ltv_ratio_percent=Decimal("80.00"), 

167 ) 

168 

169 return 201, {"arive_loan_no": loan.arive_loan_no, "guid": loan.guid}