91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
from datetime import datetime
|
|
from backend.db.models.houses import Houses, Areas
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.schemas.house import HouseCreate, HouseUpdate, AreaUpdate
|
|
|
|
class RepositoryHouses:
|
|
def __init__(self):
|
|
self.houses = Houses()
|
|
self.areas = Areas()
|
|
|
|
def get_all(self, skip: int = 0, limit: int = 100):
|
|
return self.houses.query.filter_by(deleted_at=None).offset(skip*limit).limit(limit).all()
|
|
|
|
def get_all_areas(self, house_id: str, skip: int = 0, limit: int = 100):
|
|
return self.areas.query.filter_by(deleted_at=None, house_id=house_id).offset(skip).limit(limit).all()
|
|
|
|
def get_count_all(self):
|
|
return self.houses.query.filter_by(deleted_at=None).count()
|
|
|
|
def get_by_id(self, house_id: str):
|
|
return self.houses.query.filter_by(id=house_id).one()
|
|
|
|
def create(self, db: Session, house: HouseCreate):
|
|
try:
|
|
areas = getattr(house, "areas")
|
|
db_house = Houses(**house.dict(exclude={"areas"}))
|
|
for area in areas:
|
|
db_house.areas.append(Areas(**area.dict()))
|
|
db.add(db_house)
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
db.refresh(db_house)
|
|
return db_house
|
|
|
|
def update_area(self, db: Session, areas: list[AreaUpdate], house_id: str):
|
|
db_house = self.get_by_id(house_id)
|
|
if not db_house:
|
|
return None
|
|
try:
|
|
db_areaId = [area.id for area in db_house.areas]
|
|
area_update_id = [area.id for area in areas if area.id is not None]
|
|
area_delete_id = [area_id for area_id in db_areaId if area_id not in area_update_id]
|
|
if area_delete_id:
|
|
self.areas.query.where(Areas.id.in_(area_delete_id)).delete()
|
|
|
|
for area in areas:
|
|
area_obj = Areas(**area.dict(), house_id=house_id)
|
|
db.merge(area_obj)
|
|
|
|
# areaList = [dict(**area.dict(), house_id=house_id) for area in areas]
|
|
# db.execute(update(Areas), areaList)
|
|
# self.areas.query.update(areaList)
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
|
|
def update(self, db: Session, house: HouseUpdate):
|
|
db_house = self.get_by_id(house.id)
|
|
if not db_house:
|
|
return None
|
|
try:
|
|
self.houses.query.where(Houses.id == house.id).update(house.dict(exclude={"id", 'areas'}, exclude_unset=True, exclude_none=True))
|
|
db.commit()
|
|
self.update_area(db, house.areas, house.id)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
db.refresh(db_house)
|
|
return db_house
|
|
|
|
def delete(self, db: Session, house_id: str):
|
|
db_house = self.get_by_id(house_id)
|
|
if not db_house:
|
|
return None
|
|
try:
|
|
self.houses.query.where(Houses.id == house_id).update({"deleted_at": datetime.utcnow()})
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
db.refresh(db_house)
|
|
return db_house
|