Coverage for apps/core_los/custom_fields_api.py: 69%

1from __future__ import annotations 

2 

3import re 

4from typing import Any 

5from uuid import UUID 

6 

7from ninja import Body, Router, Schema 

8 

9from apps.accounts.auth import requires 

10from apps.core_los.models import ( 

11 CustomFieldDefinition, 

12 CustomTrackerDefinition, 

13 Loan, 

14 LoanCustomFieldValues, 

15) 

16from apps.workflow.models import LoanTracker 

17from apps.workflow.schemas import TrackerOut 

18from domain.result import AppError, DomainError 

19 

20custom_fields_router = Router(tags=["custom-fields-trackers"]) 

21loan_custom_fields_router = Router(tags=["loan-custom-fields"]) 

22 

23DEFAULT_TRACKER_STATUSES = ["NotStarted", "InProgress", "Completed", "NA", "Overdue"] 

24 

25 

26def _key_of(label: str) -> str: 

27 cleaned = re.sub(r"[^a-zA-Z0-9]+", "_", label.strip().lower()) 

28 return cleaned.strip("_") 

29 

30 

31class CustomFieldOut(Schema): 

32 id: int 

33 key: str 

34 label: str 

35 field_type: str 

36 options: list[str] | None = None 

37 section: str 

38 active: bool 

39 

40 

41class NewCustomFieldInput(Schema): 

42 label: str 

43 field_type: str 

44 section: str | None = "Loan" 

45 options: list[str] | None = None 

46 key: str | None = None 

47 

48 

49class CustomFieldPatch(Schema): 

50 label: str | None = None 

51 field_type: str | None = None 

52 section: str | None = None 

53 options: list[str] | None = None 

54 active: bool | None = None 

55 

56 

57class CustomTrackerOut(Schema): 

58 id: int 

59 key: str 

60 label: str 

61 category: str 

62 applies_to: str 

63 statuses: list[str] = [] 

64 active: bool 

65 

66 

67class NewTrackerInput(Schema): 

68 label: str 

69 category: str | None = "Custom" 

70 applies_to: str | None = "AllLoans" 

71 statuses: list[str] | None = None 

72 

73 

74class StatusDateOut(Schema): 

75 stage: str 

76 at: str 

77 

78 

79class AddCustomTrackerIn(Schema): 

80 def_id: int 

81 

82 

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

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

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

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

87 return loan 

88 

89 

90# --- Org-level Custom Fields -------------------------------------------------- 

91@custom_fields_router.get("/custom-fields", response=list[CustomFieldOut], auth=requires("LOAN_VIEW")) 

92def list_custom_fields(request): 

93 return list(CustomFieldDefinition.objects.all()) 

94 

95 

96@custom_fields_router.post( 

97 "/custom-fields", response={201: CustomFieldOut}, auth=requires("LOAN_EDIT") 

98) 

99def create_custom_field(request, payload: NewCustomFieldInput): 

100 label = payload.label.strip() if payload.label else "" 

101 if not label: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 raise DomainError([AppError("VALIDATION", "Field label is required.", "label")]) 

103 key = payload.key.strip() if payload.key else _key_of(label) 

104 if CustomFieldDefinition.objects.filter(key=key).exists(): 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true

105 raise DomainError([AppError("VALIDATION", f'Field key "{key}" already exists.', "key")]) 

106 

107 options = None 

108 if payload.field_type == "select" and payload.options: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 options = [o.strip() for o in payload.options if o.strip()] 

110 

111 cf = CustomFieldDefinition.objects.create( 

112 key=key, 

113 label=label, 

114 field_type=payload.field_type, 

115 options=options, 

116 section=payload.section.strip() if payload.section else "Loan", 

117 active=True, 

118 ) 

119 return 201, cf 

120 

121 

122@custom_fields_router.patch( 

123 "/custom-fields/{field_id}", response=CustomFieldOut, auth=requires("LOAN_EDIT") 

124) 

125def update_custom_field(request, field_id: int, payload: CustomFieldPatch): 

126 cf = CustomFieldDefinition.objects.filter(pk=field_id).first() 

127 if cf is None: 

128 raise DomainError([AppError("NOT_FOUND", f"Custom field {field_id} not found.")]) 

129 

130 data = payload.dict(exclude_unset=True) 

131 if "label" in data and data["label"]: 

132 cf.label = data["label"].strip() 

133 if "field_type" in data and data["field_type"]: 

134 cf.field_type = data["field_type"] 

135 if "section" in data and data["section"]: 

136 cf.section = data["section"].strip() 

137 if "options" in data: 

138 cf.options = data["options"] 

139 if "active" in data and data["active"] is not None: 

140 cf.active = data["active"] 

141 

142 cf.save() 

143 return cf 

144 

145 

146@custom_fields_router.delete( 

147 "/custom-fields/{field_id}", response={204: None}, auth=requires("LOAN_EDIT") 

148) 

149def delete_custom_field(request, field_id: int): 

150 cf = CustomFieldDefinition.objects.filter(pk=field_id).first() 

