city_retrofit/persistence/repositories/building_repo.py

"""
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, db_name):
    super().__init__(db_name)
    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}')