Coverage for app/maintenance/routes.py: 100%
217 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-28 23:33 +0000
1from datetime import date as _date, timedelta
3from typing import Any
5from flask import ( # pyright: ignore[reportMissingImports]
6 Blueprint,
7 abort,
8 flash,
9 redirect,
10 render_template,
11 request,
12 session,
13 url_for,
14)
15from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
17from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
19from models import (
20 Aircraft,
21 MaintenanceRecord,
22 MaintenanceTrigger,
23 Role,
24 Snag,
25 TenantUser,
26 TriggerType,
27 db,
28) # pyright: ignore[reportMissingImports]
29from services.authorization import AuthorizationService # pyright: ignore[reportMissingImports]
30from utils import (
31 accessible_aircraft,
32 activity,
33 compute_aircraft_statuses,
34 login_required,
35 require_maint_access,
36 require_role,
37 user_can_access_aircraft,
38) # pyright: ignore[reportMissingImports]
40maintenance_bp = Blueprint("maintenance", __name__)
42_MAINT_ROLES = (Role.ADMIN, Role.OWNER, Role.MAINTENANCE)
45def _tenant_id() -> int:
46 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
47 if not tu:
48 abort(403)
49 return int(tu.tenant_id)
52def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
53 ac = db.session.get(Aircraft, aircraft_id)
54 if (
55 not ac
56 or ac.tenant_id != _tenant_id()
57 or not user_can_access_aircraft(aircraft_id)
58 ):
59 abort(404)
60 return ac
63def _get_trigger_or_404(aircraft: Aircraft, trigger_id: int) -> MaintenanceTrigger:
64 t = db.session.get(MaintenanceTrigger, trigger_id)
65 if not t or t.aircraft_id != aircraft.id:
66 abort(404)
67 return t
70# ── Fleet maintenance overview ────────────────────────────────────────────────
73@maintenance_bp.route("/maintenance")
74@login_required
75@require_maint_access
76def fleet_overview() -> ResponseReturnValue:
77 aircraft = accessible_aircraft(_tenant_id()).all()
78 aircraft_ids = [ac.id for ac in aircraft]
79 ac_by_id = {ac.id: ac for ac in aircraft}
80 hobbs_by_id = {ac.id: ac.total_engine_hours for ac in aircraft}
82 triggers = (
83 (
84 MaintenanceTrigger.query.filter(
85 MaintenanceTrigger.aircraft_id.in_(aircraft_ids)
86 ).all()
87 )
88 if aircraft_ids
89 else []
90 )
92 from datetime import date as _date_cls, datetime as _datetime
94 # Annotate each trigger with its status
95 trigger_rows = [
96 (t, t.status(hobbs_by_id.get(t.aircraft_id)), ac_by_id[t.aircraft_id])
97 for t in triggers
98 ]
100 # Sort: overdue → due_soon → ok; within status: calendar triggers by due_date asc,
101 # hours-based triggers (no reliable date) after calendar ones.
102 _status_order = {"overdue": 0, "due_soon": 1, "ok": 2}
103 _far_future = _date_cls(9999, 12, 31)
105 def _trigger_sort_key(row: Any) -> Any:
106 t, status, ac = row
107 due = (
108 t.due_date
109 if t.trigger_type == TriggerType.CALENDAR and t.due_date
110 else _far_future
111 )
112 return (_status_order[status], due)
114 trigger_rows.sort(key=_trigger_sort_key)
116 # Open grounding snags — oldest reported first (most overdue on top)
117 grounding_snags = (
118 (
119 Snag.query.filter(
120 Snag.aircraft_id.in_(aircraft_ids),
121 Snag.is_grounding.is_(True),
122 Snag.resolved_at.is_(None),
123 )
124 .order_by(Snag.reported_at.asc())
125 .all()
126 )
127 if aircraft_ids
128 else []
129 )
130 grounding_snag_rows = [(s, ac_by_id[s.aircraft_id]) for s in grounding_snags]
132 # Open non-grounding snags — oldest reported first
133 open_snags = (
134 (
135 Snag.query.filter(
136 Snag.aircraft_id.in_(aircraft_ids),
137 Snag.is_grounding.is_(False),
138 Snag.resolved_at.is_(None),
139 )
140 .order_by(Snag.reported_at.asc())
141 .all()
142 )
143 if aircraft_ids
144 else []
145 )
146 open_snag_rows = [(s, ac_by_id[s.aircraft_id]) for s in open_snags]
148 aircraft_status = compute_aircraft_statuses(aircraft, triggers, hobbs_by_id)
150 # Chronological view: single list sorted by due/reported date asc.
151 # Hours-based triggers have no reliable date → sorted after all dated items.
152 # Tuple structure: (sort_date, kind_order, label, obj, ac, extra)
153 # kind_order: grounding=0, snag=1, maintenance=2 (tiebreak within same date)
154 _far_dt = _datetime(_far_future.year, _far_future.month, _far_future.day)
155 chron_items = []
156 for s, ac in grounding_snag_rows:
157 dt = _datetime.combine(
158 s.reported_at.date() if hasattr(s.reported_at, "date") else s.reported_at,
159 _datetime.min.time(),
160 )
161 chron_items.append(("grounding", dt, s, ac, None))
162 for s, ac in open_snag_rows:
163 dt = _datetime.combine(
164 s.reported_at.date() if hasattr(s.reported_at, "date") else s.reported_at,
165 _datetime.min.time(),
166 )
167 chron_items.append(("snag", dt, s, ac, None))
168 for t, status, ac in trigger_rows:
169 if status in ("overdue", "due_soon"):
170 if t.trigger_type == TriggerType.CALENDAR and t.due_date:
171 dt = _datetime(t.due_date.year, t.due_date.month, t.due_date.day)
172 else:
173 dt = _far_dt # hours-based: push to end
174 chron_items.append(("maintenance", dt, t, ac, status))
176 _kind_order = {"grounding": 0, "snag": 1, "maintenance": 2}
177 chron_items.sort(key=lambda x: (x[1], _kind_order[x[0]]))
179 view = request.args.get("view", "by-type")
181 return render_template(
182 "maintenance/fleet.html",
183 aircraft=aircraft,
184 aircraft_status=aircraft_status,
185 trigger_rows=trigger_rows,
186 grounding_snag_rows=grounding_snag_rows,
187 open_snag_rows=open_snag_rows,
188 chron_items=chron_items,
189 hobbs_by_id=hobbs_by_id,
190 view=view,
191 )
194# ── Trigger list ──────────────────────────────────────────────────────────────
197@maintenance_bp.route("/aircraft/<int:aircraft_id>/maintenance")
198@login_required
199def list_triggers(aircraft_id: int) -> ResponseReturnValue:
200 ac = _get_aircraft_or_404(aircraft_id)
201 current_hobbs = ac.total_engine_hours
202 all_triggers = (
203 MaintenanceTrigger.query.filter_by(aircraft_id=ac.id)
204 .order_by(MaintenanceTrigger.name)
205 .all()
206 )
207 tid = _tenant_id()
208 uid = session["user_id"]
209 maint_view = AuthorizationService.maintenance_view_level(uid, aircraft_id, tid)
210 # Limited view: show only overdue and due-soon items
211 if maint_view == "limited":
212 triggers = [
213 t
214 for t in all_triggers
215 if t.status(current_hobbs) in ("overdue", "due_soon")
216 ]
217 else:
218 triggers = all_triggers
219 trigger_rows = [(t, t.status(current_hobbs)) for t in triggers]
220 return render_template(
221 "maintenance/list.html",
222 aircraft=ac,
223 trigger_rows=trigger_rows,
224 current_hobbs=current_hobbs,
225 maint_view=maint_view,
226 )
229# ── Add trigger ───────────────────────────────────────────────────────────────
232@maintenance_bp.route(
233 "/aircraft/<int:aircraft_id>/maintenance/new", methods=["GET", "POST"]
234)
235@login_required
236@require_role(*_MAINT_ROLES)
237def new_trigger(aircraft_id: int) -> ResponseReturnValue:
238 ac = _get_aircraft_or_404(aircraft_id)
239 if request.method == "POST":
240 return _save_trigger(ac, None)
241 return render_template(
242 "maintenance/trigger_form.html",
243 aircraft=ac,
244 trigger=None,
245 trigger_types=TriggerType,
246 )
249# ── Edit trigger ──────────────────────────────────────────────────────────────
252@maintenance_bp.route(
253 "/aircraft/<int:aircraft_id>/maintenance/<int:trigger_id>/edit",
254 methods=["GET", "POST"],
255)
256@login_required
257@require_role(*_MAINT_ROLES)
258def edit_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
259 ac = _get_aircraft_or_404(aircraft_id)
260 t = _get_trigger_or_404(ac, trigger_id)
261 if request.method == "POST":
262 return _save_trigger(ac, t)
263 return render_template(
264 "maintenance/trigger_form.html",
265 aircraft=ac,
266 trigger=t,
267 trigger_types=TriggerType,
268 )
271def _save_trigger(ac: Aircraft, t: MaintenanceTrigger | None) -> ResponseReturnValue:
272 name = request.form.get("name", "").strip()
273 trigger_type = request.form.get("trigger_type", "").strip()
274 due_date_raw = request.form.get("due_date", "").strip()
275 interval_days_raw = request.form.get("interval_days", "").strip()
276 due_engine_hours_raw = request.form.get("due_engine_hours", "").strip()
277 interval_hours_raw = request.form.get("interval_hours", "").strip()
278 notes = request.form.get("notes", "").strip() or None
280 errors = []
281 if not name:
282 errors.append(_("Name is required."))
283 if trigger_type not in TriggerType.ALL:
284 errors.append(_("Trigger type must be 'calendar' or 'hours'."))
286 due_date = interval_days = due_engine_hours = interval_hours = None
288 if trigger_type == TriggerType.CALENDAR:
289 if not due_date_raw:
290 errors.append(_("Due date is required for calendar triggers."))
291 else:
292 try:
293 due_date = _date.fromisoformat(due_date_raw)
294 except ValueError:
295 errors.append(_("Due date must be a valid date (YYYY-MM-DD)."))
296 if interval_days_raw:
297 try:
298 interval_days = int(interval_days_raw)
299 if interval_days <= 0:
300 raise ValueError
301 except ValueError:
302 errors.append(_("Interval (days) must be a positive integer."))
304 elif trigger_type == TriggerType.HOURS:
305 if not due_engine_hours_raw:
306 errors.append(_("Due engine hours is required for hours triggers."))
307 else:
308 try:
309 due_engine_hours = float(due_engine_hours_raw)
310 if due_engine_hours < 0:
311 raise ValueError
312 except ValueError:
313 errors.append(_("Due engine hours must be a positive number."))
314 if interval_hours_raw:
315 try:
316 interval_hours = float(interval_hours_raw)
317 if interval_hours <= 0:
318 raise ValueError
319 except ValueError:
320 errors.append(_("Interval (hours) must be a positive number."))
322 if errors:
323 for msg in errors:
324 flash(msg, "danger")
325 return render_template(
326 "maintenance/trigger_form.html",
327 aircraft=ac,
328 trigger=t,
329 trigger_types=TriggerType,
330 )
332 if t is None:
333 t = MaintenanceTrigger(aircraft_id=ac.id)
334 db.session.add(t)
336 t.name = name
337 t.trigger_type = trigger_type
338 t.due_date = due_date
339 t.interval_days = interval_days
340 t.due_engine_hours = due_engine_hours
341 t.interval_hours = interval_hours
342 t.notes = notes
343 db.session.commit()
345 flash(_("Maintenance item '%(name)s' saved.", name=t.name), "success")
346 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
349# ── Delete trigger ────────────────────────────────────────────────────────────
352@maintenance_bp.route(
353 "/aircraft/<int:aircraft_id>/maintenance/<int:trigger_id>/delete", methods=["POST"]
354)
355@login_required
356@require_role(*_MAINT_ROLES)
357def delete_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
358 ac = _get_aircraft_or_404(aircraft_id)
359 t = _get_trigger_or_404(ac, trigger_id)
360 name = t.name
361 db.session.delete(t)
362 db.session.commit()
363 flash(_("'%(name)s' deleted.", name=name), "success")
364 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
367# ── Mark as serviced ──────────────────────────────────────────────────────────
370@maintenance_bp.route(
371 "/aircraft/<int:aircraft_id>/maintenance/<int:trigger_id>/service",
372 methods=["GET", "POST"],
373)
374@login_required
375@require_role(*_MAINT_ROLES)
376def service_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
377 ac = _get_aircraft_or_404(aircraft_id)
378 t = _get_trigger_or_404(ac, trigger_id)
380 if request.method == "POST":
381 performed_raw = request.form.get("performed_at", "").strip()
382 hobbs_raw = request.form.get("hobbs_at_service", "").strip()
383 notes = request.form.get("notes", "").strip() or None
385 errors = []
386 performed_at = None
387 if not performed_raw:
388 errors.append(_("Service date is required."))
389 else:
390 try:
391 performed_at = _date.fromisoformat(performed_raw)
392 except ValueError:
393 errors.append(_("Service date must be a valid date (YYYY-MM-DD)."))
395 hobbs_at_service = None
396 if t.trigger_type == TriggerType.HOURS:
397 if not hobbs_raw:
398 errors.append(
399 _("Hobbs at service is required for hours-based triggers.")
400 )
401 else:
402 try:
403 hobbs_at_service = float(hobbs_raw)
404 if hobbs_at_service < 0:
405 raise ValueError
406 except ValueError:
407 errors.append(_("Hobbs at service must be a positive number."))
408 elif hobbs_raw:
409 try:
410 hobbs_at_service = float(hobbs_raw)
411 except ValueError:
412 hobbs_at_service = None
414 if errors:
415 for msg in errors:
416 flash(msg, "danger")
417 return render_template(
418 "maintenance/service_form.html",
419 aircraft=ac,
420 trigger=t,
421 current_hobbs=ac.total_engine_hours,
422 today=_date.today().isoformat(),
423 )
425 record = MaintenanceRecord(
426 trigger_id=t.id,
427 performed_at=performed_at,
428 hobbs_at_service=hobbs_at_service,
429 notes=notes,
430 )
431 db.session.add(record)
433 # Advance the trigger's due value if an interval is configured
434 if t.trigger_type == TriggerType.CALENDAR and t.interval_days and performed_at:
435 t.due_date = performed_at + timedelta(days=t.interval_days)
436 elif (
437 t.trigger_type == TriggerType.HOURS
438 and t.interval_hours
439 and hobbs_at_service is not None
440 ):
441 t.due_engine_hours = hobbs_at_service + float(t.interval_hours)
443 db.session.commit()
444 activity(
445 "maintenance.serviced",
446 trigger_id=t.id,
447 aircraft_id=aircraft_id,
448 trigger_name=t.name,
449 record_id=record.id,
450 )
451 flash(_("'%(name)s' marked as serviced.", name=t.name), "success")
452 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
454 return render_template(
455 "maintenance/service_form.html",
456 aircraft=ac,
457 trigger=t,
458 current_hobbs=ac.total_engine_hours,
459 today=_date.today().isoformat(),
460 )