"""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)