Module geode.lib.database.sqlite

Classes

class SqliteDatabase (path=None)

Constructor

Using None as parameter allow to use the sqlite memory backend

Parameters

path : str, optional

Database file path (Default: None)

Expand source code

class SqliteDatabase(CommonPath):

    class Type(object):
        CREATE = 0
        ALTER = 1
        DROP = 2
        SELECT = 3
        INSERT = 4
        UPDATE = 5
        DELETE = 6

    def __init__(self, path=None):
        """ Constructor

        Using None as parameter allow to use the sqlite memory backend

        Parameters
        ----------
        path : str, optional
            Database file path (Default: None)
        """

        CommonPath.__init__(self, path)

        self.__connection = None

    def __del__(self):
        """ Close database instance when object is deleted
        """

        if self.is_connected:
            self.__connection.close()

    def __execute(self, request_type, request, *args):
        """ Execute a specific SQL request

        Parameters
        ----------
        request_type : geode.lib.database.SqliteDatabase.Type
            Request type, used to manage results
        request : str
            SQL request string

        Returns
        -------
        tuple
            if request_type is SELECT and some results are available

        Raises
        ------
        SystemError
            when method is called without database connection
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        if sqlite.complete_statement(request):

            try:
                self.__cursor.execute(request, args)

                if request_type == SqliteDatabase.Type.SELECT:
                    return self.__cursor.fetchall() or None

                elif self.__connection.in_transaction:
                    self.__connection.commit()

            except sqlite.OperationalError as error:
                raise SystemError(error)

            except sqlite.IntegrityError as error:

                if request_type == SqliteDatabase.Type.INSERT:
                    raise IndexError(error)

                raise ValueError(error)

    @property
    def is_connected(self):
        """ Retrieve database connection status

        Returns
        -------
        bool
            True if database is connected, False otherwise
        """

        return self.__connection is not None

    def connect(self):
        """ Start database connection

        This function generate a new sqlite database at self.path position on
        filesystem.

        If self.path is None, database use memory backend.

        Raises
        ------
        SystemError
            when a connection already exists
        """

        if self.is_connected:
            raise SystemError("Database already connected")

        connection_type = ":memory:"

        if self.path is not None:
            connection_type = self.path

        self.__connection = sqlite.connect(str(connection_type))

        self.__cursor = self.__connection.cursor()

    def alter_table(self, table, **kwargs):
        """ Alter a database table

        This function allow to alter table with two modes : rename and
        add_column. These two functions can be used together, but the rename is
        launched before the add_column.

        Parameters
        ----------
        table : str
            Table name
        rename : str
            New table name
        add_column : str
            New column name and attribute to add

        Raises
        ------
        SystemError
            when database connection is not etablish
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        request = f"ALTER TABLE \"{table}\""

        if "rename" in kwargs.keys():
            request += f" RENAME TO \"{kwargs.get('rename')}\";"

        if "add_column" in kwargs.keys():
            request += f" ADD COLUMN \"{kwargs.get('add_column')}\";"

        self.__execute(SqliteDatabase.Type.ALTER, request)

    def create_table(self, table, *args):
        """ Create a new table in database

        Parameters
        ----------
        table : str
            Table name
        *args : (str,)
            Table columns strings

        Raises
        ------
        SystemError
            when database connection is not etablish
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        request = f"CREATE TABLE \"{table}\" ({','.join(args)});"

        self.__execute(SqliteDatabase.Type.CREATE, request)

    def drop_table(self, table, if_exists=True):
        """ Drop a specific table from database

        Parameters
        ----------
        table : str
            Table name
        if_exists : bool, optional
            Raises an error if False, nothing otherwise (Default: True)

        Raises
        ------
        SystemError
            when database connection is not etablish
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        request = f"DROP TABLE {'IF EXISTS ' if if_exists else ''}{table};"

        self.__execute(SqliteDatabase.Type.DROP, request)

    def pragma_table(self, table):
        """ Execute specific function on a specific table

        Parameters
        ----------
        table : str
            Table name

        Returns
        -------
        tuple
            Request results

        Raises
        ------
        SystemError
            when database connection is not etablish
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        request = f"PRAGMA table_info(\"{table}\");"

        return self.__execute(SqliteDatabase.Type.SELECT, request)

    def delete(self, table, where=None):
        """ Delete rows from a specific table

        Without the where condition, this function erase all the data from the
        specified table.

        Parameters
        ----------
        table : str
            Table name
        where : list(str,), optional
            Conditional arguments to filter request (Default: None)

        Raises
        ------
        SystemError
            when database connection is not etablish
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        request = f"DELETE FROM \"{table}\""

        if where is not None:

            if type(where) is list:
                where = " AND ".join(where)

            request += f" WHERE {where}"

        request += ';'

        self.__execute(SqliteDatabase.Type.DELETE, request)

    def select(self, table, *args, where=None):
        """ Retrieve information from specified table

        Parameters
        ----------
        table : str
            Table name
        args : tuple(str,)
            Columns to retrieve from specified table
        where : list(str,), optional
            Conditional arguments to filter request (Default: None)

        Returns
        -------
        list
            Found rows

        Raises
        ------
        SystemError
            when database connection is not etablish
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        request = f"SELECT {escape_request(*args)} FROM \"{table}\""

        if where is not None:

            if type(where) is list:
                where = " AND ".join(where)

            request += f" WHERE {where}"

        request += ';'

        return self.__execute(SqliteDatabase.Type.SELECT, request)

    def insert(self, table, **kwargs):
        """ Insert a new row in specified table

        Parameters
        ----------
        table : str
            Table name
        kwargs : dict(str: str,)
            Values to insert with column as key

        Raises
        ------
        SystemError
            when database connection is not etablish
        AttributeError
            when values to insert are missing from arguments
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        if not kwargs:
            raise AttributeError("No argument available to manage request")

        keys = escape_request(*kwargs.keys())

        values = ','.join(['?'] * len(kwargs))

        request = f"INSERT INTO \"{table}\" ({keys}) VALUES ({values});"

        self.__execute(
            SqliteDatabase.Type.INSERT, request, *kwargs.values())

    def update(self, table, where=None, **kwargs):
        """ Update data from table for a specific row

        Parameters
        ----------
        table : str
            Table name
        kwargs : dict(str: str,)
            Values to update with column as key
        where : list(str,), optional
            Conditional arguments to filter request (Default: None)

        Raises
        ------
        SystemError
            when database connection is not etablish
        AttributeError
            when values to update are missing from arguments
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        if not kwargs:
            raise AttributeError("No argument available to manage request")

        request = f"UPDATE \"{table}\" SET {escape_request(**kwargs)}"

        if where is not None:

            if type(where) is list:
                where = " AND ".join(where)

            request += f" WHERE {where}"

        request += ';'

        self.__execute(SqliteDatabase.Type.UPDATE, request)

    def set_progress_handler(self, handler, step=1):
        """ Set a progress handler for database requests

        Parameters
        ----------
        handler : builtin_function_or_method
            Function to call every updated step
        step : int, optional
            Step to wait between each update (Default: 1)

        Raises
        ------
        SystemError
            when database connection is not etablish
        """

        if self.__connection is None:
            raise SystemError("Cannot execute a request without connection")

        self.__connection.set_progress_handler(handler, step)

Ancestors

• CommonPath

Class variables

var Type

Expand source code

class Type(object):
    CREATE = 0
    ALTER = 1
    DROP = 2
    SELECT = 3
    INSERT = 4
    UPDATE = 5
    DELETE = 6

Instance variables

var is_connected

Retrieve database connection status

Returns

bool

True if database is connected, False otherwise

Expand source code

@property
def is_connected(self):
    """ Retrieve database connection status

    Returns
    -------
    bool
        True if database is connected, False otherwise
    """

    return self.__connection is not None

Methods

def alter_table(self, table, **kwargs)

Alter a database table

This function allow to alter table with two modes : rename and add_column. These two functions can be used together, but the rename is launched before the add_column.

Parameters

table : str

Table name

rename : str

New table name

add_column : str

New column name and attribute to add

Raises

SystemError

when database connection is not etablish

Expand source code

def alter_table(self, table, **kwargs):
    """ Alter a database table

    This function allow to alter table with two modes : rename and
    add_column. These two functions can be used together, but the rename is
    launched before the add_column.

    Parameters
    ----------
    table : str
        Table name
    rename : str
        New table name
    add_column : str
        New column name and attribute to add

    Raises
    ------
    SystemError
        when database connection is not etablish
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    request = f"ALTER TABLE \"{table}\""

    if "rename" in kwargs.keys():
        request += f" RENAME TO \"{kwargs.get('rename')}\";"

    if "add_column" in kwargs.keys():
        request += f" ADD COLUMN \"{kwargs.get('add_column')}\";"

    self.__execute(SqliteDatabase.Type.ALTER, request)

