city_retrofit/hub/persistence/repositories/application.py

100 lines
3.4 KiB
Python
Raw Normal View History

2023-01-31 13:11:39 -05:00
"""
Application repository with database CRUD operations
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Guille Gutierrez Guillermo.GutierrezMorote@concordia.ca
"""
import datetime
from typing import Union, Dict
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from hub.hub_logger import logger
2023-02-01 06:05:12 -05:00
from hub.persistence import Repository
from hub.persistence.models import Application as Model
2023-01-31 13:11:39 -05:00
2023-02-01 06:05:12 -05:00
class Application(Repository):
2023-01-31 13:11:39 -05:00
_instance = None
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def __new__(cls, db_name, dotenv_path, app_env):
"""
Implemented for a singleton pattern
"""
if cls._instance is None:
2023-02-01 06:05:12 -05:00
cls._instance = super(Application, cls).__new__(cls)
2023-01-31 13:11:39 -05:00
return cls._instance
2023-02-01 06:05:12 -05:00
def insert(self, name: str, description: str, application_uuid: str) -> Union[Model, Dict]:
"""
Inserts a new application
:param name: Application name
:param description: Application description
:param application_uuid: Unique identifier for the application
:return: application and dictionary
"""
2023-01-31 13:11:39 -05:00
application = self.get_by_uuid(application_uuid)
if application is None:
try:
2023-02-01 06:05:12 -05:00
application = Model(name=name, description=description, application_uuid=application_uuid)
2023-02-22 22:45:33 -05:00
2023-01-31 13:11:39 -05:00
self.session.add(application)
self.session.commit()
return application
except SQLAlchemyError as err:
logger.error(f'An error occurred while creating application: {err}')
else:
return {'message': f'An application with {application_uuid} application uuid, already exists'}
def update(self, application_uuid: str, name: str, description: str) -> Union[Dict, None]:
"""
Updates an application
:param application_uuid: the application uuid of the application to be updated
:param name: the application name
:param description: the application description
:return:
"""
try:
2023-02-01 06:05:12 -05:00
self.session.query(Model).filter(
Model.application_uuid == application_uuid
).update({'name': name, 'description': description, 'updated': datetime.datetime.utcnow()})
2023-01-31 13:11:39 -05:00
self.session.commit()
except SQLAlchemyError as err:
logger.error(f'Error while updating application: {err}')
return {'message': 'Error occurred while updating application'}
def delete(self, application_uuid: str):
"""
Deletes an application with the application_uuid
:param application_uuid: The application uuid
:return: None
"""
try:
self.session.query(Model).filter(Model.application_uuid == application_uuid).delete()
self.session.flush()
self.session.commit()
except SQLAlchemyError as err:
logger.error(f'Error while deleting application: {err}')
2023-01-31 13:11:39 -05:00
def get_by_uuid(self, application_uuid: str) -> Union[Model, None]:
2023-01-31 13:11:39 -05:00
"""
Fetch Application based on the application uuid
:param application_uuid: the application uuid
:return: Application with the provided application_uuid or None
2023-01-31 13:11:39 -05:00
"""
try:
result_set = self.session.execute(select(Model).where(
2023-02-01 06:05:12 -05:00
Model.application_uuid == application_uuid)
).first()
if result_set is None:
return None
return result_set[0]
2023-01-31 13:11:39 -05:00
except SQLAlchemyError as err:
logger.error(f'Error while fetching application by application_uuid: {err}')