status_timestamp.py 1.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142
  1. from app.schemas.device.device import Device
  2. from typing import Optional
  3. from sqlalchemy.orm import Session
  4. from app.crud.base import CRUDBase
  5. from app.models.devices.status_timestamp import BlowyFeedbackTime, HighSpeedTime
  6. from app.schemas.device.status_timestamp import (
  7. BlowyFeedbackTimeCreate,
  8. BlowyFeedbackTimeUpdate,
  9. HighSpeedTimeCreate,
  10. HighSpeedTimeUpdate,
  11. )
  12. class CRUDBlowyFeedbackTime(
  13. CRUDBase[
  14. BlowyFeedbackTime,
  15. BlowyFeedbackTimeCreate,
  16. BlowyFeedbackTimeUpdate,
  17. ]
  18. ):
  19. def get_time_by_device(
  20. self, db: Session, device_id: str
  21. ) -> Optional[BlowyFeedbackTime]:
  22. return (
  23. db.query(self.model)
  24. .filter(BlowyFeedbackTime.device_id == device_id)
  25. .first()
  26. )
  27. class CRUDHighSpeedTime(
  28. CRUDBase[HighSpeedTime, HighSpeedTimeCreate, HighSpeedTimeUpdate]
  29. ):
  30. def get_time_by_device(
  31. self, db: Session, device_id: str
  32. ) -> Optional[HighSpeedTime]:
  33. return db.query(self.model).filter(HighSpeedTime.device_id == device_id).first()
  34. blowy_feedback_time = CRUDBlowyFeedbackTime(BlowyFeedbackTime)
  35. high_speed_time = CRUDHighSpeedTime(HighSpeedTime)