Skip to content

BaseDB

hololinked.storage.bases.BaseDB

Bases: BaseConfigurationRepository

Base class for database engine based configuration management.

Implements common functionality like loading DB configuration and context management for batch calls. Specific database engines should inherit from this class and implement the required methods from BaseConfigurationRepository.

Source code in repo/hololinked/hololinked/storage/bases.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class BaseDB(BaseConfigurationRepository):
    """
    Base class for database engine based configuration management.

    Implements common functionality like loading DB configuration and context management for batch calls.
    Specific database engines should inherit from this class and implement the required methods from
    `BaseConfigurationRepository`.
    """

    def __init__(self, thing: Thing, config_file: str | None = None) -> None:
        """
        Initialize the BaseDB instance.

        Parameters
        ----------
        thing: Thing
            The `Thing` instance which uses this database engine for configuration storage.
        config_file: str, optional
            Path to the database configuration file. `sqlite` backend with default settings will be used if not provided.
        """
        super().__init__(thing=thing)
        self.config = BaseDB.load_conf(
            config_file=config_file,
            default_file_path=os.path.join(
                global_config.TEMP_DIR_DB,
                get_sanitized_filename_from_random_string(
                    f"{self.thing.__class__.__name__}.{self.thing.id}",
                    extension="db",
                ),
            ),
        )
        self._batch_call_context = {}

    @classmethod
    def from_thing(cls, thing: Thing, **kwargs: Any) -> Self:
        """
        Build the database engine for a `Thing`, taking its configuration from the `db_config_file` argument.

        Parameters
        ----------
        thing: Thing
            the `Thing` instance whose configuration is stored in this database
        kwargs: dict[str, Any]
            the keyword arguments given to the `Thing`; `db_config_file` is read from here and the rest ignored

        Returns
        -------
        Self
            the database engine, connected and ready to use
        """
        return cls(thing=thing, config_file=kwargs.get("db_config_file"))

    @staticmethod
    def load_conf(
        config_file: str | None,
        default_file_path: str = "",
    ) -> SQLDBConfig | SQLiteConfig | MongoDBConfig:
        """
        Load configuration file using JSON serializer.

        Parameters
        ----------
        config_file: str
            path to configuration file, expected to be in JSON format with fields according to the config classes.
        default_file_path: str
            fallback file path if config_file is not provided, only used for SQLiteConfig, default is empty string
            which leads to a DB with name of thing ID

        Returns
        -------
        SQLDBConfig | SQLiteConfig | MongoDBConfig
            configuration object according to the provider specified in the config file

        Raises
        ------
        NotImplementedError
            if the provider specified in the config file is not supported
        ValueError
            if the config file is not in JSON format
        """
        if not config_file:
            return SQLiteConfig(file=default_file_path)
        if not config_file.endswith(".json"):
            raise ValueError("config files of extension {} expected, given file name {}".format(["json"], config_file))
        file = open(config_file, "r")
        conf = JSONSerializer.load(file)
        if not isinstance(conf, dict):
            raise ValueError("config file expected to contain a JSON object/dictionary, given {}".format(type(conf)))
        if conf.get("provider", None) in ["postgresql", "mysql"]:
            return SQLDBConfig.model_validate(conf, strict=True, from_attributes=True)
        elif conf.get("provider", None) == "sqlite":
            return SQLiteConfig.model_validate(conf, strict=True, from_attributes=True)
        elif conf.get("provider", None) == "mongo":
            return MongoDBConfig.model_validate(conf, strict=True, from_attributes=True)
        raise NotImplementedError("only postgresql, mysql, sqlite and mongo are supported")

    @property
    def in_batch_call_context(self):
        return threading.get_ident() in self._batch_call_context

Attributes

config instance-attribute 

config = load_conf(config_file=config_file, default_file_path=join(TEMP_DIR_DB, get_sanitized_filename_from_random_string(f'{__name__}.{id}', extension='db')))

in_batch_call_context property 

in_batch_call_context

Functions

__init__ 

__init__(thing: Thing, config_file: str | None = None) -> None

Initialize the BaseDB instance.

Parameters:

Name Type Description Default
thing
Thing

The Thing instance which uses this database engine for configuration storage.

 required
config_file
str | None

Path to the database configuration file. sqlite backend with default settings will be used if not provided.

 None
