forked from s_ranjbar/city_retrofit
Reintroduce function in business logic
This commit is contained in:
parent
fc25c12149
commit
a3a382dae9
|
@ -5,6 +5,7 @@ Copyright © 2022 Concordia CERC group
|
||||||
Project CoderPeter Yefi peteryefi@gmail.com
|
Project CoderPeter Yefi peteryefi@gmail.com
|
||||||
"""
|
"""
|
||||||
import json
|
import json
|
||||||
|
from typing import Union, Dict
|
||||||
|
|
||||||
from hub.persistence import City
|
from hub.persistence import City
|
||||||
from hub.persistence import Application
|
from hub.persistence import Application
|
||||||
|
@ -25,10 +26,11 @@ class DBFactory:
|
||||||
self._city_object = CityObject(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
|
self._city_object = CityObject(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
|
||||||
self._simulation_results = SimulationResults(db_name=db_name, dotenv_path=dotenv_path, app_env=app_env)
|
self._simulation_results = SimulationResults(db_name=db_name, dotenv_path=dotenv_path, app_env=app_env)
|
||||||
|
|
||||||
def application_info(self, application_uuid):
|
def application_info(self, application_uuid) -> Union[Application, None]:
|
||||||
"""
|
"""
|
||||||
Retrieve the application info for the given uuid
|
Retrieve the application info for the given uuid
|
||||||
:param application_uuid: the uuid for the application
|
:param application_uuid: the uuid for the application
|
||||||
|
:return: Application or None
|
||||||
"""
|
"""
|
||||||
return self._application.get_by_uuid(application_uuid)
|
return self._application.get_by_uuid(application_uuid)
|
||||||
|
|
||||||
|
@ -38,6 +40,7 @@ class DBFactory:
|
||||||
:param name: the user name
|
:param name: the user name
|
||||||
:param password: the user password
|
:param password: the user password
|
||||||
:param application_id: the application id
|
:param application_id: the application id
|
||||||
|
:return: User or None
|
||||||
"""
|
"""
|
||||||
return self._user.get_by_name_application_id_and_password(name, password, application_id)
|
return self._user.get_by_name_application_id_and_password(name, password, application_id)
|
||||||
|
|
||||||
|
@ -47,13 +50,29 @@ class DBFactory:
|
||||||
:param name: the user name
|
:param name: the user name
|
||||||
:param password: the user password
|
:param password: the user password
|
||||||
:param application_uuid: the application uuid
|
:param application_uuid: the application uuid
|
||||||
|
:return: User or None
|
||||||
"""
|
"""
|
||||||
return self._user.get_by_name_application_uuid_and_password(name, password, application_uuid)
|
return self._user.get_by_name_application_uuid_and_password(name, password, application_uuid)
|
||||||
|
|
||||||
def cities_by_user_and_application(self, user_id, application_id) -> [City]:
|
def cities_by_user_and_application(self, user_id, application_id) -> [City]:
|
||||||
|
"""
|
||||||
|
Retrieve the cities belonging to the user and the application
|
||||||
|
:param user_id: User id
|
||||||
|
:param application_id: Application id
|
||||||
|
:return: [City]
|
||||||
|
"""
|
||||||
return self._city.get_by_user_id_and_application_id(user_id, application_id)
|
return self._city.get_by_user_id_and_application_id(user_id, application_id)
|
||||||
|
|
||||||
def results(self, user_id, application_id, cities, result_names=[]):
|
def building_info(self, name, city_id) -> Union[CityObject, None]:
|
||||||
|
"""
|
||||||
|
Retrieve the building info
|
||||||
|
:param name: Building name
|
||||||
|
:param city_id: City Id
|
||||||
|
:return: CityObject or None
|
||||||
|
"""
|
||||||
|
return self._city_object.get_by_name_and_city(name, city_id)
|
||||||
|
|
||||||
|
def results(self, user_id, application_id, cities, result_names=None) -> Dict:
|
||||||
"""
|
"""
|
||||||
Retrieve the simulation results for the given cities
|
Retrieve the simulation results for the given cities
|
||||||
:param user_id: the user id owning the results
|
:param user_id: the user id owning the results
|
||||||
|
@ -61,6 +80,8 @@ class DBFactory:
|
||||||
:param cities: dictionary containing the city and building names for the results
|
:param cities: dictionary containing the city and building names for the results
|
||||||
:param result_names: if given, filter the results to the selected names
|
:param result_names: if given, filter the results to the selected names
|
||||||
"""
|
"""
|
||||||
|
if result_names is None:
|
||||||
|
result_names = []
|
||||||
results = {}
|
results = {}
|
||||||
for city in cities['cities']:
|
for city in cities['cities']:
|
||||||
city_name = next(iter(city))
|
city_name = next(iter(city))
|
||||||
|
@ -83,6 +104,3 @@ class DBFactory:
|
||||||
values["building"] = building_name
|
values["building"] = building_name
|
||||||
results[city_name].append(values)
|
results[city_name].append(values)
|
||||||
return results
|
return results
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -105,7 +105,7 @@ class CityObject(Repository):
|
||||||
except SQLAlchemyError as err:
|
except SQLAlchemyError as err:
|
||||||
logger.error(f'Error while deleting application: {err}')
|
logger.error(f'Error while deleting application: {err}')
|
||||||
|
|
||||||
def get_by_name_and_city(self, name, city_id) -> [Model]:
|
def get_by_name_and_city(self, name, city_id) -> Union[Model, None]:
|
||||||
"""
|
"""
|
||||||
Fetch a city object based on name and city id
|
Fetch a city object based on name and city id
|
||||||
:param name: city object name
|
:param name: city object name
|
||||||
|
|
|
@ -60,7 +60,7 @@ class User(Repository):
|
||||||
:param name: the name of the user
|
:param name: the name of the user
|
||||||
:param password: the password of the user
|
:param password: the password of the user
|
||||||
:param role: the role of the user
|
:param role: the role of the user
|
||||||
:return:
|
:return: None, Dictionary
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
self.session.query(Model).filter(Model.id == user_id).update({
|
self.session.query(Model).filter(Model.id == user_id).update({
|
||||||
|
@ -86,7 +86,7 @@ class User(Repository):
|
||||||
except SQLAlchemyError as err:
|
except SQLAlchemyError as err:
|
||||||
logger.error(f'Error while fetching user: {err}')
|
logger.error(f'Error while fetching user: {err}')
|
||||||
|
|
||||||
def get_by_name_and_application(self, name: str, application_id: int) -> [Model]:
|
def get_by_name_and_application(self, name: str, application_id: int) -> Model:
|
||||||
"""
|
"""
|
||||||
Fetch user based on the email address
|
Fetch user based on the email address
|
||||||
:param name: User name
|
:param name: User name
|
||||||
|
@ -96,18 +96,17 @@ class User(Repository):
|
||||||
try:
|
try:
|
||||||
return self.session.execute(
|
return self.session.execute(
|
||||||
select(Model).where(Model.name == name, Model.application_id == application_id)
|
select(Model).where(Model.name == name, Model.application_id == application_id)
|
||||||
).first()
|
).first()[0]
|
||||||
except SQLAlchemyError as err:
|
except SQLAlchemyError as err:
|
||||||
logger.error(f'Error while fetching user by name and application: {err}')
|
logger.error(f'Error while fetching user by name and application: {err}')
|
||||||
|
|
||||||
def get_by_name_application_id_and_password(self, name: str, password: str, application_id: int) -> [Model]:
|
def get_by_name_application_id_and_password(self, name: str, password: str, application_id: int) -> Union[Model, None]:
|
||||||
"""
|
"""
|
||||||
Fetch user based on the email and password
|
Fetch user based on the email and password
|
||||||
:param name: User name
|
:param name: User name
|
||||||
:param password: User password
|
:param password: User password
|
||||||
:param application_id: User password
|
:param application_id: User password
|
||||||
|
:return: User
|
||||||
:return: [User]
|
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
user = self.session.execute(
|
user = self.session.execute(
|
||||||
|
@ -119,14 +118,13 @@ class User(Repository):
|
||||||
except SQLAlchemyError as err:
|
except SQLAlchemyError as err:
|
||||||
logger.error(f'Error while fetching user by email: {err}')
|
logger.error(f'Error while fetching user by email: {err}')
|
||||||
|
|
||||||
def get_by_name_application_uuid_and_password(self, name: str, password: str, application_uuid: str) -> [Model]:
|
def get_by_name_application_uuid_and_password(self, name: str, password: str, application_uuid: str) -> Union[Model, None]:
|
||||||
"""
|
"""
|
||||||
Fetch user based on the email and password
|
Fetch user based on the email and password
|
||||||
:param name: User name
|
:param name: User name
|
||||||
:param password: User password
|
:param password: User password
|
||||||
:param application_uuid: Application uuid
|
:param application_uuid: Application uuid
|
||||||
|
:return: User
|
||||||
:return: [User]
|
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
application = self.session.execute(
|
application = self.session.execute(
|
||||||
|
|
Loading…
Reference in New Issue
Block a user