65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
from datetime import datetime
|
|
from typing import List
|
|
|
|
from sqlalchemy import and_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.appointment import Appointment, AppointmentStatus
|
|
from app.schemas.appointment import AppointmentCreate, AppointmentUpdate
|
|
|
|
|
|
class CRUDAppointment(CRUDBase[Appointment, AppointmentCreate, AppointmentUpdate]):
|
|
def get_by_doctor_and_date_range(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
doctor_id: int,
|
|
start_date: datetime,
|
|
end_date: datetime
|
|
) -> List[Appointment]:
|
|
return (
|
|
db.query(Appointment)
|
|
.filter(
|
|
and_(
|
|
Appointment.doctor_id == doctor_id,
|
|
Appointment.appointment_datetime >= start_date,
|
|
Appointment.appointment_datetime <= end_date
|
|
)
|
|
)
|
|
.all()
|
|
)
|
|
|
|
def get_by_patient(
|
|
self, db: Session, *, patient_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Appointment]:
|
|
return (
|
|
db.query(Appointment)
|
|
.filter(Appointment.patient_id == patient_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_doctor(
|
|
self, db: Session, *, doctor_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Appointment]:
|
|
return (
|
|
db.query(Appointment)
|
|
.filter(Appointment.doctor_id == doctor_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def update_status(
|
|
self, db: Session, *, db_obj: Appointment, status: AppointmentStatus
|
|
) -> Appointment:
|
|
db_obj.status = status
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
|
|
appointment = CRUDAppointment(Appointment) |