Source code in repo/hololinked/hololinked/storage/bases.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(self, thing: Thing, config_file: str | None = None) -> None:
    """
    Initialize the BaseDB instance.

    Parameters
    ----------
    thing: Thing
        The `Thing` instance which uses this database engine for configuration storage.
    config_file: str, optional
        Path to the database configuration file. `sqlite` backend with default settings will be used if not provided.
    """
    super().__init__(thing=thing)
    self.config = BaseDB.load_conf(
        config_file=config_file,
        default_file_path=os.path.join(
            global_config.TEMP_DIR_DB,
            get_sanitized_filename_from_random_string(
                f"{self.thing.__class__.__name__}.{self.thing.id}",
                extension="db",
            ),
        ),
    )
    self._batch_call_context = {}

from_thing classmethod 

from_thing(thing: Thing, **kwargs: Any) -> Self

Build the database engine for a Thing, taking its configuration from the db_config_file argument.

Parameters:

Name Type Description Default
thing
Thing

the Thing instance whose configuration is stored in this database

 required
kwargs
Any

the keyword arguments given to the Thing; db_config_file is read from here and the rest ignored

 {}

Returns:

Type Description
Self

the database engine, connected and ready to use
Source code in repo/hololinked/hololinked/storage/bases.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def from_thing(cls, thing: Thing, **kwargs: Any) -> Self:
    """
    Build the database engine for a `Thing`, taking its configuration from the `db_config_file` argument.

    Parameters
    ----------
    thing: Thing
        the `Thing` instance whose configuration is stored in this database
    kwargs: dict[str, Any]
        the keyword arguments given to the `Thing`; `db_config_file` is read from here and the rest ignored

    Returns
    -------
    Self
        the database engine, connected and ready to use
    """
    return cls(thing=thing, config_file=kwargs.get("db_config_file"))

load_conf staticmethod 

load_conf(config_file: str | None, default_file_path: str = '') -> SQLDBConfig | SQLiteConfig | MongoDBConfig

Load configuration file using JSON serializer.

Parameters:

Name Type Description Default
config_file
str | None

path to configuration file, expected to be in JSON format with fields according to the config classes.

 required
default_file_path
str

fallback file path if config_file is not provided, only used for SQLiteConfig, default is empty string which leads to a DB with name of thing ID

 ''

Returns:

Type Description
SQLDBConfig | SQLiteConfig | MongoDBConfig

configuration object according to the provider specified in the config file

Raises:

Type Description
NotImplementedError

if the provider specified in the config file is not supported
ValueError

if the config file is not in JSON format
Source code in repo/hololinked/hololinked/storage/bases.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@staticmethod
def load_conf(
    config_file: str | None,
    default_file_path: str = "",
) -> SQLDBConfig | SQLiteConfig | MongoDBConfig:
    """
    Load configuration file using JSON serializer.

    Parameters
    ----------
    config_file: str
        path to configuration file, expected to be in JSON format with fields according to the config classes.
    default_file_path: str
        fallback file path if config_file is not provided, only used for SQLiteConfig, default is empty string
        which leads to a DB with name of thing ID

    Returns
    -------
    SQLDBConfig | SQLiteConfig | MongoDBConfig
        configuration object according to the provider specified in the config file

    Raises
    ------
    NotImplementedError
        if the provider specified in the config file is not supported
    ValueError
        if the config file is not in JSON format
    """
    if not config_file:
        return SQLiteConfig(file=default_file_path)
    if not config_file.endswith(".json"):
        raise ValueError("config files of extension {} expected, given file name {}".format(["json"], config_file))
    file = open(config_file, "r")
    conf = JSONSerializer.load(file)
    if not isinstance(conf, dict):
        raise ValueError("config file expected to contain a JSON object/dictionary, given {}".format(type(conf)))
    if conf.get("provider", None) in ["postgresql", "mysql"]:
        return SQLDBConfig.model_validate(conf, strict=True, from_attributes=True)
    elif conf.get("provider", None) == "sqlite":
        return SQLiteConfig.model_validate(conf, strict=True, from_attributes=True)
    elif conf.get("provider", None) == "mongo":
        return MongoDBConfig.model_validate(conf, strict=True, from_attributes=True)
    raise NotImplementedError("only postgresql, mysql, sqlite and mongo are supported")