def connect(self)

Start database connection

This function generate a new sqlite database at self.path position on filesystem.

If self.path is None, database use memory backend.

Raises

SystemError

when a connection already exists

Expand source code

def connect(self):
    """ Start database connection

    This function generate a new sqlite database at self.path position on
    filesystem.

    If self.path is None, database use memory backend.

    Raises
    ------
    SystemError
        when a connection already exists
    """

    if self.is_connected:
        raise SystemError("Database already connected")

    connection_type = ":memory:"

    if self.path is not None:
        connection_type = self.path

    self.__connection = sqlite.connect(str(connection_type))

    self.__cursor = self.__connection.cursor()

def create_table(self, table, *args)

Create a new table in database

Parameters

table : str

Table name

*args : (str,)

Table columns strings

Raises

SystemError

when database connection is not etablish

Expand source code

def create_table(self, table, *args):
    """ Create a new table in database

    Parameters
    ----------
    table : str
        Table name
    *args : (str,)
        Table columns strings

    Raises
    ------
    SystemError
        when database connection is not etablish
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    request = f"CREATE TABLE \"{table}\" ({','.join(args)});"

    self.__execute(SqliteDatabase.Type.CREATE, request)

def delete(self, table, where=None)

Delete rows from a specific table

Without the where condition, this function erase all the data from the specified table.

Parameters

table : str

Table name

where : list(str,), optional

Conditional arguments to filter request (Default: None)

Raises

SystemError

when database connection is not etablish

Expand source code

def delete(self, table, where=None):
    """ Delete rows from a specific table

    Without the where condition, this function erase all the data from the
    specified table.

    Parameters
    ----------
    table : str
        Table name
    where : list(str,), optional
        Conditional arguments to filter request (Default: None)

    Raises
    ------
    SystemError
        when database connection is not etablish
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    request = f"DELETE FROM \"{table}\""

    if where is not None:

        if type(where) is list:
            where = " AND ".join(where)

        request += f" WHERE {where}"

    request += ';'

    self.__execute(SqliteDatabase.Type.DELETE, request)

