52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
from database import get_db
|
|
from sqlalchemy.orm import Session
|
|
from models import Clinics
|
|
from schemas.UpdateSchemas import ClinicUpdate
|
|
from schemas.ResponseSchemas import Clinic
|
|
from typing import List
|
|
from exceptions import ResourceNotFoundException
|
|
|
|
class ClinicServices:
|
|
def __init__(self):
|
|
self.db: Session = next(get_db())
|
|
|
|
def get_clinics(self, limit:int, offset:int) -> List[Clinic]:
|
|
clinics = self.db.query(Clinics).limit(limit).offset(offset).all()
|
|
|
|
clinic_response = [Clinic(**clinic.__dict__.copy()) for clinic in clinics]
|
|
|
|
return clinic_response
|
|
|
|
|
|
def get_clinic_by_id(self, clinic_id: int) -> Clinic:
|
|
clinic = self.db.query(Clinics).filter(Clinics.id == clinic_id).first()
|
|
|
|
if clinic is None:
|
|
raise ResourceNotFoundException("Clinic not found")
|
|
|
|
clinic_response = Clinic(**clinic.__dict__.copy())
|
|
return clinic_response
|
|
|
|
def update_clinic(self, clinic_id: int, clinic_data: ClinicUpdate):
|
|
clinic = self.db.query(Clinics).filter(Clinics.id == clinic_id).first()
|
|
|
|
if clinic is None:
|
|
raise ResourceNotFoundException("Clinic not found")
|
|
|
|
update_data = clinic_data.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(clinic, key, value)
|
|
|
|
self.db.add(clinic)
|
|
self.db.commit()
|
|
self.db.refresh(clinic)
|
|
return Clinic(**clinic.__dict__.copy())
|
|
|
|
def delete_clinic(self, clinic_id: int):
|
|
clinic = self.db.query(Clinics).filter(Clinics.id == clinic_id).first()
|
|
|
|
if clinic is None:
|
|
raise ResourceNotFoundException("Clinic not found")
|
|
|
|
self.db.delete(clinic)
|
|
self.db.commit() |