Concert Ticket Booking

  1  from __future__ import annotations
  2  from enum import Enum
  3  from dataclasses import dataclass, field
  4  from datetime import datetime, timedelta
  5  from threading import Lock
  6  from typing import Protocol
  7  import uuid
  8  
  9  
 10  class SeatStatus(Enum):
 11      AVAILABLE = "available"
 12      LOCKED = "locked"
 13      BOOKED = "booked"
 14  
 15  SEAT_TRANSITIONS: dict[SeatStatus, set[SeatStatus]] = {
 16      SeatStatus.AVAILABLE: {SeatStatus.LOCKED},
 17      SeatStatus.LOCKED: {SeatStatus.BOOKED, SeatStatus.AVAILABLE},
 18      SeatStatus.BOOKED: {SeatStatus.AVAILABLE},
 19  }
 20  
 21  
 22  class Seat:
 23      def __init__(self, seat_id: str, section: str, row: int, number: int, price: float):
 24          self.seat_id = seat_id
 25          self.section = section
 26          self.row = row
 27          self.number = number
 28          self.price = price
 29          self.status = SeatStatus.AVAILABLE
 30  
 31      def _transition(self, target: SeatStatus) -> None:
 32          allowed = SEAT_TRANSITIONS.get(self.status, set())
 33          if target not in allowed:
 34              raise RuntimeError(
 35                  f"Seat {self.seat_id}: cannot go from {self.status.value} to {target.value}"
 36              )
 37          self.status = target
 38  
 39      def lock(self) -> None:
 40          self._transition(SeatStatus.LOCKED)
 41  
 42      def book(self) -> None:
 43          self._transition(SeatStatus.BOOKED)
 44  
 45      def release(self) -> None:
 46          self._transition(SeatStatus.AVAILABLE)
 47  
 48      def __repr__(self) -> str:
 49          return f"Seat({self.seat_id}, ${self.price:.0f}, {self.status.value})"
 50  
 51  
 52  @dataclass
 53  class Section:
 54      name: str
 55      rows: int
 56      seats_per_row: int
 57      base_price: float
 58  
 59  
 60  @dataclass
 61  class Venue:
 62      venue_id: str
 63      name: str
 64      sections: list[Section]
 65  
 66      @property
 67      def total_capacity(self) -> int:
 68          return sum(s.rows * s.seats_per_row for s in self.sections)
 69  
 70  
 71  class Event:
 72      def __init__(self, event_id: str, name: str, venue: Venue, event_date: datetime):
 73          self.event_id = event_id
 74          self.name = name
 75          self.venue = venue
 76          self.date = event_date
 77          self.seats: dict[str, Seat] = {}
 78          self._build_seats()
 79  
 80      def _build_seats(self) -> None:
 81          for section in self.venue.sections:
 82              for row in range(1, section.rows + 1):
 83                  for num in range(1, section.seats_per_row + 1):
 84                      seat_id = f"{section.name}-R{row}-S{num}"
 85                      self.seats[seat_id] = Seat(
 86                          seat_id, section.name, row, num, section.base_price
 87                      )
 88  
 89      def available_seats(self, section_name: str | None = None) -> list[Seat]:
 90          seats = list(self.seats.values())
 91          if section_name:
 92              seats = [s for s in seats if s.section == section_name]
 93          return [s for s in seats if s.status == SeatStatus.AVAILABLE]
 94  
 95      def get_seat(self, seat_id: str) -> Seat:
 96          seat = self.seats.get(seat_id)
 97          if seat is None:
 98              raise ValueError(f"Seat {seat_id} does not exist")
 99          return seat
