Coverage for app/flights/crew_invites.py: 100%
170 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +0000
1"""Crew invites: ask another pilot of the same tenant to confirm their slot on
2a flight someone else logged (docs/backlog.md "Follow-up: let a second crew
3member claim their own slot on an existing flight").
5Flow: the logger picks a tenant pilot in the PIC / second-crew name field
6(the picker posts that pilot's user id alongside the free-text name) →
7``sync_crew_invites`` records a pending ``FlightCrewInvite`` → the invited
8pilot is e-mailed and sees it on their dashboard and pilot logbook →
9``accept_invite`` writes their user id into the slot, ``decline_invite``
10leaves the slot name-only.
12The reverse direction — a *claim* — uses the same table (``kind="claim"``):
13a pilot logging a flight that the duplicate warning finds already logged by
14someone else asks to be added to it (``claim_option`` / ``create_claim``),
15and the pilot who owns that flight's shared fields approves or declines.
17The slot's user id is only ever set on acceptance — never silently by the
18logger or the claimant — so a flight can't appear in someone's logbook (or
19count towards their totals/currency) without both sides agreeing.
20"""
22import logging
23from datetime import UTC, datetime
24from typing import Any
26from flask import url_for # pyright: ignore[reportMissingImports]
27from models import ( # pyright: ignore[reportMissingImports]
28 CrewInviteKind,
29 CrewInviteStatus,
30 CrewRole,
31 CrewSlot,
32 Flight,
33 FlightCrewInvite,
34 NotificationType,
35 User,
36 db,
37)
38from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
39from sqlalchemy.orm import ( # pyright: ignore[reportMissingImports]
40 contains_eager,
41 joinedload,
42)
43from utils import tenant_pilots # pyright: ignore[reportMissingImports]
45from flights.shared_flight import ( # pyright: ignore[reportMissingImports]
46 SECOND_CREW_FUNCTION_FIELD,
47 dispatch_to_user,
48 shared_editor_ids,
49)
51log = logging.getLogger(__name__)
53# Form field carrying the picked pilot's user id, per slot.
54SLOT_USER_ID_FIELDS: dict[str, str] = {
55 CrewSlot.PIC: "crew_user_id_0",
56 CrewSlot.SECOND: "crew_user_id_1",
57}
58# Free-text name field of the same slot.
59SLOT_NAME_FIELDS: dict[str, str] = {
60 CrewSlot.PIC: "crew_name_0",
61 CrewSlot.SECOND: "crew_name_1",
62}
65def _slot_user_id(fe: Flight, slot: str) -> int | None:
66 user_id: int | None = (
67 fe.pic_user_id if slot == CrewSlot.PIC else fe.second_crew_user_id
68 )
69 return user_id
72def _invite_flight(invite: FlightCrewInvite) -> Flight:
73 fe = db.session.get(Flight, invite.flight_id)
74 assert fe is not None, "flight_id is a non-null CASCADE foreign key"
75 return fe
78def own_slot(pilot_role: str) -> str | None:
79 """The slot the logged-in user occupies for a flight form ``pilot_role``."""
80 return {"pic": CrewSlot.PIC, "dual": CrewSlot.SECOND}.get(pilot_role)
83def requested_invites(
84 form: Any, uid: int, tenant_id: int, pilot_role: str
85) -> dict[str, int | None]:
86 """Resolve the picker's hidden user-id fields into ``{slot: user_id}``.
88 A user id is only honoured when it belongs to an active pilot of
89 *tenant_id* and isn't the logged-in user; the slot the logged-in user
90 occupies themselves (per ``pilot_role``) never gets an invite. Anything
91 else resolves to ``None`` — the name stays free text.
92 """
93 labels = dict(tenant_pilots(tenant_id))
94 mine = own_slot(pilot_role)
95 result: dict[str, int | None] = {}
96 for slot, field in SLOT_USER_ID_FIELDS.items():
97 raw = (form.get(field) or "").strip()
98 user_id = int(raw) if raw.isdigit() else None
99 # The name actually submitted must still be that pilot's suggestion
100 # label — guards against a stale hidden id after the text was edited.
101 name = (form.get(SLOT_NAME_FIELDS[slot]) or "").strip().casefold()
102 if (
103 slot == mine
104 or user_id is None
105 or user_id == uid
106 or user_id not in labels
107 or labels[user_id].casefold() != name
108 ):
109 user_id = None
110 result[slot] = user_id
111 return result
114def pending_invites_for_flight(fe: Flight) -> dict[str, FlightCrewInvite]:
115 return {
116 inv.slot: inv
117 for inv in FlightCrewInvite.query.filter_by(
118 flight_id=fe.id,
119 status=CrewInviteStatus.PENDING,
120 kind=CrewInviteKind.INVITE,
121 ).all()
122 }
125def sync_crew_invites(
126 fe: Flight, inviter_id: int, requested: dict[str, int | None]
127) -> list[FlightCrewInvite]:
128 """Bring *fe*'s pending invites in line with *requested* and return the
129 newly created ones (the caller notifies them after committing).
131 - A slot that already holds a user id is left alone: an accepted invite
132 is never undone by the logger re-saving the form.
133 - A pending invite whose user is no longer requested is cancelled.
134 - A user who already declined this slot isn't asked again automatically.
135 """
136 now = datetime.now(UTC)
137 pending = pending_invites_for_flight(fe)
138 created: list[FlightCrewInvite] = []
139 for slot in CrewSlot.ALL:
140 target = requested.get(slot)
141 current = pending.get(slot)
142 if _slot_user_id(fe, slot) is not None:
143 target = None
144 if current is not None and current.invited_user_id == target:
145 continue
146 if current is not None:
147 current.status = CrewInviteStatus.CANCELLED
148 current.responded_at = now
149 if target is None or target in (fe.pic_user_id, fe.second_crew_user_id):
150 continue
151 already_declined = FlightCrewInvite.query.filter_by(
152 flight_id=fe.id,
153 slot=slot,
154 invited_user_id=target,
155 status=CrewInviteStatus.DECLINED,
156 kind=CrewInviteKind.INVITE,
157 ).first()
158 if already_declined:
159 continue
160 # A flight logged before created_by was tracked: whoever names the
161 # crew becomes its logger, so the invited pilot can't edit its shared
162 # fields once they confirm (flights/shared_flight.py).
163 if fe.created_by_user_id is None:
164 fe.created_by_user_id = inviter_id
165 invite = FlightCrewInvite(
166 flight_id=fe.id,
167 slot=slot,
168 invited_user_id=target,
169 invited_by_user_id=inviter_id,
170 )
171 db.session.add(invite)
172 created.append(invite)
173 return created
176def pending_invites_for_user(user_id: int) -> list[FlightCrewInvite]:
177 # contains_eager reuses the join already needed for ordering by flight
178 # date instead of issuing a second query per row; invited_by is loaded
179 # the same way since every row in the dashboard/logbook panel needs it.
180 invites: list[FlightCrewInvite] = (
181 FlightCrewInvite.query.join(Flight, Flight.id == FlightCrewInvite.flight_id)
182 .options(
183 contains_eager(FlightCrewInvite.flight), # type: ignore[arg-type]
184 joinedload(FlightCrewInvite.invited_by), # type: ignore[arg-type]
185 )
186 .filter(
187 FlightCrewInvite.invited_user_id == user_id,
188 FlightCrewInvite.status == CrewInviteStatus.PENDING,
189 FlightCrewInvite.kind == CrewInviteKind.INVITE,
190 )
191 .order_by(Flight.date.desc(), Flight.id.desc())
192 .all()
193 )
194 return invites
197def pending_invite_count(user_id: int) -> int:
198 count: int = FlightCrewInvite.query.filter_by(
199 invited_user_id=user_id,
200 status=CrewInviteStatus.PENDING,
201 kind=CrewInviteKind.INVITE,
202 ).count()
203 return count
206# ── Claims: asking to be added to a flight someone else logged ────────────────
209def claim_option(fe: Flight, user_id: int, pilot_role: str) -> dict[str, Any] | None:
210 """Whether *user_id*, logging what turns out to be *fe* (duplicate
211 warning) in role *pilot_role*, can ask to be added to it instead.
213 Needs a free slot matching their role and at least one pilot who owns the
214 flight's shared fields to approve. Returns the slot, the approvers' names
215 and whether they already asked, or ``None``.
216 """
217 slot = own_slot(pilot_role)
218 if slot is None or fe.slot_for(user_id) is not None:
219 return None
220 if _slot_user_id(fe, slot) is not None:
221 return None
222 approvers = sorted(shared_editor_ids(fe))
223 if not approvers:
224 return None
225 already = (
226 FlightCrewInvite.query.filter_by(
227 flight_id=fe.id,
228 invited_user_id=user_id,
229 status=CrewInviteStatus.PENDING,
230 kind=CrewInviteKind.CLAIM,
231 ).first()
232 is not None
233 )
234 names = []
235 for approver_id in approvers:
236 approver = db.session.get(User, approver_id)
237 names.append(approver.display_name if approver else "—")
238 return {
239 "slot": slot,
240 "approver_names": ", ".join(names),
241 "already_requested": already,
242 }
245def create_claim(
246 fe: Flight, user_id: int, slot: str, requested_role: str | None
247) -> FlightCrewInvite:
248 role = (
249 requested_role
250 if slot == CrewSlot.SECOND and requested_role in SECOND_CREW_ROLES
251 else None
252 )
253 claim = FlightCrewInvite(
254 flight_id=fe.id,
255 slot=slot,
256 invited_user_id=user_id,
257 invited_by_user_id=user_id,
258 kind=CrewInviteKind.CLAIM,
259 requested_role=role,
260 )
261 db.session.add(claim)
262 return claim
265SECOND_CREW_ROLES = (CrewRole.IP, CrewRole.COPILOT, CrewRole.STUDENT, CrewRole.SP)
268def pending_claims_to_review(user_id: int) -> list[FlightCrewInvite]:
269 """Pending claims on flights *user_id* owns the shared fields of."""
270 candidates: list[FlightCrewInvite] = (
271 FlightCrewInvite.query.join(Flight, Flight.id == FlightCrewInvite.flight_id)
272 .options(
273 contains_eager(FlightCrewInvite.flight), # type: ignore[arg-type]
274 joinedload(FlightCrewInvite.invited_user), # type: ignore[arg-type]
275 )
276 .filter(
277 FlightCrewInvite.status == CrewInviteStatus.PENDING,
278 FlightCrewInvite.kind == CrewInviteKind.CLAIM,
279 or_(Flight.pic_user_id == user_id, Flight.second_crew_user_id == user_id),
280 )
281 .order_by(Flight.date.desc(), Flight.id.desc())
282 .all()
283 )
284 return [c for c in candidates if user_id in shared_editor_ids(_invite_flight(c))]
287def accept_invite(invite: FlightCrewInvite, user: User) -> bool:
288 """Write *user* into the invite's slot. Returns False (and cancels the
289 invite) when the slot was meanwhile filled or the user already occupies
290 the flight's other slot."""
291 fe = _invite_flight(invite)
292 now = datetime.now(UTC)
293 if _slot_user_id(fe, invite.slot) is not None or user.id in (
294 fe.pic_user_id,
295 fe.second_crew_user_id,
296 ):
297 invite.status = CrewInviteStatus.CANCELLED
298 invite.responded_at = now
299 return False
301 if invite.slot == CrewSlot.PIC:
302 fe.pic_user_id = user.id
303 if not fe.pic_name:
304 fe.pic_name = user.display_name
305 if fe.function_pic is None:
306 fe.function_pic = fe.flight_time
307 else:
308 fe.second_crew_user_id = user.id
309 if not fe.second_crew_name:
310 fe.second_crew_name = user.display_name
311 if not fe.second_crew_role and invite.requested_role:
312 fe.second_crew_role = invite.requested_role
313 # A safety pilot has no dedicated EASA function column, so none is set.
314 field = SECOND_CREW_FUNCTION_FIELD.get(fe.second_crew_role or "")
315 if field and getattr(fe, field) is None:
316 setattr(fe, field, fe.flight_time)
318 invite.status = CrewInviteStatus.ACCEPTED
319 invite.responded_at = now
320 return True
323def decline_invite(invite: FlightCrewInvite) -> None:
324 invite.status = CrewInviteStatus.DECLINED
325 invite.responded_at = datetime.now(UTC)
328def notify_invite_answered(invite: FlightCrewInvite, accepted: bool) -> None:
329 """Tell the pilot who sent the invite that it was confirmed or declined."""
330 if invite.invited_by_user_id is None:
331 return
332 from flask_babel import lazy_gettext as _l # pyright: ignore[reportMissingImports]
334 fe = _invite_flight(invite)
335 answered_by = db.session.get(User, invite.invited_user_id)
336 name = answered_by.display_name if answered_by else "—"
337 route = f"{fe.departure_icao or '?'} → {fe.arrival_icao or '?'}"
338 if accepted:
339 title = _l("%(name)s confirmed your flight")
340 message = _l(
341 "%(name)s confirmed the flight %(route)s on %(date)s; it is now in "
342 "their pilot logbook too."
343 )
344 else:
345 title = _l("%(name)s declined your flight")
346 message = _l(
347 "%(name)s declined the flight %(route)s on %(date)s; their name stays "
348 "on your entry without a link to their logbook."
349 )
350 dispatch_to_user(
351 invite.invited_by_user_id,
352 NotificationType.CREW_INVITE_ANSWERED,
353 {
354 "subject_key": title,
355 "subject_args": {"name": name},
356 "notification_title_key": title,
357 "notification_title_args": {"name": name},
358 "notification_message_key": message,
359 "notification_message_args": {
360 "name": name,
361 "route": route,
362 "date": fe.date.isoformat(),
363 },
364 "cta_url": url_for("pilots.view_entry", entry_id=fe.id, _external=True),
365 },
366 )
369def notify_claim_created(claim: FlightCrewInvite) -> None:
370 """Ask the pilot(s) owning the flight's shared fields to approve a claim."""
371 from flask_babel import lazy_gettext as _l # pyright: ignore[reportMissingImports]
373 fe = _invite_flight(claim)
374 claimant = db.session.get(User, claim.invited_user_id)
375 name = claimant.display_name if claimant else "—"
376 route = f"{fe.departure_icao or '?'} → {fe.arrival_icao or '?'}"
377 for approver_id in sorted(shared_editor_ids(fe)):
378 dispatch_to_user(
379 approver_id,
380 NotificationType.CREW_CLAIM,
381 {
382 "subject_key": _l("%(name)s asks to be added to your flight"),
383 "subject_args": {"name": name},
384 "notification_title_key": _l(
385 "%(name)s asks to be added to your flight"
386 ),
387 "notification_title_args": {"name": name},
388 "notification_message_key": _l(
389 "%(name)s says they were on your flight %(route)s on %(date)s "
390 "and asks to add it to their pilot logbook. Approve or decline "
391 "the request."
392 ),
393 "notification_message_args": {
394 "name": name,
395 "route": route,
396 "date": fe.date.isoformat(),
397 },
398 "cta_url": url_for("pilots.logbook", _external=True) + "#crew-claims",
399 "cta_label": _l("Review in OpenHangar"),
400 },
401 )
404def notify_claim_answered(
405 claim: FlightCrewInvite, accepted: bool, actor_id: int
406) -> None:
407 """Tell the pilot who claimed a slot whether they were added."""
408 from flask_babel import lazy_gettext as _l # pyright: ignore[reportMissingImports]
410 fe = _invite_flight(claim)
411 actor = db.session.get(User, actor_id)
412 name = actor.display_name if actor else "—"
413 route = f"{fe.departure_icao or '?'} → {fe.arrival_icao or '?'}"
414 if accepted:
415 title = _l("%(name)s approved your request")
416 message = _l(
417 "%(name)s approved your request: the flight %(route)s on %(date)s is "
418 "now in your pilot logbook."
419 )
420 else:
421 title = _l("%(name)s declined your request")
422 message = _l(
423 "%(name)s declined your request to be added to the flight %(route)s "
424 "on %(date)s."
425 )
426 dispatch_to_user(
427 claim.invited_user_id,
428 NotificationType.CREW_INVITE_ANSWERED,
429 {
430 "subject_key": title,
431 "subject_args": {"name": name},
432 "notification_title_key": title,
433 "notification_title_args": {"name": name},
434 "notification_message_key": message,
435 "notification_message_args": {
436 "name": name,
437 "route": route,
438 "date": fe.date.isoformat(),
439 },
440 "cta_url": url_for("pilots.logbook", _external=True),
441 },
442 )
445def notify_invites(invites: list[FlightCrewInvite], tenant_id: int) -> None:
446 """E-mail each invited pilot (respecting their notification preferences).
447 Failures are logged, never raised — the flight is already saved."""
448 if not invites:
449 return
450 from flask_babel import lazy_gettext as _l # pyright: ignore[reportMissingImports]
451 from services.notification_service import ( # pyright: ignore[reportMissingImports]
452 dispatch,
453 )
455 for invite in invites:
456 fe = _invite_flight(invite)
457 inviter_user = (
458 db.session.get(User, invite.invited_by_user_id)
459 if invite.invited_by_user_id
460 else None
461 )
462 inviter = inviter_user.display_name if inviter_user else "—"
463 route = f"{fe.departure_icao or '?'} → {fe.arrival_icao or '?'}"
464 if invite.slot == CrewSlot.PIC:
465 slot_label = _l("PIC / Commander")
466 elif fe.second_crew_role in CrewRole.LABELS:
467 slot_label = _l(CrewRole.LABELS[fe.second_crew_role])
468 else:
469 slot_label = _l("Second crew")
470 try:
471 dispatch(
472 NotificationType.CREW_INVITE,
473 tenant_id,
474 {
475 "subject_key": _l("Confirm your flight on %(date)s"),
476 "subject_args": {"date": fe.date.isoformat()},
477 "notification_title_key": _l("%(name)s added you to a flight"),
478 "notification_title_args": {"name": inviter},
479 "notification_message_key": _l(
480 "%(name)s logged a flight with you on board. Confirm it "
481 "to add it to your pilot logbook, or decline if it isn't "
482 "yours."
483 ),
484 "notification_message_args": {"name": inviter},
485 "details": [
486 (_l("Date"), fe.date.isoformat()),
487 (_l("Route"), route),
488 (_l("Aircraft"), fe.display_registration or "—"),
489 (_l("Your role"), slot_label),
490 ],
491 "cta_url": url_for("pilots.logbook", _external=True)
492 + "#crew-invites",
493 "cta_label": _l("Review in OpenHangar"),
494 },
495 target_user_ids=[invite.invited_user_id],
496 )
497 except Exception:
498 log.exception("Failed to dispatch crew invite notification")