Project: A Clinic Booking API

python -m pip install django djangorestframework djangorestframework-simplejwt drf-spectacular django-filter pytest pytest-django factory-boy
python manage.py startapp accounts      # custom User, created before the first migrate
python manage.py startapp doctors       # Doctor, Slot
python manage.py startapp bookings      # Appointment, booking rules

# config/settings/base.py (excerpt)
INSTALLED_APPS = [
    # ... Django's contrib apps
    "rest_framework",
    "rest_framework_simplejwt.token_blacklist",
    "django_filters",
    "drf_spectacular",
    "accounts",
    "doctors",
    "bookings",
]
AUTH_USER_MODEL = "accounts.User"
TIME_ZONE = "Asia/Kolkata"
USE_TZ = True

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ["rest_framework_simplejwt.authentication.JWTAuthentication"],
    "DEFAULT_PERMISSION_CLASSES": ["rest_framework.permissions.IsAuthenticated"],
    "DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.LimitOffsetPagination",
    "PAGE_SIZE": 20,
    "DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"],
    "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema",
}
TASKS = {"default": {"BACKEND": "django.tasks.backends.immediate.ImmediateBackend"}}
CANCEL_HOURS = 2

# accounts/models.py
class User(AbstractUser):
    phone = models.CharField(max_length=10, blank=True)


# doctors/models.py
class Doctor(models.Model):
    user = models.OneToOneField(settings.AUTH_USER_MODEL, on_delete=models.PROTECT,
                                related_name="doctor")
    name = models.CharField(max_length=100)
    speciality = models.CharField(max_length=60)
    fee = models.PositiveIntegerField(help_text="Consultation fee in rupees")

    def __str__(self):
        return f"Dr. {self.name} ({self.speciality})"


class Slot(models.Model):
    doctor = models.ForeignKey(Doctor, on_delete=models.CASCADE, related_name="slots")
    starts_at = models.DateTimeField()
    minutes = models.PositiveSmallIntegerField(default=15)

    class Meta:
        ordering = ["starts_at"]
        constraints = [
            models.UniqueConstraint(fields=["doctor", "starts_at"], name="one_slot_per_doctor_time"),
        ]


# bookings/models.py
class Appointment(models.Model):
    class Status(models.TextChoices):
        BOOKED = "booked", "Booked"
        CANCELLED = "cancelled", "Cancelled"

    slot = models.ForeignKey("doctors.Slot", on_delete=models.PROTECT, related_name="appointments")
    patient = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT,
                                related_name="appointments")
    reason = models.CharField(max_length=200, blank=True)
    status = models.CharField(max_length=10, choices=Status, default=Status.BOOKED)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["slot"], condition=models.Q(status="booked"),
                name="one_active_booking_per_slot"),
        ]

# bookings/services.py
class BookingError(Exception):
    pass


@transaction.atomic
def book_slot(*, patient, slot_id: int, reason: str = "") -> Appointment:
    slot = Slot.objects.select_for_update().select_related("doctor").get(pk=slot_id)
    if slot.starts_at <= timezone.now():
        raise BookingError("This slot has already started.")
    if Appointment.objects.filter(slot=slot, status=Appointment.Status.BOOKED).exists():
        raise BookingError("This slot is already booked.")
    local_day = timezone.localtime(slot.starts_at).date()
    same_day = Appointment.objects.filter(
        patient=patient, status=Appointment.Status.BOOKED,
        slot__doctor=slot.doctor, slot__starts_at__date=local_day)
    if same_day.exists():
        raise BookingError("You already have an appointment with this doctor that day.")
    try:
        with transaction.atomic():
            appt = Appointment.objects.create(slot=slot, patient=patient, reason=reason)
    except IntegrityError:                      # lost a race on a database without row locks
        raise BookingError("This slot is already booked.")
    transaction.on_commit(partial(send_booking_sms.enqueue, appt.pk))
    logger.info("Appointment %s booked: slot %s, patient %s", appt.pk, slot.pk, patient.pk)
    return appt


@transaction.atomic
def cancel_appointment(*, appointment: Appointment, by) -> Appointment:
    appt = Appointment.objects.select_for_update().select_related("slot").get(pk=appointment.pk)
    if appt.status == Appointment.Status.CANCELLED:
        raise BookingError("Already cancelled.")
    limit = appt.slot.starts_at - timedelta(hours=settings.CANCEL_HOURS)
    if not by.is_staff and timezone.now() > limit:
        raise BookingError(f"Cancel at least {settings.CANCEL_HOURS} hours before; call the desk.")
    appt.status = Appointment.Status.CANCELLED
    appt.save(update_fields=["status"])
    logger.info("Appointment %s cancelled by user %s", appt.pk, by.pk)
    return appt

# bookings/tasks.py
@task
def send_booking_sms(appointment_id: int) -> str:
    appt = Appointment.objects.select_related("slot__doctor", "patient").get(pk=appointment_id)
    when = timezone.localtime(appt.slot.starts_at)
    text = f"Confirmed: {appt.slot.doctor} on {when:%d %b at %H:%M}. Kochi Care Clinic"
    # A real project calls the SMS provider here.
    return f"to {appt.patient.phone}: {text}"

# doctors/selectors.py
def open_slots():
    """Future slots without an active booking."""
    taken = Appointment.objects.filter(slot=OuterRef("pk"), status=Appointment.Status.BOOKED)
    return (Slot.objects.select_related("doctor")
            .filter(starts_at__gt=timezone.now())
            .exclude(Exists(taken)))


# doctors/api.py
@method_decorator(cache_page(60 * 10), name="list")
class DoctorViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = Doctor.objects.order_by("name")
    serializer_class = DoctorSerializer
    permission_classes = [permissions.AllowAny]


