"""Version control
This file provides both project and version managers for interacting with the database.
This file can be imported as a module and contains the following classes:
* ResponseType: Enum for representing the status of a response from a manager.
* ResponseObject: Holds the status and response data of a function from a manager.
* VersionResponse: The data it holds is a Version from the database.
* ProjectResponse: The data it holds is a Project from the database.
* GlobalResponse: The data it holds is a Global from the database.
* ProjectManager: The manager for executing project based functions in the database.
* VersionManager: The manager for executing version based functions in the database.
"""
import re
import logging
from enum import Enum
from typing import List
from sqlalchemy import select
from uppaal2jetracer.versioncontrol.database import DatabaseConnection
from uppaal2jetracer.versioncontrol.models import Project, Version, Global
logger = logging.getLogger("version_control")
[docs]
class ResponseType(Enum):
"""
An Enum for representing the type of response.
"""
OK = 1
ERROR = -1
[docs]
class ResponseObject:
"""
A class to hold a response type and payload.
:ivar _type: The type of the response.
:vartype _type: ResponseType
:ivar _payload: The payload of the response.
:vartype _payload: object
:ivar _message: The message of the response in case of an error.
:vartype _message: str
"""
__slots__ = ("_type", "_payload", "_message")
def __init__(self, r_type: ResponseType, payload: object, message: str = None):
self._type = r_type
self._payload = payload
self._message = message
@property
def type(self) -> ResponseType:
"""
The type of the response. Either 'ok' or 'error'.
"""
return self._type
@property
def payload(self) -> object:
"""
The payload of the response.
"""
return self._payload
@property
def message(self) -> str:
"""
The message of the response in case of an error.
"""
return self._message
[docs]
class VersionResponse(ResponseObject):
"""
A class to hold a response type and Version payload.
"""
def __init__(self, r_type: ResponseType, payload: Version, message: str = None):
super().__init__(r_type, payload, message)
@property
def payload(self) -> Version:
"""
The Version payload of the response.
"""
return self._payload
[docs]
class ProjectResponse(ResponseObject):
"""
A class to hold a response type and Project payload.
"""
def __init__(self, r_type: ResponseType, payload: Project, message: str = None,
versions: List[Version] = None):
super().__init__(r_type, payload, message)
self._versions = versions
@property
def payload(self) -> Project:
"""
The Project payload of the response.
"""
return self._payload
@property
def versions(self) -> List[Version]:
"""
The Versions payload of the response.
"""
return self._versions
[docs]
class GlobalResponse(ResponseObject):
"""
A class to hold a response type and Global payload.
"""
def __init__(self, r_type: ResponseType, payload: Global, message: str = None):
super().__init__(r_type, payload, message)
@property
def payload(self) -> Global:
"""
The Global payload of the response.
"""
return self._payload
[docs]
class ProjectManager:
"""
Class for managing projects in the database.
:ivar _db: The database connection.
:vartype _db: DatabaseConnection
"""
_INVALID_PROJECT_NAME_ERROR = "Project name is invalid."
_DUPLICATE_PROJECT_ERROR = "Project name is already in use."
_NO_CURRENT_PROJECT_ERROR = "No current project exists."
_NO_PROJECT_FOUND_ERROR = "No project was found with the given name."
_BELOW_VALID_LIMIT_ERROR = "Cannot set limit below an existing project's version count."
_INVALID_MAX_VALUE_ERROR = "The given maximum number of versions is invalid."
_VERSION_MAX_NAME = "version_max"
_PROJECT_NAME_FORMAT = "^[A-Za-z0-9_-]*$"
__slots__ = ("_db",)
def __init__(self, db: DatabaseConnection):
self._db = db
logger.info("Initialized Project Manager.")
@property
def db(self) -> DatabaseConnection:
"""
The database connection of the manager.
:return: The database connection of the manager.
:rtype: DatabaseConnection
"""
return self._db
[docs]
def add_project(self, name: str) -> ProjectResponse:
"""
Add a new project to the database.
:param name: Name of the new project.
:type name: str
:return: Response with status and new project or an error.
:rtype: ProjectResponse
"""
if (name is None or len(name) == 0 or len(name) > self.db.config.DB_PROJECT_NAME_MAX
or not re.match(self._PROJECT_NAME_FORMAT, name)):
return ProjectResponse(ResponseType.ERROR, None, self._INVALID_PROJECT_NAME_ERROR)
select_project = select(Project).where(Project._name == name)
dup_project = self.db.session.scalars(select_project).first()
if dup_project is not None:
return ProjectResponse(ResponseType.ERROR, None, self._DUPLICATE_PROJECT_ERROR)
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
project = Project(_name = name, _current_user = g_user)
self.db.session.add(project)
g_user.current_project = project
self.db.session.commit()
success_msg = f"Added new project: {name}."
logger.debug(success_msg)
return ProjectResponse(ResponseType.OK, project, success_msg)
[docs]
def get_current_project(self) -> ProjectResponse:
"""
Get the current project.
:return: Response with status and the current project or an error.
:rtype: ProjectResponse
"""
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
project = g_user.current_project
if project is None:
return ProjectResponse(ResponseType.ERROR, None, self._NO_CURRENT_PROJECT_ERROR)
success_msg = "Fetched current project."
logger.debug(success_msg)
return ProjectResponse(ResponseType.OK, project, success_msg)
[docs]
def get_all_projects(self) -> List[ProjectResponse]:
"""
Get all existing projects.
:return: List of Responses, each with a status and one of the existing projects.
:rtype: List[ProjectResponse]
"""
select_projects = select(Project).order_by(Project._p_id.desc())
projects = self.db.session.scalars(select_projects).all()
res = []
success_msg = "Fetched all projects."
for project in projects:
select_versions = select(Version).where(
Version._p_id == project.p_id
).order_by(Version._v_id.desc())
versions = self.db.session.scalars(select_versions).all()
res.append(ProjectResponse(ResponseType.OK, project, success_msg, versions))
logger.debug(success_msg)
return res
[docs]
def get_config(self) -> GlobalResponse:
"""
Get all user configuration settings.
:return: Response with status and the configurations.
:rtype: GlobalResponse
"""
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
success_msg = "Fetched the config."
logger.debug(success_msg)
return GlobalResponse(ResponseType.OK, g_user, success_msg)
[docs]
def delete_project(self, name: str) -> ProjectResponse:
"""
Delete an existing project.
:param name: Name of the project you wish to delete.
:type name: str
:return: Response with status and the deleted project or an error.
:rtype: ProjectResponse
"""
select_project = select(Project).where(Project._name == name)
project = self.db.session.scalars(select_project).first()
if project is None:
return ProjectResponse(ResponseType.ERROR, None, self._NO_PROJECT_FOUND_ERROR)
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
if g_user.current_project is not None and name is g_user.current_project.name:
g_user.current_project = None
select_versions = project.versions.select()
versions = self.db.session.scalars(select_versions).all()
for version in versions:
self.db.session.delete(version)
self.db.session.delete(project)
self.db.session.commit()
success_msg = f"Deleted project: {name}."
logger.debug(success_msg)
return ProjectResponse(ResponseType.OK, project, success_msg)
[docs]
def select_project(self, name: str) -> ProjectResponse:
"""
Select a project to be the new current project.
:param name: Name of the project you wish to select.
:type name: str
:return: Response with status and the new current project or an error.
:rtype: ProjectResponse
"""
select_project = select(Project).where(Project._name == name)
project = self.db.session.scalars(select_project).first()
if project is None:
return ProjectResponse(ResponseType.ERROR, None, self._NO_PROJECT_FOUND_ERROR)
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
g_user.current_project = project
self.db.session.commit()
success_msg = f"Selected project: {name}."
logger.debug(success_msg)
return ProjectResponse(ResponseType.OK, project, success_msg)
[docs]
def change_version_limit(self, num: int) -> GlobalResponse:
"""
Set a new version limit.
:param num: The number the limit should be set to.
:type num: int
:return: Response with status and the updated global model or an error.
:rtype: GlobalResponse
"""
if self.db.config.VERSION_MAX_LOW <= num <= self.db.config.VERSION_MAX_HIGH:
select_projects = select(Project)
projects = self.db.session.scalars(select_projects).all()
for project in projects:
versions = self.db.session.scalars(project.versions.select()).all()
if len(versions) > num:
return GlobalResponse(ResponseType.ERROR, None, self._BELOW_VALID_LIMIT_ERROR)
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
g_user.version_max = num
self.db.session.commit()
success_msg = f"Changed version limit to: {num}."
logger.debug(success_msg)
return GlobalResponse(ResponseType.OK, g_user, success_msg)
return GlobalResponse(ResponseType.ERROR, None, self._INVALID_MAX_VALUE_ERROR)
[docs]
class VersionManager:
"""
Class for managing projects in the database.
:ivar _db: The database connection.
:vartype _db: DatabaseConnection
"""
_INVALID_VERSION_NAME_ERROR = "Version name is invalid."
_NOT_IN_CURRENT_PROJECT_ERROR = "Not in a project currently."
_MAX_VERSIONS_REACHED = "Maximum limit of favorited versions has been reached."
_NO_VERSION_FOUND_ERROR = "No version can be found."
_NO_PROJECT_FOUND_ERROR = "No project can be found."
_FILE_NAME = "file"
_VERSION_NAME_FORMAT = "^[A-Za-z0-9_-]*$"
__slots__ = ("_db",)
def __init__(self, db: DatabaseConnection):
self._db = db
logger.info("Initialized Version Manager.")
@property
def db(self) -> DatabaseConnection:
"""
The database connection of the manager.
:return: The database connection of the manager.
:rtype: DatabaseConnection
"""
return self._db
[docs]
def add_version(self, name: str, data: object) -> List[VersionResponse]:
"""
Add a new version to the database.
:param name: Name of the new version.
:type name: str
:param data: File data of the version.
:type data: object
:return: Response with status and the new version
and possibly the deleted version or an error.
:rtype: List[VersionResponse]
"""
if (name is None or len(name) == 0 or len(name) > self.db.config.DB_VERSION_NAME_MAX
or not re.match(self._VERSION_NAME_FORMAT, name)):
return [VersionResponse(ResponseType.ERROR, None, self._INVALID_VERSION_NAME_ERROR)]
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
curr_project = g_user.current_project
if curr_project is None:
return [VersionResponse(ResponseType.ERROR, None, self._NOT_IN_CURRENT_PROJECT_ERROR) ]
versions = self.db.session.scalars(curr_project.versions.select().order_by(
Version._v_id.asc()
)).all()
deleted_version = None
if len(versions) is g_user.version_max:
version_deleted = False
for version in versions:
if not version.favorite:
deleted_version = self.delete_version(version.v_id)
logger.debug("Deleted version %s.", deleted_version.payload.name)
version_deleted = True
break
if not version_deleted:
return [VersionResponse(ResponseType.ERROR, None, self._MAX_VERSIONS_REACHED)]
version = Version(_name = name, _file = data,
_p_id = curr_project.p_id, _project = curr_project)
self.db.session.add(version)
curr_project.add_version(version)
self.db.session.commit()
deleted_payload = None
if deleted_version is not None:
deleted_payload = deleted_version.payload
success_msg = f"Added version: {name}."
logger.debug(success_msg)
return [VersionResponse(ResponseType.OK, version, success_msg),
VersionResponse(ResponseType.OK, deleted_payload)]
[docs]
def get_all_versions(self, p_id: int = None) -> List[VersionResponse]:
"""
Get all existing versions.
:param p_id: ID of a project you wish to retrieve the versions from, defaults to None.
:type p_id: int, optional (retrieves versions of the current project if no id was provided)
:return: List of Responses, each with a status and one of the existing versions or an error.
:rtype: List[VersionResponse]
"""
res = []
if p_id is None:
select_user = select(Global).where(Global._g_id == self.db.config.GLOBAL_ID)
g_user = self.db.session.scalars(select_user).first()
curr_project = g_user.current_project
if curr_project is None:
return [VersionResponse(ResponseType.ERROR, None,
self._NOT_IN_CURRENT_PROJECT_ERROR)]
curr_versions = self.db.session.scalars(curr_project.versions.select().order_by(
Version._v_id.desc()
)).all()
for curr_version in curr_versions:
res.append(VersionResponse(ResponseType.OK, curr_version))
return res
select_project = select(Project).where(Project._p_id == p_id)
project = self.db.session.scalars(select_project).first()
if project is None:
return [VersionResponse(ResponseType.ERROR, None, self._NO_PROJECT_FOUND_ERROR)]
versions = self.db.session.scalars(project.versions.select().order_by(
Version._v_id.desc()
)).all()
success_msg = "Fetched all versions."
for version in versions:
res.append(VersionResponse(ResponseType.OK, version, success_msg))
logger.debug(success_msg)
return res
[docs]
def get_version(self, v_id: int) -> VersionResponse:
"""
Get a specific version.
:param v_id: ID of the version you wish to retrieve.
:type v_id: int
:return: Response with status and the requested version or an error.
:rtype: VersionResponse
"""
select_version = select(Version).where(Version._v_id == v_id)
version = self.db.session.scalars(select_version).first()
if version is None:
return VersionResponse(ResponseType.ERROR, None, self._NO_VERSION_FOUND_ERROR)
success_msg = f"Fetched version with id: {v_id}."
logger.debug(success_msg)
return VersionResponse(ResponseType.OK, version, success_msg)
[docs]
def delete_version(self, v_id: int) -> VersionResponse:
"""
Delete a specific version.
:param v_id: ID of the version you wish to delete.
:type v_id: int
:return: Response with status and the deleted version or an error.
:rtype: VersionResponse
"""
select_version = select(Version).where(Version._v_id == v_id)
version = self.db.session.scalars(select_version).first()
if version is None:
return VersionResponse(ResponseType.ERROR, None, self._NO_VERSION_FOUND_ERROR)
project = version.project
project.remove_version(version)
self.db.session.delete(version)
self.db.session.commit()
success_msg = f"Deleted version with id: {v_id}."
logger.debug(success_msg)
return VersionResponse(ResponseType.OK, version, success_msg)
[docs]
def favorite_version(self, v_id: int) -> VersionResponse:
"""
Favorite a specific version.
:param v_id: ID of the version you wish to favorite.
:type v_id: int
:return: Response with status and the updated version or an error.
:rtype: VersionResponse
"""
select_version = select(Version).where(Version._v_id == v_id)
version = self.db.session.scalars(select_version).first()
if version is None:
return VersionResponse(ResponseType.ERROR, None, self._NO_VERSION_FOUND_ERROR)
res_message_favorite = "Favorited"
if version.favorite:
res_message_favorite = "Unfavorited"
version.toggle_favorite()
self.db.session.commit()
success_msg = f"{res_message_favorite} version with id: {v_id}."
logger.debug(success_msg)
return VersionResponse(ResponseType.OK, version, success_msg)