def drop_table(self, table, if_exists=True)

Drop a specific table from database

Parameters

table : str

Table name

if_exists : bool, optional

Raises an error if False, nothing otherwise (Default: True)

Raises

SystemError

when database connection is not etablish

Expand source code

def drop_table(self, table, if_exists=True):
    """ Drop a specific table from database

    Parameters
    ----------
    table : str
        Table name
    if_exists : bool, optional
        Raises an error if False, nothing otherwise (Default: True)

    Raises
    ------
    SystemError
        when database connection is not etablish
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    request = f"DROP TABLE {'IF EXISTS ' if if_exists else ''}{table};"

    self.__execute(SqliteDatabase.Type.DROP, request)

def insert(self, table, **kwargs)

Insert a new row in specified table

Parameters

table : str

Table name

kwargs : dict(str: str,)

Values to insert with column as key

Raises

SystemError

when database connection is not etablish

AttributeError

when values to insert are missing from arguments

Expand source code

def insert(self, table, **kwargs):
    """ Insert a new row in specified table

    Parameters
    ----------
    table : str
        Table name
    kwargs : dict(str: str,)
        Values to insert with column as key

    Raises
    ------
    SystemError
        when database connection is not etablish
    AttributeError
        when values to insert are missing from arguments
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    if not kwargs:
        raise AttributeError("No argument available to manage request")

    keys = escape_request(*kwargs.keys())

    values = ','.join(['?'] * len(kwargs))

    request = f"INSERT INTO \"{table}\" ({keys}) VALUES ({values});"

    self.__execute(
        SqliteDatabase.Type.INSERT, request, *kwargs.values())