class OpenSlotViewSet(viewsets.ReadOnlyModelViewSet):
    serializer_class = SlotSerializer                # doctor shown with StringRelatedField
    permission_classes = [permissions.AllowAny]
    filterset_fields = {"doctor": ["exact"], "starts_at": ["date"]}

    def get_queryset(self):
        return open_slots()

# bookings/api.py
class IsPatientOrStaff(permissions.BasePermission):
    message = "This appointment belongs to another patient."

    def has_object_permission(self, request, view, obj):
        return request.user.is_staff or obj.patient_id == request.user.id


class AppointmentSerializer(serializers.ModelSerializer):
    slot_id = serializers.IntegerField(write_only=True)
    doctor = serializers.CharField(source="slot.doctor", read_only=True)
    starts_at = serializers.DateTimeField(source="slot.starts_at", read_only=True)

    class Meta:
        model = Appointment
        fields = ["id", "slot_id", "doctor", "starts_at", "reason", "status"]
        read_only_fields = ["status"]


class AppointmentViewSet(mixins.CreateModelMixin, mixins.ListModelMixin,
                         mixins.RetrieveModelMixin, viewsets.GenericViewSet):
    serializer_class = AppointmentSerializer
    permission_classes = [permissions.IsAuthenticated, IsPatientOrStaff]

    def get_queryset(self):
        if getattr(self, "swagger_fake_view", False):
            return Appointment.objects.none()
        appts = Appointment.objects.select_related("slot__doctor").order_by("slot__starts_at")
        if not self.request.user.is_staff:
            appts = appts.filter(patient=self.request.user)
        return appts

    def perform_create(self, serializer):
        try:
            serializer.instance = book_slot(
                patient=self.request.user,
                slot_id=serializer.validated_data["slot_id"],
                reason=serializer.validated_data.get("reason", ""))
        except BookingError as error:
            raise serializers.ValidationError({"detail": str(error)})

    @extend_schema(request=None, responses={200: AppointmentSerializer})
    @action(detail=True, methods=["post"])
    def cancel(self, request, pk=None):
        try:
            appt = cancel_appointment(appointment=self.get_object(), by=request.user)
        except BookingError as error:
            return Response({"detail": str(error)}, status=status.HTTP_400_BAD_REQUEST)
        return Response(self.get_serializer(appt).data)

# config/urls.py
router = DefaultRouter()
router.register("doctors", DoctorViewSet)
router.register("slots", OpenSlotViewSet, basename="slot")
router.register("appointments", AppointmentViewSet, basename="appointment")

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/token/", TokenObtainPairView.as_view(), name="token"),
    path("api/token/refresh/", TokenRefreshView.as_view(), name="token-refresh"),
    path("api/schema/", SpectacularAPIView.as_view(), name="schema"),
    path("api/docs/", SpectacularSwaggerView.as_view(url_name="schema"), name="docs"),
    path("api/", include(router.urls)),
]

python manage.py spectacular --file schema.yml --validate
grep -E "^  /api" schema.yml

# doctors/management/commands/make_slots.py (core of handle())
tz = timezone.get_current_timezone()
first = timezone.localdate() + timedelta(days=1)
new = []
for doctor in Doctor.objects.all():
    for d in range(days):
        day = first + timedelta(days=d)
        if day.weekday() == 6:          # clinic closed on Sundays
            continue
        t = datetime.combine(day, start_t, tz)
        while t.time() < end_t:
            new.append(Slot(doctor=doctor, starts_at=t))
            t += timedelta(minutes=15)
before = Slot.objects.count()
Slot.objects.bulk_create(new, ignore_conflicts=True)     # skip slots that already exist
added = Slot.objects.count() - before
self.stdout.write(self.style.SUCCESS(f"Added {added} new slots ({len(new)} checked)."))

python manage.py make_slots --days 3
python manage.py make_slots --days 3

# tests/test_booking.py
pytestmark = pytest.mark.django_db


def test_slot_cannot_be_booked_twice():
    slot = SlotFactory()
    book_slot(patient=UserFactory(), slot_id=slot.pk)
    with pytest.raises(BookingError, match="already booked"):
        book_slot(patient=UserFactory(), slot_id=slot.pk)


def test_past_slot_is_refused():
    slot = SlotFactory(starts_at=timezone.now() - timedelta(minutes=5))
    with pytest.raises(BookingError, match="already started"):
        book_slot(patient=UserFactory(), slot_id=slot.pk)


@pytest.mark.parametrize("hours_before, is_staff, allowed", [
    (5, False, True), (1, False, False), (1, True, True),
])
def test_cancel_window(hours_before, is_staff, allowed):
    appt = AppointmentFactory(slot__starts_at=timezone.now() + timedelta(hours=hours_before))
    user = UserFactory(is_staff=True) if is_staff else appt.patient
    if allowed:
        assert cancel_appointment(appointment=appt, by=user).status == "cancelled"
    else:
        with pytest.raises(BookingError, match="at least 2 hours"):
            cancel_appointment(appointment=appt, by=user)


def test_sms_task_queued_after_commit(django_capture_on_commit_callbacks):
    default_task_backend.clear()
    with django_capture_on_commit_callbacks(execute=True):
        appt = book_slot(patient=UserFactory(), slot_id=SlotFactory().pk)
    assert [r.args for r in default_task_backend.results] == [[appt.pk]]


def test_patients_see_only_their_appointments(django_assert_num_queries):
    mine = AppointmentFactory()
    AppointmentFactory.create_batch(3)
    client = APIClient()
    client.force_authenticate(mine.patient)
    with django_assert_num_queries(2):
        data = client.get("/api/appointments/").json()
    assert [a["id"] for a in data["results"]] == [mine.pk]