291 lines
12 KiB
Python
291 lines
12 KiB
Python
from datetime import timedelta
|
|
from typing import Any, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import crud, models, schemas
|
|
from app.api import deps
|
|
from app.models.appointment import AppointmentStatus
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[schemas.appointment.Appointment])
|
|
def read_appointments(
|
|
db: Session = Depends(deps.get_db),
|
|
doctor_id: Optional[int] = Query(None, description="Filter appointments by doctor"),
|
|
patient_id: Optional[int] = Query(None, description="Filter appointments by patient"),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Retrieve appointments.
|
|
"""
|
|
# Determine what appointments the user can see
|
|
if crud.crud_user.user.is_superuser(current_user):
|
|
# Superusers can see all appointments with optional filtering
|
|
if doctor_id:
|
|
appointments = crud.crud_appointment.appointment.get_by_doctor(
|
|
db, doctor_id=doctor_id, skip=skip, limit=limit
|
|
)
|
|
elif patient_id:
|
|
appointments = crud.crud_appointment.appointment.get_by_patient(
|
|
db, patient_id=patient_id, skip=skip, limit=limit
|
|
)
|
|
else:
|
|
appointments = crud.crud_appointment.appointment.get_multi(
|
|
db, skip=skip, limit=limit
|
|
)
|
|
else:
|
|
# Regular users can only see their own appointments
|
|
if current_user.doctor:
|
|
# Doctors can see all their appointments
|
|
appointments = crud.crud_appointment.appointment.get_by_doctor(
|
|
db, doctor_id=current_user.doctor.id, skip=skip, limit=limit
|
|
)
|
|
elif current_user.patient:
|
|
# Patients can see all their appointments
|
|
appointments = crud.crud_appointment.appointment.get_by_patient(
|
|
db, patient_id=current_user.patient.id, skip=skip, limit=limit
|
|
)
|
|
else:
|
|
# Users without doctor or patient profiles can't see any appointments
|
|
appointments = []
|
|
|
|
return appointments
|
|
|
|
|
|
@router.post("/", response_model=schemas.appointment.Appointment)
|
|
def create_appointment(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
appointment_in: schemas.appointment.AppointmentCreate,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Create new appointment.
|
|
"""
|
|
# Check if doctor exists
|
|
doctor = crud.crud_doctor.doctor.get(db, id=appointment_in.doctor_id)
|
|
if not doctor:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Doctor not found",
|
|
)
|
|
|
|
# Check if patient exists
|
|
patient = crud.crud_patient.patient.get(db, id=appointment_in.patient_id)
|
|
if not patient:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Patient not found",
|
|
)
|
|
|
|
# Check permissions
|
|
if not crud.crud_user.user.is_superuser(current_user):
|
|
# Only the patient or the doctor involved can create an appointment
|
|
if current_user.patient:
|
|
if current_user.patient.id != appointment_in.patient_id:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Not enough permissions to create an appointment for another patient",
|
|
)
|
|
elif current_user.doctor:
|
|
if current_user.doctor.id != appointment_in.doctor_id:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Not enough permissions to create an appointment for another doctor",
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Not enough permissions to create an appointment",
|
|
)
|
|
|
|
# Check for scheduling conflicts
|
|
appointment_end = appointment_in.appointment_datetime + timedelta(minutes=appointment_in.duration_minutes)
|
|
doctor_appointments = crud.crud_appointment.appointment.get_by_doctor_and_date_range(
|
|
db,
|
|
doctor_id=appointment_in.doctor_id,
|
|
start_date=appointment_in.appointment_datetime - timedelta(hours=1), # Buffer before
|
|
end_date=appointment_end + timedelta(hours=1), # Buffer after
|
|
)
|
|
|
|
for existing_appointment in doctor_appointments:
|
|
# Skip cancelled appointments
|
|
if existing_appointment.status == AppointmentStatus.CANCELLED:
|
|
continue
|
|
|
|
existing_end = existing_appointment.appointment_datetime + timedelta(minutes=existing_appointment.duration_minutes)
|
|
|
|
# Check for overlap
|
|
if (appointment_in.appointment_datetime < existing_end and
|
|
existing_appointment.appointment_datetime < appointment_end):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Scheduling conflict with an existing appointment at {existing_appointment.appointment_datetime}",
|
|
)
|
|
|
|
appointment = crud.crud_appointment.appointment.create(db, obj_in=appointment_in)
|
|
return appointment
|
|
|
|
|
|
@router.get("/{appointment_id}", response_model=schemas.appointment.Appointment)
|
|
def read_appointment(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
appointment_id: int,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get appointment by ID.
|
|
"""
|
|
appointment = crud.crud_appointment.appointment.get(db, id=appointment_id)
|
|
if not appointment:
|
|
raise HTTPException(status_code=404, detail="Appointment not found")
|
|
|
|
# Check permissions
|
|
if not crud.crud_user.user.is_superuser(current_user):
|
|
# Regular users can only see their own appointments
|
|
if current_user.patient:
|
|
if current_user.patient.id != appointment.patient_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
elif current_user.doctor:
|
|
if current_user.doctor.id != appointment.doctor_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
else:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
|
|
return appointment
|
|
|
|
|
|
@router.put("/{appointment_id}", response_model=schemas.appointment.Appointment)
|
|
def update_appointment(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
appointment_id: int,
|
|
appointment_in: schemas.appointment.AppointmentUpdate,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Update an appointment.
|
|
"""
|
|
appointment = crud.crud_appointment.appointment.get(db, id=appointment_id)
|
|
if not appointment:
|
|
raise HTTPException(status_code=404, detail="Appointment not found")
|
|
|
|
# Check permissions
|
|
if not crud.crud_user.user.is_superuser(current_user):
|
|
# Regular users can only update their own appointments
|
|
if current_user.patient:
|
|
if current_user.patient.id != appointment.patient_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
elif current_user.doctor:
|
|
if current_user.doctor.id != appointment.doctor_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
else:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
|
|
# If appointment_datetime is updated, check for scheduling conflicts
|
|
if appointment_in.appointment_datetime and appointment_in.appointment_datetime != appointment.appointment_datetime:
|
|
appointment_end = appointment_in.appointment_datetime + timedelta(
|
|
minutes=appointment_in.duration_minutes if appointment_in.duration_minutes else appointment.duration_minutes
|
|
)
|
|
|
|
doctor_appointments = crud.crud_appointment.appointment.get_by_doctor_and_date_range(
|
|
db,
|
|
doctor_id=appointment.doctor_id,
|
|
start_date=appointment_in.appointment_datetime - timedelta(hours=1),
|
|
end_date=appointment_end + timedelta(hours=1),
|
|
)
|
|
|
|
for existing_appointment in doctor_appointments:
|
|
# Skip the current appointment and cancelled appointments
|
|
if existing_appointment.id == appointment_id or existing_appointment.status == AppointmentStatus.CANCELLED:
|
|
continue
|
|
|
|
existing_end = existing_appointment.appointment_datetime + timedelta(minutes=existing_appointment.duration_minutes)
|
|
|
|
# Check for overlap
|
|
if (appointment_in.appointment_datetime < existing_end and
|
|
existing_appointment.appointment_datetime < appointment_end):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Scheduling conflict with an existing appointment at {existing_appointment.appointment_datetime}",
|
|
)
|
|
|
|
appointment = crud.crud_appointment.appointment.update(
|
|
db, db_obj=appointment, obj_in=appointment_in
|
|
)
|
|
return appointment
|
|
|
|
|
|
@router.delete("/{appointment_id}", response_model=schemas.appointment.Appointment)
|
|
def delete_appointment(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
appointment_id: int,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Delete an appointment.
|
|
"""
|
|
appointment = crud.crud_appointment.appointment.get(db, id=appointment_id)
|
|
if not appointment:
|
|
raise HTTPException(status_code=404, detail="Appointment not found")
|
|
|
|
# Check permissions
|
|
if not crud.crud_user.user.is_superuser(current_user):
|
|
# Regular users can only delete their own appointments
|
|
if current_user.patient:
|
|
if current_user.patient.id != appointment.patient_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
elif current_user.doctor:
|
|
if current_user.doctor.id != appointment.doctor_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
else:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
|
|
appointment = crud.crud_appointment.appointment.remove(db, id=appointment_id)
|
|
return appointment
|
|
|
|
|
|
@router.post("/{appointment_id}/status", response_model=schemas.appointment.Appointment)
|
|
def update_appointment_status(
|
|
*,
|
|
db: Session = Depends(deps.get_db),
|
|
appointment_id: int,
|
|
status: AppointmentStatus,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Update appointment status.
|
|
"""
|
|
appointment = crud.crud_appointment.appointment.get(db, id=appointment_id)
|
|
if not appointment:
|
|
raise HTTPException(status_code=404, detail="Appointment not found")
|
|
|
|
# Check permissions
|
|
if not crud.crud_user.user.is_superuser(current_user):
|
|
# Regular users can only update their own appointments
|
|
if current_user.patient:
|
|
if current_user.patient.id != appointment.patient_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
# Patients can only cancel their appointments
|
|
if status not in [AppointmentStatus.CANCELLED]:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Patients can only cancel appointments"
|
|
)
|
|
elif current_user.doctor:
|
|
if current_user.doctor.id != appointment.doctor_id:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
else:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
|
|
appointment = crud.crud_appointment.appointment.update_status(
|
|
db, db_obj=appointment, status=status
|
|
)
|
|
return appointment |