from sqlalchemy.orm import Session from sqlalchemy import select from typing import List, Optional from app.repositories.base_repository import BaseRepository from app.models.TechnicianModel import TechnicianProfile, TechnicianSkill, TechnicianWorkingHours, TechnicianLeave class TechnicianRepository(BaseRepository[TechnicianProfile]): def __init__(self): super().__init__(TechnicianProfile) def get_by_user_id(self, db: Session, user_id: str) -> Optional[TechnicianProfile]: stmt = select(TechnicianProfile).where(TechnicianProfile.user_id == user_id) return db.execute(stmt).scalar_one_or_none() def get_active_technicians_by_skill(self, db: Session, service_id: str) -> List[TechnicianProfile]: stmt = ( select(TechnicianProfile) .join(TechnicianSkill, TechnicianProfile.technician_id == TechnicianSkill.technician_id) .where( TechnicianProfile.active == True, TechnicianSkill.service_id == service_id ) ) return list(db.execute(stmt).scalars().all()) def get_technician_working_hours(self, db: Session, technician_id: str) -> List[TechnicianWorkingHours]: stmt = select(TechnicianWorkingHours).where(TechnicianWorkingHours.technician_id == technician_id) return list(db.execute(stmt).scalars().all()) def get_technician_leaves(self, db: Session, technician_id: str) -> List[TechnicianLeave]: stmt = select(TechnicianLeave).where(TechnicianLeave.technician_id == technician_id) return list(db.execute(stmt).scalars().all()) def is_technician_on_leave_during(self, db: Session, technician_id: str, start_time, end_time) -> bool: stmt = select(TechnicianLeave).where( TechnicianLeave.technician_id == technician_id, TechnicianLeave.start_datetime < end_time, TechnicianLeave.end_datetime > start_time ) return db.execute(stmt).first() is not None