refactor(api): give check-in a service class and typed responses

buzz/api/checkin/__init__.py

@@ -1,112 +1,14 @@
 import frappe
-from frappe import _
-from frappe.utils import format_date, format_time
 
-
-@frappe.whitelist()
-def validate_ticket_for_checkin(ticket_id: str) -> dict:
-	frappe.only_for("Frontdesk Manager", True)
-	if not frappe.db.exists("Event Ticket", ticket_id):
-		frappe.throw(_("Ticket not found"))
-
-	ticket_doc = frappe.get_cached_doc("Event Ticket", ticket_id)
-
-	if ticket_doc.docstatus == 2:
-		frappe.throw(_("This ticket has been cancelled and cannot be checked in"))
-
-	event_doc = frappe.get_cached_doc("Buzz Event", ticket_doc.event)
-	ticket_type_doc = (
-		frappe.get_cached_doc("Event Ticket Type", ticket_doc.ticket_type) if ticket_doc.ticket_type else None
-	)
-
-	checkin_date = frappe.utils.today()
-	existing_checkin = frappe.db.exists("Event Check In", {"ticket": ticket_id, "date": checkin_date})
-
-	if existing_checkin:
-		checkin_doc = frappe.get_doc("Event Check In", existing_checkin)
-		formatted_checkin_time = (
-			format_date(checkin_doc.creation) + " at " + format_time(checkin_doc.creation)
-		)
-
-		frappe.throw(_("This ticket was already checked in today ({0}).").format(formatted_checkin_time))
-
-	add_ons = frappe.db.get_all(
-		"Ticket Add-on Value",
-		filters={"parent": ticket_id},
-		fields=[
-			"add_on",
-			"add_on.title as add_on_title",
-			"add_on.user_selects_option as add_on_selects_option",
-			"value",
-			"price",
-			"currency",
-		],
-	)
-
-	return {
-		"message": _("Valid ticket ready for check-in"),
-		"ticket": {
-			"id": ticket_doc.name,
-			"attendee_name": ticket_doc.attendee_name,
-			"attendee_email": ticket_doc.attendee_email,
-			"event_title": event_doc.title,
-			"ticket_type": (ticket_type_doc.title if ticket_type_doc else ticket_doc.ticket_type),
-			"venue": event_doc.venue,
-			"start_date": event_doc.start_date,
-			"start_time": event_doc.start_time,
-			"end_date": event_doc.end_date,
-			"end_time": event_doc.end_time,
-			"is_checked_in": False,
-			"check_in_time": None,
-			"booking_id": ticket_doc.booking,
-			"add_ons": add_ons,
-		},
-		"payment_details": get_payment_details_for_ticket(ticket_id),
-	}
-
-
-def get_payment_details_for_ticket(ticket_id: str) -> dict | None:
-	booking_id = frappe.get_cached_value("Event Ticket", ticket_id, "booking")
-	if not booking_id:
-		return None
-
-	payments = frappe.db.get_all(
-		"Event Payment",
-		filters={
-			"reference_doctype": "Event Booking",
-			"reference_docname": booking_id,
-			"payment_received": 1,
-		},
-		fields=["name", "amount", "currency"],
-		limit=1,
-	)
-
-	if payments:
-		return payments[0]
+from buzz.api.checkin.schemas import CheckinResponse
+from buzz.api.checkin.services import CheckinService
 
 
 @frappe.whitelist()
-def checkin_ticket(ticket_id: str) -> dict:
-	frappe.only_for("Frontdesk Manager", True)
+def validate_ticket_for_checkin(ticket_id: str) -> CheckinResponse:
+	return CheckinService(ticket_id).validate()
 
-	checkin_date = frappe.utils.today()
-	validation_result = validate_ticket_for_checkin(ticket_id)
 
-	checkin_doc = frappe.new_doc("Event Check In")
-	checkin_doc.ticket = ticket_id
-	checkin_doc.date = checkin_date
-	checkin_doc.insert(ignore_permissions=True)
-	checkin_doc.submit()
-
-	return {
-		"message": _("Successfully checked in {attendee_name} for {checkin_date}").format(
-			attendee_name=validation_result["ticket"]["attendee_name"],
-			checkin_date=frappe.format(checkin_date, {"fieldtype": "Date"}),
-		),
-		"ticket": {
-			**validation_result["ticket"],
-			"is_checked_in": True,
-			"check_in_time": checkin_doc.creation,
-			"check_in_date": checkin_date,
-		},
-	}
+@frappe.whitelist()
+def checkin_ticket(ticket_id: str) -> CheckinResponse:
+	return CheckinService(ticket_id).checkin()

buzz/api/checkin/exceptions.py

@@ -0,0 +1,18 @@
+from frappe import _lt
+
+from buzz.api.exceptions import Conflict, ResourceNotFound
+
+
+class TicketNotFound(ResourceNotFound):
+	title = _lt("Ticket Not Found")
+	message = _lt("No ticket matches this code.")
+
+
+class TicketCancelled(Conflict):
+	title = _lt("Ticket Cancelled")
+	message = _lt("This ticket has been cancelled and cannot be checked in.")
+
+
+class AlreadyCheckedIn(Conflict):
+	title = _lt("Already Checked In")
+	message = _lt("This ticket was already checked in today, at {checked_in_at}.")

buzz/api/checkin/schemas.py

