fix(api): add basic booking throttling

backend/azionelab/settings.py

@@ -123,4 +123,11 @@ REST_FRAMEWORK = {
     "DEFAULT_PARSER_CLASSES": [
         "rest_framework.parsers.JSONParser",
     ],
+    "DEFAULT_THROTTLE_RATES": {
+        # Small-theatre defaults: stricter on public booking flows, looser for staff operations.
+        "reservation_create": "20/hour",
+        "reservation_confirm": "60/hour",
+        "check_in_preview": "600/hour",
+        "check_in_confirm": "600/hour",
+    },
 }

backend/bookings/test_api.py

@@ -1,4 +1,5 @@
 from datetime import timedelta
+from unittest.mock import patch
 
 from django.core import mail
 from django.urls import reverse
@@ -9,6 +10,7 @@ from rest_framework.test import APITestCase
 
 from bookings.models import Reservation
 from bookings.services import generate_confirmation_token
+from bookings.views import ReservationConfirmThrottle, ReservationCreateThrottle
 from shows.models import Performance, Show, Venue
 
 
@@ -104,6 +106,32 @@ class BookingApiTests(APITestCase):
         self.assertEqual(len(callbacks), 1)
         self.assertEqual(len(mail.outbox), 0)
 
+    def test_reservation_creation_is_throttled(self):
+        with patch.dict(ReservationCreateThrottle.THROTTLE_RATES, {"reservation_create": "1/minute"}, clear=False):
+            with self.captureOnCommitCallbacks(execute=True):
+                first_response = self.client.post(
+                    reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
+                    {
+                        "name": "Maria Rossi",
+                        "email": "maria@example.com",
+                        "party_size": 1,
+                    },
+                    format="json",
+                )
+
+            second_response = self.client.post(
+                reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
+                {
+                    "name": "Maria Rossi",
+                    "email": "maria@example.com",
+                    "party_size": 1,
+                },
+                format="json",
+            )
+
+        self.assertEqual(first_response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(second_response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
+
     def test_reservation_creation_with_insufficient_seats(self):
         response = self.client.post(
             reverse("api-reservation-create", kwargs={"performance_id": self.performance.id}),
@@ -188,6 +216,27 @@ class BookingApiTests(APITestCase):
         self.assertEqual(second_response.status_code, status.HTTP_409_CONFLICT)
         self.assertEqual(second_response.data["status"], "already_confirmed")
 
+    def test_confirmation_is_throttled(self):
+        with patch.dict(ReservationConfirmThrottle.THROTTLE_RATES, {"reservation_confirm": "1/minute"}, clear=False):
+            first_reservation = self.create_reservation(email="first@example.com")
+            _, first_raw_token = generate_confirmation_token(first_reservation)
+            second_reservation = self.create_reservation(email="second@example.com")
+            _, second_raw_token = generate_confirmation_token(second_reservation)
+
+            first_response = self.client.post(
+                reverse("api-reservation-confirm"),
+                {"token": first_raw_token},
+                format="json",
+            )
+            second_response = self.client.post(
+                reverse("api-reservation-confirm"),
+                {"token": second_raw_token},
+                format="json",
+            )
+
+        self.assertEqual(first_response.status_code, status.HTTP_200_OK)
+        self.assertEqual(second_response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
+
     @override_settings(SITE_BASE_URL="https://tickets.azionelab.example")
     def test_qr_retrieval_success_for_confirmed_reservation(self):
         reservation = self.create_reservation()

backend/bookings/views.py

@@ -1,7 +1,8 @@
 from django.shortcuts import get_object_or_404
 from rest_framework import status
-from rest_framework.decorators import api_view
+from rest_framework.decorators import api_view, throttle_classes
 from rest_framework.response import Response
+from rest_framework.throttling import AnonRateThrottle
 
 from shows.models import Performance
 
@@ -25,7 +26,16 @@ from .services import (
 )
 
 
+class ReservationCreateThrottle(AnonRateThrottle):
+    scope = "reservation_create"
+
+
+class ReservationConfirmThrottle(AnonRateThrottle):
+    scope = "reservation_confirm"
+
+
 @api_view(["POST"])
+@throttle_classes([ReservationCreateThrottle])
 def create_reservation(request, performance_id):
     get_object_or_404(Performance, pk=performance_id, show__is_published=True)
 
@@ -60,6 +70,7 @@ def create_reservation(request, performance_id):
 
 
 @api_view(["GET", "POST"])
+@throttle_classes([ReservationConfirmThrottle])
 def confirm_reservation(request):
     payload = request.query_params if request.method == "GET" else request.data
     serializer = ReservationConfirmSerializer(data=payload)

backend/checkins/test_api.py

@@ -1,13 +1,16 @@
 from datetime import timedelta
+from unittest.mock import patch
 
 from django.contrib.auth import get_user_model
 from django.urls import reverse
+from django.test.utils import override_settings
 from django.utils import timezone
 from rest_framework import status
 from rest_framework.test import APITestCase
 
 from bookings.models import Reservation, ReservationToken
 from checkins.models import CheckIn
+from checkins.views import CheckInPreviewThrottle
 from shows.models import Performance, Show, Venue
 
 
@@ -105,6 +108,28 @@ class CheckInApiTests(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
         self.assertEqual(response.data["status"], "invalid_token")
 
+    def test_preview_is_throttled_for_staff_user(self):
+        with patch.dict(CheckInPreviewThrottle.THROTTLE_RATES, {"check_in_preview": "1/minute"}, clear=False):
+            first_reservation = self.create_reservation(email="first@example.com")
+            _, first_raw_token = self.create_check_in_token(first_reservation)
+            second_reservation = self.create_reservation(email="second@example.com")
+            _, second_raw_token = self.create_check_in_token(second_reservation)
+            self.client.force_authenticate(user=self.staff_user)
+
+            first_response = self.client.post(
+                reverse("api-check-in-preview"),
+                {"token": first_raw_token},
+                format="json",
+            )
+            second_response = self.client.post(
+                reverse("api-check-in-preview"),
+                {"token": second_raw_token},
+                format="json",
+            )
+
+        self.assertEqual(first_response.status_code, status.HTTP_200_OK)
+        self.assertEqual(second_response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
+
     def test_check_in_success_as_staff_user(self):
         reservation = self.create_reservation()
         _, raw_token = self.create_check_in_token(reservation)

backend/checkins/views.py

@@ -1,8 +1,9 @@
 from rest_framework import status
 from rest_framework.authentication import BasicAuthentication, SessionAuthentication
-from rest_framework.decorators import api_view, authentication_classes, permission_classes
+from rest_framework.decorators import api_view, authentication_classes, permission_classes, throttle_classes
 from rest_framework.permissions import BasePermission, IsAuthenticated
 from rest_framework.response import Response
+from rest_framework.throttling import UserRateThrottle
 
 from .serializers import (
     CheckInConfirmResponseSerializer,
@@ -19,19 +20,22 @@ from .services import (
 )
 
 
+class CheckInPreviewThrottle(UserRateThrottle):
+    scope = "check_in_preview"
+
+
+class CheckInConfirmThrottle(UserRateThrottle):
+    scope = "check_in_confirm"
+
+
 class IsStaffUser(BasePermission):
     def has_permission(self, request, view):
         return bool(request.user and request.user.is_staff)
 
-
-def staff_check_in_view(view_func):
-    view_func = permission_classes([IsAuthenticated, IsStaffUser])(view_func)
-    view_func = authentication_classes([BasicAuthentication, SessionAuthentication])(view_func)
-    view_func = api_view(["POST"])(view_func)
-    return view_func
-
-
-@staff_check_in_view
+@api_view(["POST"])
+@authentication_classes([BasicAuthentication, SessionAuthentication])
+@permission_classes([IsAuthenticated, IsStaffUser])
+@throttle_classes([CheckInPreviewThrottle])
 def check_in_preview(request):
     serializer = CheckInTokenSerializer(data=request.data)
     serializer.is_valid(raise_exception=True)
@@ -73,7 +77,10 @@ def check_in_preview(request):
     return Response(response_serializer.data)
 
 
-@staff_check_in_view
+@api_view(["POST"])
+@authentication_classes([BasicAuthentication, SessionAuthentication])
+@permission_classes([IsAuthenticated, IsStaffUser])
+@throttle_classes([CheckInConfirmThrottle])
 def check_in_confirm(request):
     serializer = CheckInTokenSerializer(data=request.data)
     serializer.is_valid(raise_exception=True)