Coverage for app/reports/routes.py: 100%

1import csv 

2import io 

3from datetime import date as _date 

4from typing import Any 

5 

6from flask import ( # pyright: ignore[reportMissingImports] 

7 Blueprint, 

8 Response, 

9 abort, 

10 render_template, 

11 request, 

12 session, 

13) 

14from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports] 

15from flask_babel import gettext as _ # pyright: ignore[reportMissingImports] 

16from models import Aircraft, TenantUser, db # pyright: ignore[reportMissingImports] 

17from utils import ( # pyright: ignore[reportMissingImports] 

18 login_required, 

19 user_can_access_aircraft, 

20) 

21 

22from reports.utilization import ( # pyright: ignore[reportMissingImports] 

23 DEFAULT_PERIOD_MONTHS, 

24 PERIOD_OPTIONS, 

25 compute_utilization_report, 

26 resolve_period, 

27) 

28 

29reports_bp = Blueprint("reports", __name__) 

30 

31 

32def _tenant_id() -> int: 

33 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first() 

34 if not tu: 

35 abort(403) 

36 return int(tu.tenant_id) 

37 

38 

39def _get_aircraft_or_404(aircraft_id: int) -> Aircraft: 

40 ac = db.session.get(Aircraft, aircraft_id) 

41 if ( 

42 not ac 

43 or ac.tenant_id != _tenant_id() 

44 or not user_can_access_aircraft(aircraft_id) 

45 ): 

46 abort(404) 

47 return ac 

48 

49 

50def _resolve_report_range( 

51 today: _date, 

52) -> tuple[_date | None, _date, int | None, str, str]: 

53 """Read `?from=&to=` (an arbitrary policy year) or `?period=` (a rolling 

54 preset, default 12 months) from the query string. Returns 

55 (period_start, period_end, period_months, from_raw, to_raw) — 

56 period_months is None when a custom range was used, so the template 

57 knows which selector mode was active.""" 

58 from_raw = request.args.get("from", "").strip() 

59 to_raw = request.args.get("to", "").strip() 

60 if from_raw and to_raw: 

61 try: 

62 custom_start = _date.fromisoformat(from_raw) 

63 custom_end = _date.fromisoformat(to_raw) 

64 except ValueError: 

65 pass 

66 else: 

67 if custom_start <= custom_end: 

68 return custom_start, custom_end, None, from_raw, to_raw 

69 

70 try: 

71 period_months = int(request.args.get("period", DEFAULT_PERIOD_MONTHS)) 

72 except ValueError: 

73 period_months = DEFAULT_PERIOD_MONTHS 

74 period_start, period_end = resolve_period(period_months, today) 

75 return period_start, period_end, period_months, from_raw, to_raw 

76 

77 

78@reports_bp.route("/aircraft/<aircraft_ref:aircraft_id>/reports/utilization") 

79@login_required 

80def utilization_report(aircraft_id: int) -> ResponseReturnValue: 

81 ac = _get_aircraft_or_404(aircraft_id) 

82 today = _date.today() 

83 period_start, period_end, period_months, from_raw, to_raw = _resolve_report_range( 

84 today 

85 ) 

86 report = compute_utilization_report(ac.id, period_start, period_end) 

87 

88 return render_template( 

89 "reports/utilization.html", 

90 aircraft=ac, 

91 report=report, 

92 period_months=period_months, 

93 period_options=PERIOD_OPTIONS, 

94 from_date=from_raw, 

95 to_date=to_raw, 

96 today=today.isoformat(), 

97 ) 

98 

99 

100@reports_bp.route("/aircraft/<aircraft_ref:aircraft_id>/reports/utilization.csv") 

101@login_required 

102def utilization_report_csv(aircraft_id: int) -> ResponseReturnValue: 

103 ac = _get_aircraft_or_404(aircraft_id) 

104 today = _date.today() 

105 period_start, period_end, _period_months, _from_raw, _to_raw = ( 

106 _resolve_report_range(today) 

107 ) 

108 report = compute_utilization_report(ac.id, period_start, period_end) 

109 

110 buf = io.StringIO() 

111 writer = csv.writer(buf) 

112 period_start_str = period_start.isoformat() if period_start else _("all time") 

113 period_label = f"{period_start_str} {_('to')} {period_end.isoformat()}" 

114 writer.writerow([_("Aircraft"), ac.registration]) 

115 writer.writerow([_("Period"), period_label]) 

116 writer.writerow([_("Export date"), today.isoformat()]) 

117 writer.writerow([]) 

118 

119 current = report["current"] 

120 previous = report["previous"] 

121 header = [_("Metric"), _("This period")] 

122 if previous is not None: 

123 header.append(_("Previous period")) 

124 writer.writerow(header) 

125 

126 def _fuel_str(fuel: dict[str, float]) -> str: 

127 if not fuel: 

128 return "0" 

129 return ", ".join(f"{qty:.1f} {unit}" for unit, qty in sorted(fuel.items())) 

130 

131 def _stat_row(stats: dict[str, Any], key: str) -> str: 

132 value = stats[key] 

133 if key == "fuel_added": 

134 return _fuel_str(value) 

135 if key in ("flight_count", "landings"): 

136 return str(value) 

137 return f"{value:.2f}" if key == "oil_added_l" else f"{value:.1f}" 

138 

139 metrics = [ 

140 (_("Flight hours"), "flight_hours"), 

141 (_("Engine hours"), "engine_hours"), 

142 (_("Number of flights"), "flight_count"), 

143 (_("Landings"), "landings"), 

144 (_("Fuel added"), "fuel_added"), 

145 (_("Oil added (L)"), "oil_added_l"), 

146 ] 

147 for label, key in metrics: 

148 row = [label, _stat_row(current, key)] 

149 if previous is not None: 

150 row.append(_stat_row(previous, key)) 

151 writer.writerow(row) 

152 

153 filename = f"{ac.registration}_utilization_{period_end.isoformat()}.csv" 

154 return Response( 

155 buf.getvalue(), 

156 mimetype="text/csv", 

157 headers={"Content-Disposition": f"attachment; filename={filename}"}, 

158 )