generated from bisco/codex-bootstrap
153 lines
4.6 KiB
Python
153 lines
4.6 KiB
Python
from dataclasses import dataclass
|
|
|
|
from django.db import IntegrityError, transaction
|
|
|
|
from bookings.models import Reservation, ReservationToken
|
|
|
|
from .models import CheckIn
|
|
|
|
|
|
class CheckInServiceError(Exception):
|
|
"""Base class for check-in service domain errors."""
|
|
|
|
|
|
class InvalidToken(CheckInServiceError):
|
|
pass
|
|
|
|
|
|
class ReservationNotConfirmed(CheckInServiceError):
|
|
pass
|
|
|
|
|
|
class AlreadyCheckedIn(CheckInServiceError):
|
|
pass
|
|
|
|
|
|
class MissingStaffUser(CheckInServiceError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CheckInPreview:
|
|
reservation_id: int
|
|
performance_id: int
|
|
show_title: str
|
|
venue_name: str
|
|
starts_at: object
|
|
party_size: int
|
|
checked_in: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CheckInResult:
|
|
check_in: CheckIn
|
|
preview: CheckInPreview
|
|
|
|
|
|
def preview_check_in_token(raw_token, *, staff_user):
|
|
_validate_staff_user(staff_user)
|
|
reservation = _get_reservation_for_check_in_token(raw_token)
|
|
_validate_reservation_for_check_in(reservation)
|
|
return _build_preview(reservation)
|
|
|
|
|
|
def confirm_check_in_from_token(raw_token, *, staff_user, source=CheckIn.Source.QR_SCAN):
|
|
_validate_staff_user(staff_user)
|
|
|
|
with transaction.atomic():
|
|
reservation = _get_reservation_for_check_in_token(raw_token, lock_token=True)
|
|
reservation = (
|
|
Reservation.objects.select_for_update()
|
|
.select_related("performance__show", "performance__venue")
|
|
.get(pk=reservation.pk)
|
|
)
|
|
_validate_reservation_for_check_in(reservation)
|
|
|
|
try:
|
|
check_in = CheckIn.objects.create(
|
|
reservation=reservation,
|
|
checked_in_by=staff_user,
|
|
source=source,
|
|
)
|
|
except IntegrityError as exc:
|
|
raise AlreadyCheckedIn("Reservation has already been checked in.") from exc
|
|
|
|
return CheckInResult(check_in=check_in, preview=_build_preview(reservation))
|
|
|
|
|
|
def confirm_check_in_for_reservation(*, reservation_id, staff_user, source=CheckIn.Source.MANUAL):
|
|
_validate_staff_user(staff_user)
|
|
|
|
with transaction.atomic():
|
|
reservation = (
|
|
Reservation.objects.select_for_update()
|
|
.select_related("performance__show", "performance__venue")
|
|
.get(pk=reservation_id)
|
|
)
|
|
_validate_reservation_for_check_in(reservation)
|
|
|
|
try:
|
|
check_in = CheckIn.objects.create(
|
|
reservation=reservation,
|
|
checked_in_by=staff_user,
|
|
source=source,
|
|
)
|
|
except IntegrityError as exc:
|
|
raise AlreadyCheckedIn("Reservation has already been checked in.") from exc
|
|
|
|
return CheckInResult(check_in=check_in, preview=_build_preview(reservation))
|
|
|
|
|
|
def _validate_staff_user(staff_user):
|
|
if staff_user is None:
|
|
raise MissingStaffUser("A staff user is required for check-in.")
|
|
if not getattr(staff_user, "is_authenticated", False):
|
|
raise MissingStaffUser("An authenticated staff user is required for check-in.")
|
|
if not getattr(staff_user, "is_staff", False):
|
|
raise MissingStaffUser("A staff user is required for check-in.")
|
|
|
|
|
|
def _get_reservation_for_check_in_token(raw_token, *, lock_token=False):
|
|
if not isinstance(raw_token, str) or not raw_token:
|
|
raise InvalidToken("Check-in token is invalid.")
|
|
|
|
queryset = ReservationToken.objects.select_related(
|
|
"reservation__performance__show",
|
|
"reservation__performance__venue",
|
|
)
|
|
if lock_token:
|
|
queryset = queryset.select_for_update()
|
|
|
|
try:
|
|
token = queryset.get(
|
|
token_hash=ReservationToken.hash_token(raw_token),
|
|
purpose=ReservationToken.Purpose.CHECK_IN,
|
|
)
|
|
except ReservationToken.DoesNotExist as exc:
|
|
raise InvalidToken("Check-in token is invalid.") from exc
|
|
|
|
if token.used_at is not None or token.is_expired:
|
|
raise InvalidToken("Check-in token is invalid.")
|
|
|
|
return token.reservation
|
|
|
|
|
|
def _validate_reservation_for_check_in(reservation):
|
|
if reservation.status != Reservation.Status.CONFIRMED:
|
|
raise ReservationNotConfirmed("Reservation must be confirmed before check-in.")
|
|
if hasattr(reservation, "check_in"):
|
|
raise AlreadyCheckedIn("Reservation has already been checked in.")
|
|
|
|
|
|
def _build_preview(reservation):
|
|
performance = reservation.performance
|
|
return CheckInPreview(
|
|
reservation_id=reservation.id,
|
|
performance_id=performance.id,
|
|
show_title=performance.show.title,
|
|
venue_name=performance.venue.name,
|
|
starts_at=performance.starts_at,
|
|
party_size=reservation.party_size,
|
|
checked_in=hasattr(reservation, "check_in"),
|
|
)
|