@@ -0,0 +1,42 @@
+from datetime import date, datetime, timedelta
+
+from buzz.api.schemas import APIResponse
+
+
+class PaymentDetails(APIResponse):
+	name: str
+	amount: float
+	currency: str | None
+
+
+class TicketAddOn(APIResponse):
+	add_on: str
+	add_on_title: str | None
+	add_on_selects_option: int | None
+	value: str | None
+	price: float | None
+	currency: str | None
+
+
+class TicketCheckinDetails(APIResponse):
+	id: str
+	attendee_name: str | None
+	attendee_email: str | None
+	event_title: str
+	ticket_type: str | None
+	venue: str | None
+	start_date: date | None
+	start_time: timedelta | None
+	end_date: date | None
+	end_time: timedelta | None
+	is_checked_in: bool
+	check_in_time: datetime | None
+	booking_id: str | None
+	add_ons: list[TicketAddOn]
+	check_in_date: str | None = None
+
+
+class CheckinResponse(APIResponse):
+	message: str
+	ticket: TicketCheckinDetails
+	payment_details: PaymentDetails | None = None

buzz/api/checkin/services.py

@@ -0,0 +1,137 @@
+from typing import TYPE_CHECKING
+
+import frappe
+from frappe import _
+from frappe.utils import format_date, format_time, today
+
+from buzz.api.checkin.exceptions import AlreadyCheckedIn, TicketCancelled, TicketNotFound
+from buzz.api.checkin.schemas import (
+	CheckinResponse,
+	PaymentDetails,
+	TicketAddOn,
+	TicketCheckinDetails,
+)
+
+if TYPE_CHECKING:
+	from buzz.events.doctype.buzz_event.buzz_event import BuzzEvent
+	from buzz.events.doctype.event_check_in.event_check_in import EventCheckIn
+	from buzz.ticketing.doctype.event_ticket.event_ticket import EventTicket
+
+CANCELLED = 2
+
+
+class CheckinService:
+	"""Front-desk check-in for a single ticket. Restricted to Frontdesk Manager."""
+
+	def __init__(self, ticket_id: str):
+		frappe.only_for("Frontdesk Manager", True)
+		if not frappe.db.exists("Event Ticket", ticket_id):
+			TicketNotFound.throw()
+
+		self.ticket_id = ticket_id
+		self.date = today()
+
+	@property
+	def ticket(self) -> "EventTicket":
+		return frappe.get_cached_doc("Event Ticket", self.ticket_id)
+
+	@property
+	def event(self) -> "BuzzEvent":
+		return frappe.get_cached_doc("Buzz Event", self.ticket.event)
+
+	def validate(self) -> CheckinResponse:
+		self.ensure_checkin_allowed()
+		return CheckinResponse(
+			message=_("Valid ticket ready for check-in"),
+			ticket=self.ticket_details(),
+			payment_details=self.payment_details(),
+		)
+
+	def checkin(self) -> CheckinResponse:
+		self.ensure_checkin_allowed()
+		checkin_doc = self.record_checkin()
+
+		return CheckinResponse(
+			message=_("Successfully checked in {attendee_name} for {checkin_date}").format(
+				attendee_name=self.ticket.attendee_name,
+				checkin_date=frappe.format(self.date, {"fieldtype": "Date"}),
+			),
+			ticket=self.ticket_details(
+				is_checked_in=True,
+				check_in_time=checkin_doc.creation,
+				check_in_date=self.date,
+			),
+		)
+
+	def ensure_checkin_allowed(self) -> None:
+		if self.ticket.docstatus == CANCELLED:
+			TicketCancelled.throw()
+
+		existing = frappe.db.exists("Event Check In", {"ticket": self.ticket_id, "date": self.date})
+		if existing:
+			creation = frappe.db.get_value("Event Check In", existing, "creation")
+			AlreadyCheckedIn.throw(checked_in_at=f"{format_date(creation)} at {format_time(creation)}")
+
+	def record_checkin(self) -> "EventCheckIn":
+		checkin_doc = frappe.new_doc("Event Check In")
+		checkin_doc.ticket = self.ticket_id
+		checkin_doc.date = self.date
+		checkin_doc.insert(ignore_permissions=True)
+		checkin_doc.submit()
+		return checkin_doc
+
+	def ticket_details(self, **checkin_state) -> TicketCheckinDetails:
+		ticket_type = (
+			frappe.get_cached_value("Event Ticket Type", self.ticket.ticket_type, "title")
+			if self.ticket.ticket_type
+			else None
+		)
+
+		return TicketCheckinDetails(
+			id=self.ticket.name,
+			attendee_name=self.ticket.attendee_name,
+			attendee_email=self.ticket.attendee_email,
+			event_title=self.event.title,
+			ticket_type=ticket_type or self.ticket.ticket_type,
+			venue=self.event.venue,
+			start_date=self.event.start_date,
+			start_time=self.event.start_time,
+			end_date=self.event.end_date,
+			end_time=self.event.end_time,
+			is_checked_in=checkin_state.get("is_checked_in", False),
+			check_in_time=checkin_state.get("check_in_time"),
+			check_in_date=checkin_state.get("check_in_date"),
+			booking_id=self.ticket.booking,
+			add_ons=self.add_ons(),
+		)
+
+	def add_ons(self) -> list[TicketAddOn]:
+		rows = frappe.db.get_all(
+			"Ticket Add-on Value",
+			filters={"parent": self.ticket_id},
+			fields=[
+				"add_on",
+				"add_on.title as add_on_title",
+				"add_on.user_selects_option as add_on_selects_option",
+				"value",
+				"price",
+				"currency",
+			],
+		)
+		return [TicketAddOn(**row) for row in rows]
+
+	def payment_details(self) -> PaymentDetails | None:
+		if not self.ticket.booking:
+			return None
+
+		payments = frappe.db.get_all(
+			"Event Payment",
+			filters={
+				"reference_doctype": "Event Booking",
+				"reference_docname": self.ticket.booking,
+				"payment_received": 1,
+			},
+			fields=["name", "amount", "currency"],
+			limit=1,
+		)
+		return PaymentDetails(**payments[0]) if payments else None