Coverage for domain/report_builder.py: 89%

1"""Pure Report Builder engine: field/metric catalogs, validation, filtering, 

2grouping, sorting, totals, and CSV export (add-report-builder-api). 

3 

4Port of arive-spec's report-builder logic (`src/domain/reports.ts`, 

5`reportRows`/`validateReportDef`/`runReport`/`reportToCsv` in `src/app.ts`), now 

6operating on plain row dicts rather than mock loan refs. Pure, side-effect-free. 

7Zero dependencies on Django or external databases. 

8""" 

9 

10from __future__ import annotations 

11 

12from dataclasses import dataclass, field 

13from decimal import ROUND_HALF_UP, Decimal 

14from typing import Any, cast 

15 

16from domain.result import AppError, Result, err, ok 

17 

18_CENTS = Decimal("0.01") 

19 

20 

21@dataclass(frozen=True) 

22class ReportField: 

23 key: str 

24 label: str 

25 type: str # "text" | "currency" | "percent" | "number" | "date" 

26 filterable: bool 

27 groupable: bool 

28 

29 

30# The 17-field catalog (reports.mock.ts REPORT_FIELDS). 

31REPORT_FIELDS: tuple[ReportField, ...] = ( 

32 ReportField("loan_number", "Loan #", "text", True, False), 

33 ReportField("borrower_name", "Borrower", "text", True, False), 

34 ReportField("stage", "Stage", "text", True, True), 

35 ReportField("purpose", "Purpose", "text", True, True), 

36 ReportField("occupancy", "Occupancy", "text", True, True), 

37 ReportField("state_code", "State", "text", True, True), 

38 ReportField("county", "County", "text", True, True), 

39 ReportField("loan_officer", "Loan officer", "text", True, True), 

40 ReportField("amount", "Loan amount", "currency", True, False), 

41 ReportField("value", "Property value", "currency", True, False), 

42 ReportField("ltv", "LTV", "percent", True, False), 

43 ReportField("rate", "Rate", "percent", True, False), 

44 ReportField("lock", "Lock status", "text", True, True), 

45 ReportField("conditions", "Conditions", "number", True, False), 

46 ReportField("tasks", "Open tasks", "number", True, False), 

47 ReportField("month", "Month opened", "text", True, True), 

48 ReportField("created_at", "Created", "date", True, False), 

49) 

50_FIELD_BY_KEY: dict[str, ReportField] = {f.key: f for f in REPORT_FIELDS} 

51 

52 

53@dataclass(frozen=True) 

54class Metric: 

55 key: str 

56 label: str 

57 type: str 

58 

59 

60METRICS: tuple[Metric, ...] = ( 

61 Metric("count", "Count", "number"), 

62 Metric("volume", "Total volume", "currency"), 

63 Metric("avg_amount", "Avg loan amount", "currency"), 

64 Metric("avg_rate", "Avg rate", "percent"), 

65 Metric("avg_ltv", "Avg LTV", "percent"), 

66) 

67_METRIC_BY_KEY: dict[str, Metric] = {m.key: m for m in METRICS} 

68DEFAULT_METRICS = ("count", "volume", "avg_rate") 

69 

70 

71@dataclass 

72class ReportFilter: 

73 field: str 

74 op: str # "eq" | "neq" | "contains" | "gte" | "lte" 

75 value: Any 

76 

77 

78@dataclass 

79class ReportDefInput: 

80 name: str 

81 columns: list[str] = field(default_factory=list) 

82 filters: list[ReportFilter] = field(default_factory=list) 

83 group_by: str | None = None 

84 metrics: list[str] = field(default_factory=list) 

85 sort: dict | None = None # {"field": str, "dir": "asc" | "desc"} 

86 

87 

88@dataclass 

89class ReportResult: 

90 mode: str # "detail" | "grouped" 

91 columns: list[dict] 

92 rows: list[dict] 

93 totals: dict | None 

