Coverage for apps/core_los/report_builder_api.py: 93%
90 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""Report Builder: field catalog, saved custom reports, run, and CSV export
2(mounted at "").
3"""
5from __future__ import annotations
7from dataclasses import asdict
8from decimal import Decimal
10from django.http import HttpResponse
11from ninja import Router
13from apps.accounts.auth import requires, staff_loan_queryset
14from apps.core_los.models import CustomReport
15from apps.core_los.report_builder_schemas import (
16 ReportDefInputSchema,
17 ReportDefOut,
18 ReportFieldCatalogOut,
19 ReportResultOut,
20)
21from apps.core_los.reports_api import _is_locked
22from apps.core_los.schemas import _safe_transaction
23from domain.enums import OCCUPANCY_LABELS, STAGE_LABELS, LoanStage, OccupancyType
24from domain.report_builder import (
25 METRICS,
26 REPORT_FIELDS,
27 ReportDefInput,
28 ReportFilter,
29 run_report,
30 to_csv,
31 validate_report_def,
32)
33from domain.reports import is_active_loan
34from domain.result import AppError, DomainError
36router = Router(tags=["report-builder"])
39def _report_rows(request) -> list[dict]:
40 qs = staff_loan_queryset(request.principal, "all").select_related(
41 "transaction", "rate_lock", "rate_lock__product", "loan_officer"
42 )
43 rows = []
44 for loan in qs:
45 if not is_active_loan(loan.stage):
46 continue
47 txn = _safe_transaction(loan)
48 primary = loan.borrowers.filter(is_primary=True).first()
49 value = Decimal("0.00")
50 if txn: 50 ↛ 55line 50 didn't jump to line 55 because the condition on line 50 was always true
51 if txn.appraised_value_amt is not None: 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 value = txn.appraised_value_amt
53 elif txn.purchase_price_amt is not None: 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true
54 value = txn.purchase_price_amt
55 rows.append(
56 {
57 "loan_number": loan.arive_loan_no,
58 "borrower_name": f"{primary.first_name} {primary.last_name}" if primary else "—",
59 "stage": STAGE_LABELS[LoanStage(loan.stage)],
60 "purpose": loan.purpose,
61 "occupancy": OCCUPANCY_LABELS[OccupancyType(loan.occupancy)],
62 "state_code": loan.state_code,
63 "county": loan.county,
64 "loan_officer": (
65 f"{loan.loan_officer.first_name} {loan.loan_officer.last_name}"
66 if loan.loan_officer else None
67 ),
68 "amount": txn.base_loan_amt if txn else Decimal("0.00"),
69 "value": value,
70 "ltv": (
71 txn.ltv_ratio_percent
72 if txn and txn.ltv_ratio_percent is not None
73 else Decimal("0.00")
74 ),
75 "rate": txn.note_rate_percent if txn else None,
76 "lock": "Locked" if _is_locked(loan) else "NotLocked",
77 "conditions": loan.conditions.count(),
78 "tasks": loan.tasks.filter(completed=False).count(),
79 "month": loan.created_at.strftime("%Y-%m"),
80 "created_at": loan.created_at.date().isoformat(),
81 }
82 )
83 return rows
86def _to_domain_def(payload: ReportDefInputSchema) -> ReportDefInput:
87 return ReportDefInput(
88 name=payload.name,
89 columns=list(payload.columns),
90 filters=[ReportFilter(f.field, f.op, f.value) for f in payload.filters],
91 group_by=payload.group_by or None,
92 metrics=list(payload.metrics),
93 sort={"field": payload.sort.field, "dir": payload.sort.dir} if payload.sort else None,
94 )
97def _report_def_from_model(report: CustomReport) -> ReportDefInput:
98 return ReportDefInput(
99 name=report.name,
100 columns=list(report.columns or []),
101 filters=[ReportFilter(f["field"], f["op"], f["value"]) for f in (report.filters or [])],
102 group_by=report.group_by,
103 metrics=list(report.metrics or []),
104 sort=report.sort,
105 )
108def _validated_def(payload: ReportDefInputSchema) -> ReportDefInput:
109 d = _to_domain_def(payload)
110 result = validate_report_def(d)
111 if not result.ok:
112 raise DomainError(result.errors)
113 return d
116def _get_report(report_id: int) -> CustomReport:
117 report = CustomReport.objects.filter(pk=report_id).first()
118 if report is None: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true
119 raise DomainError([AppError("NOT_FOUND", f"Report {report_id} not found.")])
120 return report
123def _report_sort_key(report: CustomReport):
124 return (
125 0 if report.system else 1,
126 report.id if report.system else 0,
127 "" if report.system else report.name.lower(),
128 )
131@router.get("/report-fields", response=ReportFieldCatalogOut, auth=requires("LOAN_VIEW"))
132def report_fields(request):
133 return {"fields": [asdict(f) for f in REPORT_FIELDS], "metrics": [asdict(m) for m in METRICS]}
136@router.get("/custom-reports", response=list[ReportDefOut], auth=requires("LOAN_VIEW"))
137def list_custom_reports(request):
138 return sorted(CustomReport.objects.all(), key=_report_sort_key)
141@router.post("/custom-reports", response={201: ReportDefOut}, auth=requires("LOAN_VIEW"))
142def create_custom_report(request, payload: ReportDefInputSchema):
143 _validated_def(payload)
144 report = CustomReport.objects.create(
145 name=payload.name.strip(),
146 columns=list(payload.columns),
147 filters=[f.dict() for f in payload.filters],
148 group_by=payload.group_by or None,
149 metrics=list(payload.metrics),
150 sort=payload.sort.dict() if payload.sort else None,
151 system=False,
152 )
153 return 201, report
156@router.post("/custom-reports/run", response=ReportResultOut, auth=requires("LOAN_VIEW"))
157def run_custom_report(request, payload: ReportDefInputSchema):
158 # Must be registered before "/custom-reports/{report_id}" — Django's URL
159 # resolver matches the first pattern whose path shape fits and stops there,
160 # so a parameterized route registered first would swallow "run" as an id and
161 # return 405 instead of falling through to this one.
162 d = _validated_def(payload)
163 result = run_report(d, _report_rows(request))
164 return asdict(result)
167@router.patch("/custom-reports/{report_id}", response=ReportDefOut, auth=requires("LOAN_VIEW"))
168def update_custom_report(request, report_id: int, payload: ReportDefInputSchema):
169 report = _get_report(report_id)
170 _validated_def(payload)
171 report.name = payload.name.strip()
172 report.columns = list(payload.columns)
173 report.filters = [f.dict() for f in payload.filters]
174 report.group_by = payload.group_by or None
175 report.metrics = list(payload.metrics)
176 report.sort = payload.sort.dict() if payload.sort else None
177 report.save()
178 return report
181@router.delete("/custom-reports/{report_id}", response={204: None}, auth=requires("LOAN_VIEW"))
182def delete_custom_report(request, report_id: int):
183 report = _get_report(report_id)
184 if report.system:
185 raise DomainError(
186 [AppError("REPORT_PROTECTED", "Built-in reports can't be deleted — clone or edit.")]
187 )
188 report.delete()
189 return 204, None
192@router.get("/custom-reports/{report_id}/csv", auth=requires("LOAN_VIEW"))
193def custom_report_csv(request, report_id: int):
194 report = _get_report(report_id)
195 result = run_report(_report_def_from_model(report), _report_rows(request))
196 response = HttpResponse(to_csv(result), content_type="text/csv")
197 response["Content-Disposition"] = f'attachment; filename="report-{report_id}.csv"'
198 return response