100  
101  
102  @dataclass
103  class SeatLock:
104      lock_id: str
105      user_id: str
106      seats: list[Seat]
107      expires_at: datetime
108  
109      def is_expired(self) -> bool:
110          return datetime.now() > self.expires_at
111  
112  
113  @dataclass
114  class Booking:
115      booking_id: str
116      user_id: str
117      event: Event
118      seats: list[Seat]
119      total_amount: float
120      payment_receipt: str
121  
122  
123  class PaymentGateway(Protocol):
124      def charge(self, user_id: str, amount: float) -> str: ...
125  
126  class CreditCardGateway:
127      def charge(self, user_id: str, amount: float) -> str:
128          receipt = f"CC-{uuid.uuid4().hex[:8].upper()}"
129          print(f"  [CreditCard] Charged ${amount:.2f} to {user_id} -> {receipt}")
130          return receipt
131  
132  class UPIGateway:
133      def charge(self, user_id: str, amount: float) -> str:
134          receipt = f"UPI-{uuid.uuid4().hex[:8].upper()}"
135          print(f"  [UPI] Charged ${amount:.2f} to {user_id} -> {receipt}")
136          return receipt
137  
138  
139  class WaitlistObserver(Protocol):
140      def on_seats_released(self, event_id: str, count: int) -> None: ...
141  
142  class WaitlistNotifier:
143      def on_seats_released(self, event_id: str, count: int) -> None:
144          print(f"  [Waitlist] {count} seat(s) released for event {event_id}")
145  
146  
147  class BookingSystem:
148      LOCK_TTL = timedelta(minutes=5)
149  
150      def __init__(self, payment: PaymentGateway | None = None):
151          self._events: dict[str, Event] = {}
152          self._bookings: dict[str, Booking] = {}
153          self._locks: dict[str, SeatLock] = {}
154          self._payment = payment or CreditCardGateway()
155          self._waitlist_observers: list[WaitlistObserver] = []
156          self._mutex = Lock()
157  
158      def add_observer(self, observer: WaitlistObserver) -> None:
159          self._waitlist_observers.append(observer)
160  
161      def create_event(self, name: str, venue: Venue, event_date: datetime) -> Event:
162          event = Event(str(uuid.uuid4())[:8], name, venue, event_date)
163          self._events[event.event_id] = event
164          return event
165  
166      def lock_seats(self, event_id: str, user_id: str,
167                     seat_ids: list[str]) -> SeatLock:
168          with self._mutex:
169              self._cleanup_expired()
170              event = self._get_event(event_id)
171              seats = [event.get_seat(sid) for sid in seat_ids]
172              for seat in seats:
173                  seat.lock()
174              seat_lock = SeatLock(
175                  lock_id=str(uuid.uuid4())[:8],
176                  user_id=user_id,
177                  seats=seats,
178                  expires_at=datetime.now() + self.LOCK_TTL,
179              )
180              self._locks[seat_lock.lock_id] = seat_lock
181              return seat_lock
182  
183      def confirm_booking(self, lock_id: str) -> Booking:
184          with self._mutex:
185              seat_lock = self._locks.pop(lock_id, None)
186              if seat_lock is None:
187                  raise ValueError(f"Lock {lock_id} not found or already expired")
188              if seat_lock.is_expired():
189                  for seat in seat_lock.seats:
190                      if seat.status == SeatStatus.LOCKED:
191                          seat.release()
192                  raise RuntimeError("Lock expired. Seats have been released.")
193  
194              total = sum(s.price for s in seat_lock.seats)
195              receipt = self._payment.charge(seat_lock.user_id, total)
196  
197              for seat in seat_lock.seats:
198                  seat.book()
199  
200              event = self._find_event_for_seat(seat_lock.seats[0])
201              booking = Booking(
202                  booking_id=str(uuid.uuid4())[:8],
203                  user_id=seat_lock.user_id,
204                  event=event,
205                  seats=seat_lock.seats,
206                  total_amount=total,
207                  payment_receipt=receipt,
208              )
209              self._bookings[booking.booking_id] = booking
210              return booking
211  
212      def cancel_booking(self, booking_id: str) -> None:
213          with self._mutex:
214              booking = self._bookings.pop(booking_id, None)
215              if booking is None:
216                  raise ValueError(f"Booking {booking_id} not found")
217              released = 0
218              for seat in booking.seats:
219                  seat.release()
220                  released += 1
221              for obs in self._waitlist_observers:
222                  obs.on_seats_released(booking.event.event_id, released)
223  
224      def cancel_lock(self, lock_id: str) -> None:
225          with self._mutex:
226              seat_lock = self._locks.pop(lock_id, None)
227              if seat_lock is None:
228                  return
229              for seat in seat_lock.seats:
230                  if seat.status == SeatStatus.LOCKED:
231                      seat.release()
232  
233      def _cleanup_expired(self) -> None:
234          expired = [lid for lid, lk in self._locks.items() if lk.is_expired()]
235          for lid in expired:
236              lk = self._locks.pop(lid)
237              for seat in lk.seats:
238                  if seat.status == SeatStatus.LOCKED:
239                      seat.release()
240  
241      def _get_event(self, event_id: str) -> Event:
242          event = self._events.get(event_id)
243          if event is None:
244              raise ValueError(f"Event {event_id} not found")
245          return event
246  
247      def _find_event_for_seat(self, seat: Seat) -> Event:
248          for event in self._events.values():
249              if seat.seat_id in event.seats:
250                  return event
251          raise ValueError("No event found for seat")
252  
253  
254  if __name__ == "__main__":
255      venue = Venue("V1", "City Arena", [
256          Section("VIP", rows=3, seats_per_row=10, base_price=150.0),
257          Section("Floor", rows=10, seats_per_row=20, base_price=100.0),
258          Section("Balcony", rows=8, seats_per_row=25, base_price=50.0),
259      ])
260  
261      system = BookingSystem(CreditCardGateway())
262      system.add_observer(WaitlistNotifier())
263  
264      print("=== Concert Ticket Booking ===\n")
265  
266      event = system.create_event("Rock Festival 2025", venue, datetime(2025, 8, 15, 20, 0))
267      print(f"Event: {event.name}")
268      print(f"Venue: {venue.name} (capacity: {venue.total_capacity})")
269      print(f"VIP available: {len(event.available_seats('VIP'))}")
270      print(f"Floor available: {len(event.available_seats('Floor'))}")
271      print(f"Balcony available: {len(event.available_seats('Balcony'))}")
272  
273      # User A locks 2 VIP seats
274      print("\nUser A locks VIP seats:")
275      lock_a = system.lock_seats(event.event_id, "user-A", ["VIP-R1-S1", "VIP-R1-S2"])
276      for s in lock_a.seats:
277          print(f"  {s}")
278      print(f"  VIP available after lock: {len(event.available_seats('VIP'))}")
279  
280      # User A confirms and pays
281      print("\nUser A confirms booking:")
282      booking_a = system.confirm_booking(lock_a.lock_id)
283      print(f"  Booking {booking_a.booking_id} confirmed")
284      print(f"  Total: ${booking_a.total_amount:.2f}")
285      print(f"  Receipt: {booking_a.payment_receipt}")
286  
287      # User B locks Floor seats then cancels
288      print("\nUser B locks Floor seats:")
289      lock_b = system.lock_seats(event.event_id, "user-B",
290                                  ["Floor-R1-S1", "Floor-R1-S2", "Floor-R1-S3"])
291      print(f"  Locked {len(lock_b.seats)} seats")
292  
293      print("\nUser B cancels the lock:")
294      system.cancel_lock(lock_b.lock_id)
295      print(f"  Floor available after cancel: {len(event.available_seats('Floor'))}")
296  
297      # User A cancels their booking
298      print("\nUser A cancels their booking:")
299      system.cancel_booking(booking_a.booking_id)
300      print(f"  VIP available after cancellation: {len(event.available_seats('VIP'))}")