73 lines
3.0 KiB
Python
73 lines
3.0 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
from models import MasterAppointmentTypes
|
|
from schemas.BaseSchemas import MasterAppointmentTypeBase
|
|
from exceptions import ResourceNotFoundException
|
|
from interface.common_response import CommonResponse
|
|
from schemas.ResponseSchemas import MasterAppointmentTypeResponse
|
|
|
|
class MasterAppointmentServices:
|
|
def __init__(self):
|
|
self.db: Session = next(get_db())
|
|
|
|
|
|
def get_master_appointment_types(self):
|
|
appointment_types = self.db.query(MasterAppointmentTypes).all()
|
|
total= self.db.query(MasterAppointmentTypes).count()
|
|
response = CommonResponse(data=[MasterAppointmentTypeResponse(**appointment_type.__dict__.copy()) for appointment_type in appointment_types], total=total)
|
|
return response
|
|
|
|
def is_appointment_type_exists(self, appointment_type: MasterAppointmentTypeBase):
|
|
existing_appointment_type = self.db.query(MasterAppointmentTypes).filter(MasterAppointmentTypes.type == appointment_type.type.lower()).first()
|
|
return existing_appointment_type
|
|
|
|
def create_master_appointment_type(self, appointment_type: MasterAppointmentTypeBase):
|
|
|
|
# get existing appointment type
|
|
existing_appointment_type = self.is_appointment_type_exists(appointment_type)
|
|
|
|
if existing_appointment_type:
|
|
return
|
|
|
|
appointment_type.type = appointment_type.type.lower()
|
|
|
|
appointment_type = MasterAppointmentTypes(**appointment_type.model_dump())
|
|
self.db.add(appointment_type)
|
|
self.db.commit()
|
|
self.db.refresh(appointment_type)
|
|
return
|
|
|
|
def delete_master_appointment_type(self, appointment_type_id: int):
|
|
appointment_type = self.db.query(MasterAppointmentTypes).filter(MasterAppointmentTypes.id == appointment_type_id).first()
|
|
|
|
if appointment_type is None:
|
|
raise ResourceNotFoundException("Appointment type not found")
|
|
|
|
appointment_type.soft_delete(self.db)
|
|
self.db.commit()
|
|
return
|
|
|
|
def update_master_appointment_type(self, appointment_type_id: int, appointment_type: MasterAppointmentTypeBase):
|
|
appointment_type = self.db.query(MasterAppointmentTypes).filter(MasterAppointmentTypes.id == appointment_type_id).first()
|
|
|
|
if appointment_type is None:
|
|
raise ResourceNotFoundException("Appointment type not found")
|
|
|
|
appointment_type.type = appointment_type.type.lower()
|
|
|
|
# get existing appointment type
|
|
existing_appointment_type = self.is_appointment_type_exists(appointment_type)
|
|
|
|
if existing_appointment_type and existing_appointment_type.id != appointment_type_id:
|
|
raise ResourceNotFoundException("Appointment type already exists")
|
|
|
|
update_data = appointment_type.model_dump(exclude_unset=True)
|
|
|
|
for key, value in update_data.items():
|
|
setattr(appointment_type, key, value)
|
|
|
|
self.db.add(appointment_type)
|
|
self.db.commit()
|
|
self.db.refresh(appointment_type)
|
|
return |