Source code for uppaal2jetracer.versioncontrol.database

"""Database

This file provides the connection to a database used by the version control.

This file can be imported as a module and contains the following classes:

    * DatabaseConnection: Class for creating and managing a database connection.
"""

import os
import re

import sqlalchemy as db
from sqlalchemy.orm import Session
from sqlalchemy import select, Engine

from uppaal2jetracer.versioncontrol.models import Base, Global
from uppaal2jetracer.versioncontrol.config import Config


[docs] class DatabaseConnection: """ A class for holding a database connection. :ivar _abs_path: The absolute path of the database. :vartype _abs_path: str :ivar _engine: The database engine. :vartype _engine: Engine :ivar _config: The config for the database. :vartype _config: Config :ivar _session: The session used for database queries. :vartype _session: Session """ _DB_PATH_NAME_DEFAULT = "u2j" _DB_PATH_SEPERATOR = os.sep _DB_FILE_TYPE = ".db" __slots__ = ("_abs_path", "_engine", "_config", "_session") def __init__(self, relative_path: str = None, name: str = None): self._abs_path = None self._engine = None self._config = None self._session = None self.init_db(relative_path, name) @property def abs_path(self) -> str: """ Get the absolute path to the database file. :return: The absolute path of the database. :rtype: str """ return self._abs_path @property def engine(self) -> Engine: """ Get the engine of the database connection. :return: The engine of the database. :rtype: Engine """ return self._engine @property def config(self) -> Config: """ Get the config of the database. :return: The config of the database. :rtype: Config """ return self._config @property def session(self) -> Session: """ Get the session of the database connection. :return: The session of the database. :rtype: Session """ return self._session
[docs] def init_db(self, relative_path: str, name: str): """ Initialize the database. :param relative_path: Relative Path of the database file. :type relative_path: str :param name: Name of the database file. :type name: str """ db_uri = "sqlite:///" db_path = "" if self.check_valid_path(relative_path): db_path += relative_path + self._DB_PATH_SEPERATOR if self.check_valid_name(name): db_path += name else: db_path += self._DB_PATH_NAME_DEFAULT db_path += self._DB_FILE_TYPE self._abs_path = os.getcwd() + self._DB_PATH_SEPERATOR + db_path db_uri += self.abs_path db_created = os.path.isfile(self.abs_path) self._engine = db.create_engine(db_uri) self._config = Config() if not db_created: self.create_tables() self._session = Session(self.engine) select_user = select(Global).where(Global._g_id == self.config.GLOBAL_ID) g_user = self.session.scalars(select_user).first() if g_user is None: g_user = Global(_g_id = self.config.GLOBAL_ID, _version_max = self.config.VERSION_MAX_HIGH, _c_id = -1) self.session.add(g_user) self.session.commit()
[docs] def remove_db(self): """ Remove database file if it exists. """ if self.abs_path is None: return if os.path.isfile(self.abs_path): os.remove(self.abs_path)
[docs] def check_valid_path(self, path: str) -> bool: """ Check if relative path exists. :param path: Given relative path. :type path: str :return: Whether the path exists or not. :rtype: bool """ return path is not None and os.path.isdir(os.getcwd() + os.sep + path)
[docs] def check_valid_name(self, name: str) -> bool: """ Check if file name is valid. :param name: Name of the file. :type name: str :return: Whether the file name is valid or not. :rtype: bool """ if name is None or len(name) == 0: return False valid_format = "^[A-Za-z0-9_-]*$" file_splitter = "." file_segments = name.split(file_splitter) file_name = file_segments[0] return bool(re.match(valid_format, file_name))
[docs] def create_tables(self): """ Add the database tables to the database. """ Base.metadata.create_all(self.engine)