Classes

class Sqlite__Cursor (**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__Cursor(Kwargs_To_Self):
    database : Sqlite__Database

    @cache_on_self
    def cursor(self):
        return self.connection().cursor()

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

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

    def execute(self, sql_query, *params):
        try:
            self.cursor().execute(sql_query, *params)
            return status_ok()
        except Exception as error:
            return status_exception(error=f'{error}')

    def execute_and_commit(self, sql_query, *params):                   # todo: refactor this with the execute method
        try:
            self.cursor().execute(sql_query, *params)
            self.connection().commit()
            return status_ok()
        except Exception as error:
            return status_exception(error=f'{error}')

    def execute__fetch_all(self,sql_query, *params):
        self.execute(sql_query,*params)
        return self.cursor().fetchall()

    def execute__fetch_one(self,sql_query, *params):
        self.execute(sql_query, *params)
        return self.cursor().fetchone()

    def fetch_all(self):
        return self.cursor().fetchall()

    def fetch_one(self):
        return self.cursor().fetchone()

    def table_create(self, table_name, fields):
        if table_name and fields:
            sql_query = f"CREATE TABLE {table_name} ({', '.join(fields)})"
            return self.execute(sql_query=sql_query)
        return status_error(message='table_name, fields cannot be empty')

    def table_delete(self, table_name):
        sql_query = f"DROP TABLE IF EXISTS {table_name};"
        return  self.execute(sql_query=sql_query)

    def table_exists(self, table_name):
        self.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
        return self.cursor().fetchone() is not None

    def table_schema(self, table_name):
        sql_query = f"PRAGMA table_info({table_name});"
        self.execute(sql_query)
        columns   = self.cursor().fetchall()
        return columns

    def table__sqlite_master(self):                            # todo: refactor into separate class
        sql_query = "SELECT * FROM sqlite_master"
        self.execute(sql_query)
        return self.cursor().fetchall()

    def tables(self):
        sql_query = "SELECT * FROM sqlite_master WHERE type='table';"                   # Query to select all table names from the sqlite_master table
        self.execute(sql_query)
        return self.cursor().fetchall()

    def tables_names(self):
        cell_name = 'name'
        sql_query = f"SELECT {cell_name} FROM sqlite_master WHERE type='table';"                   # Query to select all table names from the sqlite_master table
        self.execute(sql_query)
        all_rows   = self.cursor().fetchall()
        all_values = [cell.get(cell_name) for cell in all_rows]
        return all_values

    def vacuum(self):
        return self.execute("VACUUM")

Ancestors

• Kwargs_To_Self

Class variables

var database : Sqlite__Database

Methods

def commit(self)

Expand source code

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

def connection(self)

Expand source code

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

def cursor(self)

Expand source code

@cache_on_self
def cursor(self):
    return self.connection().cursor()

def execute(self, sql_query, *params)

Expand source code

def execute(self, sql_query, *params):
    try:
        self.cursor().execute(sql_query, *params)
        return status_ok()
    except Exception as error:
        return status_exception(error=f'{error}')

def execute__fetch_all(self, sql_query, *params)

Expand source code

def execute__fetch_all(self,sql_query, *params):
    self.execute(sql_query,*params)
    return self.cursor().fetchall()

def execute__fetch_one(self, sql_query, *params)

Expand source code

def execute__fetch_one(self,sql_query, *params):
    self.execute(sql_query, *params)
    return self.cursor().fetchone()

def execute_and_commit(self, sql_query, *params)

Expand source code

def execute_and_commit(self, sql_query, *params):                   # todo: refactor this with the execute method
    try:
        self.cursor().execute(sql_query, *params)
        self.connection().commit()
        return status_ok()
    except Exception as error:
        return status_exception(error=f'{error}')

def fetch_all(self)

Expand source code

def fetch_all(self):
    return self.cursor().fetchall()

def fetch_one(self)

Expand source code

def fetch_one(self):
    return self.cursor().fetchone()

def table__sqlite_master(self)

Expand source code

def table__sqlite_master(self):                            # todo: refactor into separate class
    sql_query = "SELECT * FROM sqlite_master"
    self.execute(sql_query)
    return self.cursor().fetchall()

def table_create(self, table_name, fields)

Expand source code

def table_create(self, table_name, fields):
    if table_name and fields:
        sql_query = f"CREATE TABLE {table_name} ({', '.join(fields)})"
        return self.execute(sql_query=sql_query)
    return status_error(message='table_name, fields cannot be empty')

def table_delete(self, table_name)

Expand source code

def table_delete(self, table_name):
    sql_query = f"DROP TABLE IF EXISTS {table_name};"
    return  self.execute(sql_query=sql_query)

def table_exists(self, table_name)

Expand source code

def table_exists(self, table_name):
    self.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
    return self.cursor().fetchone() is not None

def table_schema(self, table_name)

Expand source code

def table_schema(self, table_name):
    sql_query = f"PRAGMA table_info({table_name});"
    self.execute(sql_query)
    columns   = self.cursor().fetchall()
    return columns

def tables(self)

Expand source code

def tables(self):
    sql_query = "SELECT * FROM sqlite_master WHERE type='table';"                   # Query to select all table names from the sqlite_master table
    self.execute(sql_query)
    return self.cursor().fetchall()

def tables_names(self)

Expand source code

def tables_names(self):
    cell_name = 'name'
    sql_query = f"SELECT {cell_name} FROM sqlite_master WHERE type='table';"                   # Query to select all table names from the sqlite_master table
    self.execute(sql_query)
    all_rows   = self.cursor().fetchall()
    all_values = [cell.get(cell_name) for cell in all_rows]
    return all_values

def vacuum(self)

Expand source code

def vacuum(self):
    return self.execute("VACUUM")

Inherited members

• Kwargs_To_Self:
  • update_from_kwargs