forked from s_ranjbar/city_retrofit
93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
"""
|
|
Building repository with database CRUD operations
|
|
SPDX - License - Identifier: LGPL - 3.0 - or -later
|
|
Copyright © 2022 Concordia CERC group
|
|
Project Coder Peter Yefi peteryefi@gmail.com
|
|
"""
|
|
|
|
from city_model_structure.building import Building as CityBuilding
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from persistence.models import Building
|
|
from persistence import BaseRepo
|
|
from helpers.city_util import CityUtil
|
|
from sqlalchemy import select
|
|
|
|
|
|
class BuildingRepo(BaseRepo):
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._city_util = CityUtil()
|
|
|
|
def insert(self, building: CityBuilding, city_id: int) -> Building:
|
|
"""
|
|
Inserts a new building into the database
|
|
:param building: the building to insert
|
|
:param city_id: the city the building belongs to
|
|
:return:
|
|
"""
|
|
|
|
model_building = Building()
|
|
model_building.name = building.name
|
|
model_building.function = building.function
|
|
model_building.construction_year = building.year_of_construction
|
|
model_building.floor_area = building.floor_area
|
|
model_building.city_id = city_id
|
|
model_building.data = self._city_util.extract_building_data(building)
|
|
try:
|
|
self.session.add(model_building)
|
|
self.session.flush()
|
|
self.session.commit()
|
|
return model_building
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while adding building: {err}')
|
|
|
|
def get_by_id(self, building_id: int) -> Building:
|
|
"""
|
|
Fetch a building based on the id
|
|
:param building_id: the building id
|
|
:return: a Building
|
|
"""
|
|
try:
|
|
return self.session.execute(select(Building).where(Building.id == building_id)).first()[0]
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while fetching building: {err}')
|
|
|
|
def get_by_name(self, building_name: str) -> [Building]:
|
|
"""
|
|
Fetch a building based on the name
|
|
:param building_name: the name of the building
|
|
:return: [Building] with the provided name
|
|
"""
|
|
try:
|
|
result_set = self.session.execute(select(Building).where(Building.name == building_name))
|
|
return [building[0] for building in result_set]
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while fetching buildings: {err}')
|
|
|
|
def get_by_construction_year(self, year: int):
|
|
"""
|
|
Fetch a building based on the year of construction
|
|
:param year: the construction year of the building
|
|
:return: [Building]
|
|
"""
|
|
try:
|
|
result_set = self.session.execute(select(Building).where(Building.construction_year == year))
|
|
return [building[0] for building in result_set]
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while fetching buildings: {err}')
|
|
|
|
def get_by_city(self, city_id: int):
|
|
"""
|
|
Get all the buildings in a city
|
|
:param city_id: the city id
|
|
:return: [Building]
|
|
"""
|
|
try:
|
|
result_set = self.session.execute(select(Building).where(Building.city_id == city_id))
|
|
return [building[0] for building in result_set]
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while fetching buildings: {err}')
|
|
|
|
|