def pragma_table(self, table)

Execute specific function on a specific table

Parameters

table : str

Table name

Returns

tuple

Request results

Raises

SystemError

when database connection is not etablish

Expand source code

def pragma_table(self, table):
    """ Execute specific function on a specific table

    Parameters
    ----------
    table : str
        Table name

    Returns
    -------
    tuple
        Request results

    Raises
    ------
    SystemError
        when database connection is not etablish
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    request = f"PRAGMA table_info(\"{table}\");"

    return self.__execute(SqliteDatabase.Type.SELECT, request)

def select(self, table, *args, where=None)

Retrieve information from specified table

Parameters

table : str

Table name

args : tuple(str,)

Columns to retrieve from specified table

where : list(str,), optional

Conditional arguments to filter request (Default: None)

Returns

list

Found rows

Raises

SystemError

when database connection is not etablish

Expand source code

def select(self, table, *args, where=None):
    """ Retrieve information from specified table

    Parameters
    ----------
    table : str
        Table name
    args : tuple(str,)
        Columns to retrieve from specified table
    where : list(str,), optional
        Conditional arguments to filter request (Default: None)

    Returns
    -------
    list
        Found rows

    Raises
    ------
    SystemError
        when database connection is not etablish
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    request = f"SELECT {escape_request(*args)} FROM \"{table}\""

    if where is not None:

        if type(where) is list:
            where = " AND ".join(where)

        request += f" WHERE {where}"

    request += ';'

    return self.__execute(SqliteDatabase.Type.SELECT, request)

def set_progress_handler(self, handler, step=1)

Set a progress handler for database requests

Parameters

handler : builtin_function_or_method

Function to call every updated step

step : int, optional

Step to wait between each update (Default: 1)

Raises

SystemError

when database connection is not etablish

Expand source code

def set_progress_handler(self, handler, step=1):
    """ Set a progress handler for database requests

    Parameters
    ----------
    handler : builtin_function_or_method
        Function to call every updated step
    step : int, optional
        Step to wait between each update (Default: 1)

    Raises
    ------
    SystemError
        when database connection is not etablish
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    self.__connection.set_progress_handler(handler, step)

def update(self, table, where=None, **kwargs)

Update data from table for a specific row

Parameters

table : str

Table name

kwargs : dict(str: str,)

Values to update with column as key

where : list(str,), optional

Conditional arguments to filter request (Default: None)

Raises

SystemError

when database connection is not etablish

AttributeError

when values to update are missing from arguments

Expand source code

def update(self, table, where=None, **kwargs):
    """ Update data from table for a specific row

    Parameters
    ----------
    table : str
        Table name
    kwargs : dict(str: str,)
        Values to update with column as key
    where : list(str,), optional
        Conditional arguments to filter request (Default: None)

    Raises
    ------
    SystemError
        when database connection is not etablish
    AttributeError
        when values to update are missing from arguments
    """

    if self.__connection is None:
        raise SystemError("Cannot execute a request without connection")

    if not kwargs:
        raise AttributeError("No argument available to manage request")

    request = f"UPDATE \"{table}\" SET {escape_request(**kwargs)}"

    if where is not None:

        if type(where) is list:
            where = " AND ".join(where)

        request += f" WHERE {where}"

    request += ';'

    self.__execute(SqliteDatabase.Type.UPDATE, request)

Inherited members

• CommonPath:
  • exists
  • path
  • set_path