Classes

class Sqlite__Database (**kwargs)

A mixin class to strictly assign keyword arguments to pre-defined instance attributes during initialization.

This base class provides an init method that assigns values from keyword arguments to instance attributes. If an attribute with the same name as a key from the kwargs is defined in the class, it will be set to the value from kwargs. If the key does not match any predefined attribute names, an exception is raised.

This behavior enforces strict control over the attributes of instances, ensuring that only predefined attributes can be set at the time of instantiation and avoids silent attribute creation which can lead to bugs in the code.

Usage

class MyConfigurableClass(Kwargs_To_Self): attribute1 = 'default_value' attribute2 = True attribute3 : str attribute4 : list attribute4 : int = 42

# Other methods can be added here

Correctly override default values by passing keyword arguments

instance = MyConfigurableClass(attribute1='new_value', attribute2=False)

This will raise an exception as 'attribute3' is not predefined

instance = MyConfigurableClass(attribute3='invalid_attribute')

this will also assign the default value to any variable that has a type defined. In the example above the default values (mapped by default__kwargs and locals) will be: attribute1 = 'default_value' attribute2 = True attribute3 = '' # default value of str attribute4 = [] # default value of list attribute4 = 42 # defined value in the class

Note

It is important that all attributes which may be set at instantiation are predefined in the class. Failure to do so will result in an exception being raised.

Methods

init(**kwargs): The initializer that handles the assignment of keyword arguments to instance attributes. It enforces strict attribute assignment rules, only allowing attributes that are already defined in the class to be set.

Initialize an instance of the derived class, strictly assigning provided keyword arguments to corresponding instance attributes.

Parameters

**kwargs: Variable length keyword arguments.

Raises

Exception

If a key from kwargs does not correspond to any attribute pre-defined in the class, an exception is raised to prevent setting an undefined attribute.

Expand source code

class Sqlite__Database(Kwargs_To_Self):
    db_path         : str  = None
    closed          : bool = False
    connected       : bool = False
    deleted         : bool = False
    in_memory       : bool = True                       # default to an in-memory database
    auto_schema_row : bool = False                      # option to map the table's schema_row when creating an table object

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        if self.db_path:                        # if self.db_path is set via the ctor (then it means that this is not in memory)
            self.in_memory = False              # todo: see if this is not better done with a direct method for a getter/setter

    def close(self):
        if self.closed is False:
            self.connection().close()
            self.closed    = True
            self.connected = False
            return True
        return False

    # def config(self, key):
    #     return self.table_config().data().get(key)

    @cache_on_self
    def connect(self):
        connection_string      = self.connection_string()
        connection             = sqlite3.connect(connection_string)
        connection.row_factory = self.dict_factory                      # this returns a dict as the row value of every query
        self.connected         = True
        return connection

    def connection(self):
        return self.connect()

    def connection_string(self):
        if self.in_memory:
            return SQLITE_DATABASE_PATH__IN_MEMORY
        if not self.db_path:
            self.db_path = self.path_temp_database()
        return self.db_path

    @cache_on_self
    def cursor(self):
        from osbot_utils.helpers.sqlite.Sqlite__Cursor import Sqlite__Cursor
        return Sqlite__Cursor(database=self)

    def delete(self):
        if self.in_memory:          # can't delete an in-memory database
            return False
        if self.deleted:
            return False
        self.close()
        if file_delete(self.db_path):
            self.deleted = True
            return True
        return False

    def dict_factory(self, cursor, row):                        # from https://docs.python.org/3/library/sqlite3.html#how-to-create-and-use-row-factories
        fields = [column[0] for column in cursor.description]
        return {key: value for key, value in zip(fields, row)}

    def exists(self):
        if self.in_memory:
            return True
        return file_exists(self.db_path)

    def path_temp_database(self, file_name=None):
        if file_name is None:
            file_name = TEMP_DATABASE__FILE_NAME_PREFIX + random_filename(extension=TEMP_DATABASE__FILE_EXTENSION)
        return path_combine(self.path_temp_databases(), file_name)

    def path_temp_databases(self):
        path_temp_databases  = path_combine(current_temp_folder(), FOLDER_NAME_TEMP_DATABASES)      # use current temp folder has the parent folder
        folder_create(path_temp_databases)                                                          # make sure it exists
        return path_temp_databases

    def save_to(self, path):
        connection = self.connection()
        file_conn  = sqlite3.connect(path)
        connection.backup(file_conn)
        file_conn.close()
        return path


    def table(self, table_name):
        from osbot_utils.helpers.sqlite.Sqlite__Table import Sqlite__Table              # need to import here due to circular imports
        table = Sqlite__Table(database=self, table_name=table_name)
        if self.auto_schema_row:
            table.row_schema__set_from_field_types()                            # todo: see if we shouldn't just propagate the auto_schema_row to the Sqlite__Table and do that on the ctor
        return table

    # def table_config(self):
    #     from osbot_utils.helpers.sqlite.tables.Sqlite__Table__Config import Sqlite__Table__Config
    #     table_config = Sqlite__Table__Config(database=self)
    #     table_config.setup()
    #     return table_config

    def table__sqlite_master(self):
        return self.table('sqlite_master')

    def tables(self):
        tables = []
        for table_data in self.tables_raw():
            table_name = table_data.get('name')
            table      = self.table(table_name)
            tables.append(table)
        return tables

    def tables_raw(self):
        return self.cursor().tables()

    def tables_names(self, include_sqlite_master=False):
        table_names = self.table__sqlite_master().select_field_values('name')
        if include_sqlite_master:
            table_names.append('sqlite_master')
        return table_names

    def purge_database(self):       # this fells like a better name than vacuum :)
        return self.vacuum()

    def vacuum(self):
        return self.cursor().vacuum()

