87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from models.rent_monitoring import RentMonitoring
|
|
from schemas.rent_monitoring import RentMonitoringCreate, RentMonitoringUpdate
|
|
import uuid
|
|
|
|
def get_all_rent_monitorings(db: Session) -> List[RentMonitoring]:
|
|
"""
|
|
Retrieves all rent monitoring records from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[RentMonitoring]: A list of all rent monitoring objects.
|
|
"""
|
|
return db.query(RentMonitoring).all()
|
|
|
|
def get_rent_monitoring_by_id(db: Session, rent_monitoring_id: uuid.UUID) -> Optional[RentMonitoring]:
|
|
"""
|
|
Retrieves a single rent monitoring record by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
rent_monitoring_id (UUID): The ID of the rent monitoring record to retrieve.
|
|
|
|
Returns:
|
|
Optional[RentMonitoring]: The rent monitoring object if found, otherwise None.
|
|
"""
|
|
return db.query(RentMonitoring).filter(RentMonitoring.id == rent_monitoring_id).first()
|
|
|
|
def create_rent_monitoring(db: Session, rent_monitoring_data: RentMonitoringCreate) -> RentMonitoring:
|
|
"""
|
|
Creates a new rent monitoring record in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
rent_monitoring_data (RentMonitoringCreate): The data for the rent monitoring record to create.
|
|
|
|
Returns:
|
|
RentMonitoring: The newly created rent monitoring object.
|
|
"""
|
|
db_rent_monitoring = RentMonitoring(**rent_monitoring_data.dict())
|
|
db.add(db_rent_monitoring)
|
|
db.commit()
|
|
db.refresh(db_rent_monitoring)
|
|
return db_rent_monitoring
|
|
|
|
def update_rent_monitoring(db: Session, rent_monitoring_id: uuid.UUID, rent_monitoring_data: RentMonitoringUpdate) -> RentMonitoring:
|
|
"""
|
|
Updates an existing rent monitoring record in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
rent_monitoring_id (UUID): The ID of the rent monitoring record to update.
|
|
rent_monitoring_data (RentMonitoringUpdate): The updated data for the rent monitoring record.
|
|
|
|
Returns:
|
|
RentMonitoring: The updated rent monitoring object.
|
|
"""
|
|
db_rent_monitoring = db.query(RentMonitoring).filter(RentMonitoring.id == rent_monitoring_id).first()
|
|
if not db_rent_monitoring:
|
|
raise ValueError(f"Rent monitoring record with ID {rent_monitoring_id} not found.")
|
|
update_data = rent_monitoring_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_rent_monitoring, key, value)
|
|
db.commit()
|
|
db.refresh(db_rent_monitoring)
|
|
return db_rent_monitoring
|
|
|
|
def delete_rent_monitoring(db: Session, rent_monitoring_id: uuid.UUID) -> bool:
|
|
"""
|
|
Deletes an existing rent monitoring record from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
rent_monitoring_id (UUID): The ID of the rent monitoring record to delete.
|
|
|
|
Returns:
|
|
bool: True if the rent monitoring record was successfully deleted, False otherwise.
|
|
"""
|
|
db_rent_monitoring = db.query(RentMonitoring).filter(RentMonitoring.id == rent_monitoring_id).first()
|
|
if not db_rent_monitoring:
|
|
return False
|
|
db.delete(db_rent_monitoring)
|
|
db.commit()
|
|
return True |