330 lines
13 KiB
Python
330 lines
13 KiB
Python
|
"""This module handles CRUD operations for vitals records in the database, based on pydanctic schemas."""
|
||
|
|
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from backend.models import records as recordmodel
|
||
|
|
from backend.models import devices as devicemodel
|
||
|
|
from backend.schemas import records as recordschema
|
||
|
|
from backend.exceptions import NotFoundException
|
||
|
|
|
||
|
|
|
||
|
|
def create_heart_rate_record(db: Session, record: recordschema.HeartRateRecordCreate, device_id: int) -> recordschema.HeartRateRecord:
|
||
|
|
"""Creates the specified heart rate record in the database."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
db_record = recordmodel.HeartRateRecord(
|
||
|
|
value=record.value,
|
||
|
|
measured=record.measured,
|
||
|
|
device_id=device_id,
|
||
|
|
)
|
||
|
|
db.add(db_record)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(db_record)
|
||
|
|
|
||
|
|
return recordschema.HeartRateRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
|
||
|
|
def read_heart_rate_records_by_device(
|
||
|
|
db: Session, device_id: int,
|
||
|
|
since: datetime | None = None, until: datetime | None = None,
|
||
|
|
skip: int = 0, limit: int = 100,
|
||
|
|
) -> list[recordschema.HeartRateRecord]:
|
||
|
|
"""Queries the db for a range of heart rate records captured by the device with the specified id."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
filters = [recordmodel.HeartRateRecord.device == db_device]
|
||
|
|
if since:
|
||
|
|
filters.append(recordmodel.HeartRateRecord.measured >= since)
|
||
|
|
if until:
|
||
|
|
filters.append(recordmodel.HeartRateRecord.measured <= until)
|
||
|
|
|
||
|
|
return db.query(recordmodel.HeartRateRecord).filter(*filters).order_by(recordmodel.HeartRateRecord.measured.desc()).offset(skip).limit(limit).all()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_heart_rate_record(db: Session, record_id: int) -> recordschema.HeartRateRecord:
|
||
|
|
"""Deletes the heart rate record with the provided id from the db."""
|
||
|
|
|
||
|
|
db_record = db.query(recordmodel.HeartRateRecord).filter(recordmodel.HeartRateRecord.id == record_id).first()
|
||
|
|
if not db_record:
|
||
|
|
raise NotFoundException(f"Record with id '{record_id}' does not exist.")
|
||
|
|
record_copy = recordschema.HeartRateRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
db.delete(db_record)
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
return record_copy
|
||
|
|
|
||
|
|
|
||
|
|
def create_avpu_score_record(db: Session, record: recordschema.AvpuScoreRecordCreate, device_id: int) -> recordschema.AvpuScoreRecord:
|
||
|
|
"""Creates the specified AVPU score record in the database."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
db_record = recordmodel.AvpuScoreRecord(
|
||
|
|
value=record.value,
|
||
|
|
measured=record.measured,
|
||
|
|
device_id=device_id,
|
||
|
|
)
|
||
|
|
db.add(db_record)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(db_record)
|
||
|
|
|
||
|
|
return recordschema.AvpuScoreRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
|
||
|
|
def read_avpu_score_records_by_device(
|
||
|
|
db: Session, device_id: int,
|
||
|
|
since: datetime | None = None, until: datetime | None = None,
|
||
|
|
skip: int = 0, limit: int = 100,
|
||
|
|
) -> list[recordschema.AvpuScoreRecord]:
|
||
|
|
"""Queries the db for a range of avpu score records captured by the device with the specified id."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
filters = [recordmodel.AvpuScoreRecord.device == db_device]
|
||
|
|
if since:
|
||
|
|
filters.append(recordmodel.AvpuScoreRecord.measured >= since)
|
||
|
|
if until:
|
||
|
|
filters.append(recordmodel.AvpuScoreRecord.measured <= until)
|
||
|
|
|
||
|
|
return db.query(recordmodel.AvpuScoreRecord).filter(*filters).order_by(recordmodel.AvpuScoreRecord.measured.desc()).offset(skip).limit(limit).all()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_avpu_score_record(db: Session, record_id: int) -> recordschema.AvpuScoreRecord:
|
||
|
|
"""Deletes the avpu score record with the provided id from the db."""
|
||
|
|
|
||
|
|
db_record = db.query(recordmodel.AvpuScoreRecord).filter(recordmodel.AvpuScoreRecord.id == record_id).first()
|
||
|
|
if not db_record:
|
||
|
|
raise NotFoundException(f"Record with id '{record_id}' does not exist.")
|
||
|
|
record_copy = recordschema.AvpuScoreRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
db.delete(db_record)
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
return record_copy
|
||
|
|
|
||
|
|
|
||
|
|
def create_body_temperature_record(db: Session, record: recordschema.BodyTemperatureRecordCreate, device_id: int) -> recordschema.BodyTemperatureRecord:
|
||
|
|
"""Creates the specified body temperature record in the database."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
db_record = recordmodel.BodyTemperatureRecord(
|
||
|
|
value=record.value,
|
||
|
|
measured=record.measured,
|
||
|
|
device_id=device_id,
|
||
|
|
)
|
||
|
|
db.add(db_record)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(db_record)
|
||
|
|
|
||
|
|
return recordschema.BodyTemperatureRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
|
||
|
|
def read_body_temperature_records_by_device(
|
||
|
|
db: Session, device_id: int,
|
||
|
|
since: datetime | None = None, until: datetime | None = None,
|
||
|
|
skip: int = 0, limit: int = 100,
|
||
|
|
) -> list[recordschema.BodyTemperatureRecord]:
|
||
|
|
"""Queries the db for a range of body temperature records captured by the device with the specified id."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
filters = [recordmodel.BodyTemperatureRecord.device == db_device]
|
||
|
|
if since:
|
||
|
|
filters.append(recordmodel.BodyTemperatureRecord.measured >= since)
|
||
|
|
if until:
|
||
|
|
filters.append(recordmodel.BodyTemperatureRecord.measured <= until)
|
||
|
|
|
||
|
|
return db.query(recordmodel.BodyTemperatureRecord).filter(*filters).order_by(recordmodel.BodyTemperatureRecord.measured.desc()).offset(skip).limit(limit).all()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_body_temperature_record(db: Session, record_id: int) -> recordschema.BodyTemperatureRecord:
|
||
|
|
"""Deletes the body temperature record with the provided id from the db."""
|
||
|
|
|
||
|
|
db_record = db.query(recordmodel.BodyTemperatureRecord).filter(recordmodel.BodyTemperatureRecord.id == record_id).first()
|
||
|
|
if not db_record:
|
||
|
|
raise NotFoundException(f"Record with id '{record_id}' does not exist.")
|
||
|
|
record_copy = recordschema.BodyTemperatureRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
db.delete(db_record)
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
return record_copy
|
||
|
|
|
||
|
|
|
||
|
|
def create_blood_pressure_record(db: Session, record: recordschema.BloodPressureRecordCreate, device_id: int) -> recordschema.BloodPressureRecord:
|
||
|
|
"""Creates the specified blood pressure record in the database."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
db_record = recordmodel.BloodPressureRecord(
|
||
|
|
value_systolic=record.value_systolic,
|
||
|
|
value_diastolic=record.value_diastolic,
|
||
|
|
measured=record.measured,
|
||
|
|
device_id=device_id,
|
||
|
|
)
|
||
|
|
db.add(db_record)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(db_record)
|
||
|
|
|
||
|
|
return recordschema.BloodPressureRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
|
||
|
|
def read_blood_pressure_records_by_device(
|
||
|
|
db: Session, device_id: int,
|
||
|
|
since: datetime | None = None, until: datetime | None = None,
|
||
|
|
skip: int = 0, limit: int = 100,
|
||
|
|
) -> list[recordschema.BloodPressureRecord]:
|
||
|
|
"""Queries the db for a range of blood pressure records captured by the device with the specified id."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
filters = [recordmodel.BloodPressureRecord.device == db_device]
|
||
|
|
if since:
|
||
|
|
filters.append(recordmodel.BloodPressureRecord.measured >= since)
|
||
|
|
if until:
|
||
|
|
filters.append(recordmodel.BloodPressureRecord.measured <= until)
|
||
|
|
|
||
|
|
return db.query(recordmodel.BloodPressureRecord).filter(*filters).order_by(recordmodel.BloodPressureRecord.measured.desc()).offset(skip).limit(limit).all()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_blood_pressure_record(db: Session, record_id: int) -> recordschema.BloodPressureRecord:
|
||
|
|
"""Deletes the blood pressure record with the provided id from the db."""
|
||
|
|
|
||
|
|
db_record = db.query(recordmodel.BloodPressureRecord).filter(recordmodel.BloodPressureRecord.id == record_id).first()
|
||
|
|
if not db_record:
|
||
|
|
raise NotFoundException(f"Record with id '{record_id}' does not exist.")
|
||
|
|
record_copy = recordschema.BloodPressureRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
db.delete(db_record)
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
return record_copy
|
||
|
|
|
||
|
|
|
||
|
|
def create_blood_oxygen_record(db: Session, record: recordschema.BloodOxygenRecordCreate, device_id: int) -> recordschema.BloodOxygenRecord:
|
||
|
|
"""Creates the specified blood oxygen record in the database."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
db_record = recordmodel.BloodOxygenRecord(
|
||
|
|
value=record.value,
|
||
|
|
measured=record.measured,
|
||
|
|
device_id=device_id,
|
||
|
|
)
|
||
|
|
db.add(db_record)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(db_record)
|
||
|
|
|
||
|
|
return recordschema.BloodOxygenRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
|
||
|
|
def read_blood_oxygen_records_by_device(
|
||
|
|
db: Session, device_id: int,
|
||
|
|
since: datetime | None = None, until: datetime | None = None,
|
||
|
|
skip: int = 0, limit: int = 100,
|
||
|
|
) -> list[recordschema.BloodOxygenRecord]:
|
||
|
|
"""Queries the db for a range of blood oxygen records captured by the device with the specified id."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
filters = [recordmodel.BloodOxygenRecord.device == db_device]
|
||
|
|
if since:
|
||
|
|
filters.append(recordmodel.BloodOxygenRecord.measured >= since)
|
||
|
|
if until:
|
||
|
|
filters.append(recordmodel.BloodOxygenRecord.measured <= until)
|
||
|
|
|
||
|
|
return db.query(recordmodel.BloodOxygenRecord).filter(*filters).order_by(recordmodel.BloodOxygenRecord.measured.desc()).offset(skip).limit(limit).all()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_blood_oxygen_record(db: Session, record_id: int) -> recordschema.BloodOxygenRecord:
|
||
|
|
"""Deletes the blood oxygen record with the provided id from the db."""
|
||
|
|
|
||
|
|
db_record = db.query(recordmodel.BloodOxygenRecord).filter(recordmodel.BloodOxygenRecord.id == record_id).first()
|
||
|
|
if not db_record:
|
||
|
|
raise NotFoundException(f"Record with id '{record_id}' does not exist.")
|
||
|
|
record_copy = recordschema.BloodOxygenRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
db.delete(db_record)
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
return record_copy
|
||
|
|
|
||
|
|
|
||
|
|
def create_respiration_score_record(db: Session, record: recordschema.RespirationScoreRecordCreate, device_id: int) -> recordschema.RespirationScoreRecord:
|
||
|
|
"""Creates the specified respiration score record in the database."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
db_record = recordmodel.RespirationScoreRecord(
|
||
|
|
value=record.value,
|
||
|
|
measured=record.measured,
|
||
|
|
device_id=device_id,
|
||
|
|
)
|
||
|
|
db.add(db_record)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(db_record)
|
||
|
|
|
||
|
|
return recordschema.RespirationScoreRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
|
||
|
|
def read_respiration_score_records_by_device(
|
||
|
|
db: Session, device_id: int,
|
||
|
|
since: datetime | None = None, until: datetime | None = None,
|
||
|
|
skip: int = 0, limit: int = 100,
|
||
|
|
) -> list[recordschema.RespirationScoreRecord]:
|
||
|
|
"""Queries the db for a range of respiration score records captured by the device with the specified id."""
|
||
|
|
|
||
|
|
db_device = db.query(devicemodel.Device).filter(devicemodel.Device.id == device_id).first()
|
||
|
|
if not db_device:
|
||
|
|
raise NotFoundException(f"Device with id '{device_id}' does not exist.")
|
||
|
|
|
||
|
|
filters = [recordmodel.RespirationScoreRecord.device == db_device]
|
||
|
|
if since:
|
||
|
|
filters.append(recordmodel.RespirationScoreRecord.measured >= since)
|
||
|
|
if until:
|
||
|
|
filters.append(recordmodel.RespirationScoreRecord.measured <= until)
|
||
|
|
|
||
|
|
return db.query(recordmodel.RespirationScoreRecord).filter(*filters).order_by(recordmodel.RespirationScoreRecord.measured.desc()).offset(skip).limit(limit).all()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_respiration_score_record(db: Session, record_id: int) -> recordschema.RespirationScoreRecord:
|
||
|
|
"""Deletes the respiration score record with the provided id from the db."""
|
||
|
|
|
||
|
|
db_record = db.query(recordmodel.RespirationScoreRecord).filter(recordmodel.RespirationScoreRecord.id == record_id).first()
|
||
|
|
if not db_record:
|
||
|
|
raise NotFoundException(f"Record with id '{record_id}' does not exist.")
|
||
|
|
record_copy = recordschema.RespirationScoreRecord.from_orm(db_record)
|
||
|
|
|
||
|
|
db.delete(db_record)
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
return record_copy
|