94 row_count: int 

95 

96 

97def validate_report_def(d: ReportDefInput) -> Result: 

98 """Port of arive-spec `validateReportDef`.""" 

99 errors: list[AppError] = [] 

100 if not d.name or not d.name.strip(): 

101 errors.append(AppError("VALIDATION", "Report name is required.", "name")) 

102 if d.group_by: 

103 gf = _FIELD_BY_KEY.get(d.group_by) 

104 if gf is None or not gf.groupable: 

105 errors.append(AppError("VALIDATION", f'"{d.group_by}" can\'t be grouped.', "group_by")) 

106 elif not d.columns: 

107 errors.append( 

108 AppError("VALIDATION", "Pick at least one column (or a group-by).", "columns") 

109 ) 

110 for c in d.columns: 

111 if c not in _FIELD_BY_KEY: 

112 errors.append(AppError("VALIDATION", f'Unknown column "{c}".', "columns")) 

113 for f in d.filters: 

114 ff = _FIELD_BY_KEY.get(f.field) 

115 if ff is None or not ff.filterable: 115 ↛ 113line 115 didn't jump to line 113 because the condition on line 115 was always true

116 errors.append(AppError("VALIDATION", f'"{f.field}" can\'t be filtered.', "filters")) 

117 return err(*errors) if errors else ok() 

118 

119 

120def filter_matches(row: dict, f: ReportFilter) -> bool: 

121 """Filter match; gte/lte on a date field compares the raw ISO strings.""" 

122 v = row.get(f.field) 

123 is_date = (_FIELD_BY_KEY.get(f.field) or ReportField("", "", "", False, False)).type == "date" 

124 if f.op == "eq": 

125 return str(v if v is not None else "").lower() == str(f.value).lower() 

126 if f.op == "neq": 

127 return str(v if v is not None else "").lower() != str(f.value).lower() 

128 if f.op == "contains": 

129 return str(f.value).lower() in str(v if v is not None else "").lower() 

130 if f.op == "gte": 

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

132 return str(v if v is not None else "") >= str(f.value) 

133 return v is not None and Decimal(str(v)) >= Decimal(str(f.value)) 

134 if f.op == "lte": 134 ↛ 138line 134 didn't jump to line 138 because the condition on line 134 was always true

135 if is_date: 135 ↛ 137line 135 didn't jump to line 137 because the condition on line 135 was always true

136 return v is not None and str(v) <= str(f.value) 

137 return v is not None and Decimal(str(v)) <= Decimal(str(f.value)) 

138 return False 

139 

140 

141def _nums(rows: list[dict], key: str) -> list[Decimal]: 

142 return [Decimal(str(r.get(key) or 0)) for r in rows] 

143 

144 

145def metric_value(metric: str, rows: list[dict]) -> Decimal: 

146 if metric == "count": 

147 return Decimal(len(rows)) 

148 if not rows: 

149 return Decimal("0") 

150 if metric == "volume": 

151 return sum(_nums(rows, "amount"), Decimal("0")) 

152 if metric == "avg_amount": 

153 return (sum(_nums(rows, "amount"), Decimal("0")) / len(rows)).quantize( 

154 Decimal("1"), rounding=ROUND_HALF_UP 

155 ) 

156 if metric == "avg_rate": 

157 return (sum(_nums(rows, "rate"), Decimal("0")) / len(rows)).quantize( 

158 _CENTS, rounding=ROUND_HALF_UP 

159 ) 

160 if metric == "avg_ltv": 160 ↛ 164line 160 didn't jump to line 164 because the condition on line 160 was always true

161 return (sum(_nums(rows, "ltv"), Decimal("0")) / len(rows)).quantize( 

162 _CENTS, rounding=ROUND_HALF_UP 

163 ) 

164 raise ValueError(f"Unknown metric: {metric}") 

165 

166 

167def _column(key: str) -> dict: 