151 if cf is None: 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true

152 raise DomainError([AppError("NOT_FOUND", f"Custom field {field_id} not found.")]) 

153 cf.delete() 

154 return 204, None 

155 

156 

157# --- Org-level Custom Trackers ------------------------------------------------ 

158@custom_fields_router.get( 

159 "/custom-trackers", response=list[CustomTrackerOut], auth=requires("LOAN_VIEW") 

160) 

161def list_custom_trackers(request): 

162 return list(CustomTrackerDefinition.objects.all()) 

163 

164 

165@custom_fields_router.post( 

166 "/custom-trackers", response={201: CustomTrackerOut}, auth=requires("LOAN_EDIT") 

167) 

168def create_custom_tracker(request, payload: NewTrackerInput): 

169 label = payload.label.strip() if payload.label else "" 

170 if not label: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true

171 raise DomainError([AppError("VALIDATION", "Tracker label is required.", "label")]) 

172 key = _key_of(label) 

173 if CustomTrackerDefinition.objects.filter(key=key).exists(): 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true

174 raise DomainError([AppError("VALIDATION", f'Tracker key "{key}" already exists.', "key")]) 

175 

176 statuses = payload.statuses if payload.statuses else DEFAULT_TRACKER_STATUSES 

177 

178 tracker_def = CustomTrackerDefinition.objects.create( 

179 key=key, 

180 label=label, 

181 category=payload.category.strip() if payload.category else "Custom", 

182 applies_to=payload.applies_to if payload.applies_to else "AllLoans", 

183 statuses=statuses, 

184 active=True, 

185 ) 

186 return 201, tracker_def 

187 

188 

189@custom_fields_router.delete( 

190 "/custom-trackers/{tracker_id}", response={204: None}, auth=requires("LOAN_EDIT") 

191) 

192def delete_custom_tracker(request, tracker_id: int): 

193 tracker_def = CustomTrackerDefinition.objects.filter(pk=tracker_id).first() 

194 if tracker_def is None: 

195 raise DomainError([AppError("NOT_FOUND", f"Custom tracker {tracker_id} not found.")]) 

196 tracker_def.delete() 

197 return 204, None 

198 

199 

200# --- Loan-level Custom Fields & Trackers -------------------------------------- 

201@loan_custom_fields_router.get("/{guid}/custom-fields", auth=requires("LOAN_VIEW")) 

202def get_loan_custom_fields(request, guid: UUID): 

203 loan = _scoped_loan(request, guid) 

204 bag = getattr(loan, "custom_fields_bag", None) 

205 return bag.values if bag else {} 

206 

207 

208@loan_custom_fields_router.put("/{guid}/custom-fields", auth=requires("LOAN_EDIT")) 

209def save_loan_custom_fields(request, guid: UUID, payload: dict[str, Any] = Body(...)): 

210 loan = _scoped_loan(request, guid) 

211 bag, _ = LoanCustomFieldValues.objects.get_or_create(loan=loan) 

212 bag.values = payload 

213 bag.save() 

214 return bag.values 

215 

216 

217@loan_custom_fields_router.get( 

218 "/{guid}/status-dates", response=list[StatusDateOut], auth=requires("LOAN_VIEW") 

219) 

220def get_status_dates(request, guid: UUID): 

221 loan = _scoped_loan(request, guid) 

222 history = loan.stage_history.order_by("entered_at") 

223 if history.exists(): 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true

224 return [{"stage": h.stage, "at": h.entered_at.isoformat()} for h in history] 

225 return [{"stage": loan.stage, "at": loan.created_at.isoformat()}] 

226 

227 

228@loan_custom_fields_router.post( 

229 "/{guid}/custom-trackers", response={201: TrackerOut}, auth=requires("LOAN_EDIT") 

230) 

231def add_custom_tracker(request, guid: UUID, payload: AddCustomTrackerIn): 

232 loan = _scoped_loan(request, guid) 

233 tracker_def = CustomTrackerDefinition.objects.filter(pk=payload.def_id).first() 

234 if tracker_def is None: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true

235 raise DomainError([AppError("NOT_FOUND", f"Custom tracker {payload.def_id} not found.")]) 

236 

237 # Check applies_to constraint 

238 if tracker_def.applies_to != "AllLoans" and tracker_def.applies_to != loan.purpose: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 raise DomainError( 

240 [AppError("VALIDATION", f'Tracker "{tracker_def.label}" does not apply to {loan.purpose} loans.')] 

241 ) 

242 

243 tracker_key = f"custom_{tracker_def.key}" 

244 if loan.trackers.filter(key=tracker_key).exists(): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true

245 raise DomainError( 

246 [AppError("VALIDATION", f'Tracker "{tracker_def.label}" is already on this loan.')] 

247 ) 

248 

249 tracker = LoanTracker.objects.create( 

250 loan=loan, 

251 key=tracker_key, 

252 label=tracker_def.label, 

253 category=tracker_def.category, 

254 status="NotStarted", 

255 ) 

256 return 201, tracker