85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
|
"""
|
||
|
|
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
|
||
|
|
from hub.persistence import BaseRepo
|
||
|
|
from hub.persistence.models import Application
|
||
|
|
|
||
|
|
|
||
|
|
class ApplicationRepo(BaseRepo):
|
||
|
|
_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:
|
||
|
|
cls._instance = super(ApplicationRepo, cls).__new__(cls)
|
||
|
|
return cls._instance
|
||
|
|
|
||
|
|
def insert(self, name: str, description: str, application_uuid: str) -> Union[Application, Dict]:
|
||
|
|
application = self.get_by_uuid(application_uuid)
|
||
|
|
if application is None:
|
||
|
|
try:
|
||
|
|
application = Application(name=name, description=description, application_uuid=application_uuid)
|
||
|
|
self.session.add(application)
|
||
|
|
self.session.flush()
|
||
|
|
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:
|
||
|
|
self.session.query(Application).filter(Application.application_uuid == application_uuid) \
|
||
|
|
.update({'name': name, 'description': description, 'updated': datetime.datetime.utcnow()})
|
||
|
|
self.session.commit()
|
||
|
|
except SQLAlchemyError as err:
|
||
|
|
logger.error(f'Error while updating application: {err}')
|
||
|
|
return {'err_msg': 'Error occurred while updating application'}
|
||
|
|
|
||
|
|
def get_by_uuid(self, application_uuid: str) -> [Application]:
|
||
|
|
"""
|
||
|
|
Fetch Application based on the application uuid
|
||
|
|
:param application_uuid: the application uuid
|
||
|
|
:return: [Application] with the provided application_uuid
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
return self.session.execute(select(Application).where(Application.application_uuid == application_uuid)).first()
|
||
|
|
except SQLAlchemyError as err:
|
||
|
|
logger.error(f'Error while fetching application by application_uuid: {err}')
|
||
|
|
|
||
|
|
def delete_application(self, application_uuid: str):
|
||
|
|
"""
|
||
|
|
Deletes an application with the application_uuid
|
||
|
|
:param application_uuid: The application uuid
|
||
|
|
:return: None
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
self.session.query(Application).filter(Application.application_uuid == application_uuid).delete()
|
||
|
|
self.session.commit()
|
||
|
|
except SQLAlchemyError as err:
|
||
|
|
logger.error(f'Error while deleting application: {err}')
|