168 f = _FIELD_BY_KEY[key] 

169 return {"key": f.key, "label": f.label, "type": f.type} 

170 

171 

172def _metric_column(key: str) -> dict: 

173 m = _METRIC_BY_KEY[key] 

174 return {"key": m.key, "label": m.label, "type": m.type} 

175 

176 

177def run_report(d: ReportDefInput, rows: list[dict]) -> ReportResult: 

178 """Port of arive-spec `runReport`: filter, then group-with-metrics or 

179 detail-with-columns, then sort, then totals. 

180 

181 Assumes ``d`` already passed ``validate_report_def`` — callers raise on an 

182 invalid definition before reaching this function, same as every other pure 

183 domain module in this backend (e.g. ``domain.validation``). 

184 """ 

185 data = [r for r in rows if all(filter_matches(r, f) for f in d.filters)] 

186 

187 if d.group_by: 

188 metrics = d.metrics or list(DEFAULT_METRICS) 

189 groups: dict[str, list[dict]] = {} 

190 for r in data: 

191 key = str(r.get(d.group_by) or "—") 

192 groups.setdefault(key, []).append(r) 

193 

194 group_rows = [ 

195 {d.group_by: key, **{m: metric_value(m, rs) for m in metrics}} 

196 for key, rs in groups.items() 

197 ] 

198 sort_key = ( 

199 d.sort["field"] 

200 if d.sort and d.sort.get("field") in metrics 

201 else ("volume" if "volume" in metrics else metrics[0]) 

202 ) 

203 asc = bool(d.sort and d.sort.get("dir") == "asc") 

204 group_rows.sort(key=lambda r: cast(Decimal, r.get(sort_key) or Decimal("0")), reverse=not asc) 

205 

206 return ReportResult( 

207 mode="grouped", 

208 columns=[_column(d.group_by), *[_metric_column(m) for m in metrics]], 

209 rows=group_rows, 

210 totals={d.group_by: "Total", **{m: metric_value(m, data) for m in metrics}}, 

211 row_count=len(data), 

212 ) 

213 

214 if d.sort and d.sort.get("field") in _FIELD_BY_KEY: 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true

215 field_key = d.sort["field"] 

216 asc = d.sort.get("dir") == "asc" 

217 numeric = _FIELD_BY_KEY[field_key].type in ("number", "currency", "percent") 

218 

219 def sort_key(r: dict): 

220 v = r.get(field_key) 

221 if numeric: 

222 return Decimal(str(v)) if v is not None else Decimal("0") 

223 return str(v) if v is not None else "" 

224 

225 data = sorted(data, key=sort_key, reverse=not asc) 

226 

227 total_amount = sum(_nums(data, "amount"), Decimal("0")) if "amount" in d.columns else None 

228 return ReportResult( 

229 mode="detail", 

230 columns=[_column(c) for c in d.columns], 

231 rows=[{c: r.get(c) for c in d.columns} for r in data], 

232 totals=( 

233 {d.columns[0]: "Total", "amount": total_amount} if total_amount is not None else None 

234 ), 

235 row_count=len(data), 

236 ) 

237 

238 

239def _csv_escape(v: Any) -> str: 

240 s = "" if v is None else str(v) 

241 if any(ch in s for ch in (",", '"', "\n")): 

242 return '"' + s.replace('"', '""') + '"' 

243 return s 

244 

245 

246def to_csv(result: ReportResult) -> str: 

247 """Port of arive-spec `reportToCsv`: header labels, rows, then totals.""" 

248 lines = [",".join(_csv_escape(c["label"]) for c in result.columns)] 

249 for row in result.rows: 

250 lines.append(",".join(_csv_escape(row.get(c["key"])) for c in result.columns)) 

251 if result.totals: 

252 lines.append( 

253 ",".join(_csv_escape(result.totals.get(c["key"], "")) for c in result.columns) 

254 ) 

255 return "\n".join(lines)