Ancestors

• Kwargs_To_Self

Subclasses

• PubSub__Sqlite
• Sqlite__DB__Local

Class variables

var auto_schema_row : bool

var closed : bool

var connected : bool

var db_path : str

var deleted : bool

var in_memory : bool

Methods

def close(self)

Expand source code

def close(self):
    if self.closed is False:
        self.connection().close()
        self.closed    = True
        self.connected = False
        return True
    return False

def connect(self)

Expand source code

@cache_on_self
def connect(self):
    connection_string      = self.connection_string()
    connection             = sqlite3.connect(connection_string)
    connection.row_factory = self.dict_factory                      # this returns a dict as the row value of every query
    self.connected         = True
    return connection

def connection(self)

Expand source code

def connection(self):
    return self.connect()

def connection_string(self)

Expand source code

def connection_string(self):
    if self.in_memory:
        return SQLITE_DATABASE_PATH__IN_MEMORY
    if not self.db_path:
        self.db_path = self.path_temp_database()
    return self.db_path

def cursor(self)

Expand source code

@cache_on_self
def cursor(self):
    from osbot_utils.helpers.sqlite.Sqlite__Cursor import Sqlite__Cursor
    return Sqlite__Cursor(database=self)

def delete(self)

Expand source code

def delete(self):
    if self.in_memory:          # can't delete an in-memory database
        return False
    if self.deleted:
        return False
    self.close()
    if file_delete(self.db_path):
        self.deleted = True
        return True
    return False

def dict_factory(self, cursor, row)

Expand source code

def dict_factory(self, cursor, row):                        # from https://docs.python.org/3/library/sqlite3.html#how-to-create-and-use-row-factories
    fields = [column[0] for column in cursor.description]
    return {key: value for key, value in zip(fields, row)}

def exists(self)

Expand source code

def exists(self):
    if self.in_memory:
        return True
    return file_exists(self.db_path)

def path_temp_database(self, file_name=None)

Expand source code

def path_temp_database(self, file_name=None):
    if file_name is None:
        file_name = TEMP_DATABASE__FILE_NAME_PREFIX + random_filename(extension=TEMP_DATABASE__FILE_EXTENSION)
    return path_combine(self.path_temp_databases(), file_name)

def path_temp_databases(self)

Expand source code

def path_temp_databases(self):
    path_temp_databases  = path_combine(current_temp_folder(), FOLDER_NAME_TEMP_DATABASES)      # use current temp folder has the parent folder
    folder_create(path_temp_databases)                                                          # make sure it exists
    return path_temp_databases

def purge_database(self)

Expand source code

def purge_database(self):       # this fells like a better name than vacuum :)
    return self.vacuum()

def save_to(self, path)

Expand source code

def save_to(self, path):
    connection = self.connection()
    file_conn  = sqlite3.connect(path)
    connection.backup(file_conn)
    file_conn.close()
    return path

def table(self, table_name)

Expand source code

def table(self, table_name):
    from osbot_utils.helpers.sqlite.Sqlite__Table import Sqlite__Table              # need to import here due to circular imports
    table = Sqlite__Table(database=self, table_name=table_name)
    if self.auto_schema_row:
        table.row_schema__set_from_field_types()                            # todo: see if we shouldn't just propagate the auto_schema_row to the Sqlite__Table and do that on the ctor
    return table

def table__sqlite_master(self)

Expand source code

def table__sqlite_master(self):
    return self.table('sqlite_master')

def tables(self)

Expand source code

def tables(self):
    tables = []
    for table_data in self.tables_raw():
        table_name = table_data.get('name')
        table      = self.table(table_name)
        tables.append(table)
    return tables

def tables_names(self, include_sqlite_master=False)

Expand source code

def tables_names(self, include_sqlite_master=False):
    table_names = self.table__sqlite_master().select_field_values('name')
    if include_sqlite_master:
        table_names.append('sqlite_master')
    return table_names

def tables_raw(self)

Expand source code

def tables_raw(self):
    return self.cursor().tables()

def vacuum(self)

Expand source code

def vacuum(self):
    return self.cursor().vacuum()

Inherited members

• Kwargs_To_Self:
  • update_from_kwargs