Module geode.lib.database.filesystem

Classes

class FileSystemDatabase (*paths, extension='ini', mkdir=False, **kwargs)

Constructor

Parameters

paths : tuple

Represents database directory paths on filesystem based of str or pathlib.Path values.

extension : str, optional

Extension used to mark database file (Default: ini).

mkdir : bool, optional

Generate the database directory when the path not exists and the value is True, raises an exception otherwise (Default: False).

kwargs : dict, optional

Positional arguments used by pathlib.Path.mkdir method.

See Also

pathlib.Path.mkdir

Create a new directoy at the specified path.

Raises

NotADirectoryError

When the specified path is not a directory.

FileNotFoundError

When the specified path not exists and mkdir flag is set to False.

Expand source code

class FileSystemDatabase(CommonPath):

    def __init__(self, *paths, extension="ini", mkdir=False, **kwargs):
        """ Constructor

        Parameters
        ----------
        paths : tuple
            Represents database directory paths on filesystem based of str or
            pathlib.Path values.
        extension : str, optional
            Extension used to mark database file (Default: ini).
        mkdir : bool, optional
            Generate the database directory when the path not exists and the
            value is True, raises an exception otherwise (Default: False).
        kwargs : dict, optional
            Positional arguments used by pathlib.Path.mkdir method.

        See Also
        --------
        pathlib.Path.mkdir : Create a new directoy at the specified path.

        Raises
        ------
        NotADirectoryError
            When the specified path is not a directory.
        FileNotFoundError
            When the specified path not exists and mkdir flag is set to False.
        """

        CommonPath.__init__(self, *paths)

        if self.path.exists() and not self.path.is_dir():
            raise NotADirectoryError(
                f"Specified path '{self.path}' is not a directory")

        if not self.path.exists():
            if not mkdir:
                raise FileNotFoundError(
                    f"Cannot found '{self.path}' directory on filesystem")

            self.path.mkdir(**kwargs)

        if type(extension) is not str:
            raise TypeError(f"File extension needs to be a str, "
                            f"not a {type(extension).__name__}")

        self.__extension = extension

        # Keep table files based on filename without extension as keys
        self.__tables = dict()

    def __get_table(self, table):
        """ Retrieve a specific table from local cache

        Parameters
        ----------
        table : str
            Table name to find in local cache.

        Returns
        -------
        geode.lib.common.CommonConfiguration
            Table object

        Raises
        ------
        TypeError
            When the specified table name type is not a str.
        FileNotFoundError
            When the specified table name is not in local cache.
        """

        if type(table) is not str:
            raise TypeError(
                f"Table name needs to be a str, not a {type(table).__name__}")

        if table not in self.__tables.keys():
            raise FileNotFoundError(
                f"Cannot found table '{table}' in database folder")

        return self.__tables[table]

    def read(self):
        """ Read and retrieve available files in database directory

        Files are retrieving from the 'extension' attribute set in object
        constructor.

        Raises
        ------
        FileNotFoundError
            When the database directory was not found on filesystem.
        """

        if not self.exists():
            raise FileNotFoundError(
                f"Cannot found '{self.path}' directory on filesystem")

        for filepath in self.path.glob(f"*.{self.__extension}"):
            self.__tables[filepath.stem] = CommonConfiguration(filepath)

    def rename_table(self, table, new_name):
        """ Rename a specific table

        Parameters
        ----------
        table : str
            Table name to find in local cache.
        new_name : str
            New table name.

        Returns
        -------
        geode.lib.common.CommonConfiguration
            Renamed database object.

        Raises
        ------
        TypeError
            When the specified new table name type is not a str.
        FileExistsError
            When the specified new table name was already used by another
            database file on filesystem.
        """

        table_object = self.__get_table(table)

        if type(new_name) is not str:
            raise TypeError(f"New table name needs to be a str, "
                            f"not a {type(new_name).__name__}")

        filepath = self.path.joinpath(f"{new_name}.{self.__extension}")

        if filepath.exists():
            raise FileExistsError(f"Cannot use '{new_name}' as new table name "
                                  f"since this name is already taken")

        table_object.path.rename(filepath)

        self.__tables[new_name] = table_object

        del self.__tables[table]

        return table_object

    def create_table(self, table):
        """ Create a new table from a specific name

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

        Returns
        -------
        geode.lib.common.CommonConfiguration
            Created database object.

        Raises
        ------
        TypeError
            When the specified table name type is not a str.
        FileExistsError
            When the specified table name is already used by another database
            file.
        """

        if type(table) is not str:
            raise TypeError(
                f"Table name needs to be a str, not a {type(table).__name__}")

        if table in self.__tables.keys():
            raise FileExistsError(
                f"Table '{table}' already exists in database folder")

        filepath = self.path.joinpath(f"{table}.{self.__extension}")

        table_object = CommonConfiguration(filepath)
        table_object.write()

        self.__tables[filepath.stem] = table_object

        return table_object

    def drop_table(self, table):
        """ Drop a specific table

        Parameters
        ----------
        table : str
            Table name to find in local cache.

        Raises
        ------
        TypeError
            When the specified table name type is not a str.
        FileNotFoundError
            When the specified table name cannot be found in database
            directory.
        """

        if type(table) is not str:
            raise TypeError(
                f"Table name needs to be a str, not a {type(table).__name__}")

        if table not in self.__tables.keys():
            raise FileNotFoundError(
                f"Table '{table}' did not exists and cannot be removed")

        table_object = self.__get_table(table)
        table_object.path.unlink()

        del table_object, self.__tables[table]

    def delete(self, table, row):
        """ Delete a specific row from specified table

        Parameters
        ----------
        table : str
            Table name to find in local cache.
        row : str

        Raises
        ------
        TypeError
            When the specified row identifier type is not a str.
        KeyError
            When the specified row identifier did not exists in table object.
        """

        table_object = self.__get_table(table)

        if type(row) is not str:
            raise TypeError(f"Row identifier needs to be a str, "
                            f"not a {type(row).__name__}")

        if not table_object.has_section(row):
            raise KeyError(f"Cannot update row '{row}' since this one did not "
                           f"exists in table '{table}'")

        table_object.remove_section(row)
        table_object.write()

    def select(self, table, *rows):
        """ Select specific rows from specified table

        Parameters
        ----------
        table : str
            Table name to find in local cache.
        rows : tuple of str, optional
            Rows identifiers to check in founded table object (Default: all).

        Returns
        -------
        dict
            Founded rows from specified table with row identifier as keys.
        """

        table_object = self.__get_table(table)

        if not rows:
            rows = table_object.sections()

        results = dict()

        for row in rows:
            if type(row) is not str:
                raise TypeError(f"Row identifier needs to be a str, "
                                f"not a {type(row).__name__}")

            if not table_object.has_section(row):
                continue

            results[row] = dict(table_object.items(row))

        return results

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

        Parameters
        ----------
        table : str
            Table name to find in local cache.
        row : str
            Row identifier key to check in founded table object.
        data : dict
            Row data.

        Raises
        ------
        TypeError
            When the specified row identifier type is not a str.
        ValueError
            When the specified row identifier already exists in table object.
        """

        table_object = self.__get_table(table)

        if type(row) is not str:
            raise TypeError(f"Row identifier needs to be a str, "
                            f"not a {type(row).__name__}")

        if table_object.has_section(row):
            raise ValueError(f"Cannot insert row '{row}' since this one "
                             f"already exists in table '{table}'")

        table_object.add_section(row)

        try:
            for key, value in data.items():
                table_object.set(row, key, value)

            table_object.write()

        except TypeError as error:
            table_object.remove_section(row)

            raise TypeError(error)

    def update(self, table, row, **data):
        """ Update a specific row in specified table

        Parameters
        ----------
        table : str
            Table name to find in local cache.
        row : str
            Row identifier key to check in founded table object.
        data : dict
            New row data.

        Raises
        ------
        TypeError
            When the specified row identifier type is not a str.
        KeyError
            When the specified row identifier did not exists in table object.
        """

        if type(row) is not str:
            raise TypeError(f"Row identifier needs to be a str, "
                            f"not a {type(row).__name__}")

        table_object = self.__get_table(table)

        if not table_object.has_section(row):
            raise KeyError(f"Cannot update row '{row}' since this one did not "
                           f"exists in table '{table}'")

        for key, value in data.items():
            table_object.set(row, key, value)

        table_object.write()

Ancestors

• CommonPath

Methods

def create_table(self, table)

Create a new table from a specific name

Parameters

table : str

Table name.

Returns

CommonConfiguration

Created database object.

Raises

TypeError

When the specified table name type is not a str.

FileExistsError

When the specified table name is already used by another database file.

Expand source code

def create_table(self, table):
    """ Create a new table from a specific name

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

    Returns
    -------
    geode.lib.common.CommonConfiguration
        Created database object.

    Raises
    ------
    TypeError
        When the specified table name type is not a str.
    FileExistsError
        When the specified table name is already used by another database
        file.
    """

    if type(table) is not str:
        raise TypeError(
            f"Table name needs to be a str, not a {type(table).__name__}")

    if table in self.__tables.keys():
        raise FileExistsError(
            f"Table '{table}' already exists in database folder")

    filepath = self.path.joinpath(f"{table}.{self.__extension}")

    table_object = CommonConfiguration(filepath)
    table_object.write()

    self.__tables[filepath.stem] = table_object

    return table_object

def delete(self, table, row)

Delete a specific row from specified table

Parameters

table : str

Table name to find in local cache.

row : str

 

Raises

TypeError

When the specified row identifier type is not a str.

KeyError

When the specified row identifier did not exists in table object.

Expand source code

def delete(self, table, row):
    """ Delete a specific row from specified table

    Parameters
    ----------
    table : str
        Table name to find in local cache.
    row : str

    Raises
    ------
    TypeError
        When the specified row identifier type is not a str.
    KeyError
        When the specified row identifier did not exists in table object.
    """

    table_object = self.__get_table(table)

    if type(row) is not str:
        raise TypeError(f"Row identifier needs to be a str, "
                        f"not a {type(row).__name__}")

    if not table_object.has_section(row):
        raise KeyError(f"Cannot update row '{row}' since this one did not "
                       f"exists in table '{table}'")

    table_object.remove_section(row)
    table_object.write()

def drop_table(self, table)

Drop a specific table

Parameters

table : str

Table name to find in local cache.

Raises

TypeError

When the specified table name type is not a str.

FileNotFoundError

When the specified table name cannot be found in database directory.

Expand source code

def drop_table(self, table):
    """ Drop a specific table

    Parameters
    ----------
    table : str
        Table name to find in local cache.

    Raises
    ------
    TypeError
        When the specified table name type is not a str.
    FileNotFoundError
        When the specified table name cannot be found in database
        directory.
    """

    if type(table) is not str:
        raise TypeError(
            f"Table name needs to be a str, not a {type(table).__name__}")

    if table not in self.__tables.keys():
        raise FileNotFoundError(
            f"Table '{table}' did not exists and cannot be removed")

    table_object = self.__get_table(table)
    table_object.path.unlink()

    del table_object, self.__tables[table]

def insert(self, table, row, **data)

Insert a new row in specified table

Parameters

table : str

Table name to find in local cache.

row : str

Row identifier key to check in founded table object.

data : dict

Row data.

Raises

TypeError

When the specified row identifier type is not a str.

ValueError

When the specified row identifier already exists in table object.

Expand source code

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

    Parameters
    ----------
    table : str
        Table name to find in local cache.
    row : str
        Row identifier key to check in founded table object.
    data : dict
        Row data.

    Raises
    ------
    TypeError
        When the specified row identifier type is not a str.
    ValueError
        When the specified row identifier already exists in table object.
    """

    table_object = self.__get_table(table)

    if type(row) is not str:
        raise TypeError(f"Row identifier needs to be a str, "
                        f"not a {type(row).__name__}")

    if table_object.has_section(row):
        raise ValueError(f"Cannot insert row '{row}' since this one "
                         f"already exists in table '{table}'")

    table_object.add_section(row)

    try:
        for key, value in data.items():
            table_object.set(row, key, value)

        table_object.write()

    except TypeError as error:
        table_object.remove_section(row)

        raise TypeError(error)

def read(self)

Read and retrieve available files in database directory

Files are retrieving from the 'extension' attribute set in object constructor.

Raises

FileNotFoundError

When the database directory was not found on filesystem.

Expand source code

def read(self):
    """ Read and retrieve available files in database directory

    Files are retrieving from the 'extension' attribute set in object
    constructor.

    Raises
    ------
    FileNotFoundError
        When the database directory was not found on filesystem.
    """

    if not self.exists():
        raise FileNotFoundError(
            f"Cannot found '{self.path}' directory on filesystem")

    for filepath in self.path.glob(f"*.{self.__extension}"):
        self.__tables[filepath.stem] = CommonConfiguration(filepath)

def rename_table(self, table, new_name)

Rename a specific table

Parameters

table : str

Table name to find in local cache.

new_name : str

New table name.

Returns

CommonConfiguration

Renamed database object.

Raises

TypeError

When the specified new table name type is not a str.

FileExistsError

When the specified new table name was already used by another database file on filesystem.

Expand source code

def rename_table(self, table, new_name):
    """ Rename a specific table

    Parameters
    ----------
    table : str
        Table name to find in local cache.
    new_name : str
        New table name.

    Returns
    -------
    geode.lib.common.CommonConfiguration
        Renamed database object.

    Raises
    ------
    TypeError
        When the specified new table name type is not a str.
    FileExistsError
        When the specified new table name was already used by another
        database file on filesystem.
    """

    table_object = self.__get_table(table)

    if type(new_name) is not str:
        raise TypeError(f"New table name needs to be a str, "
                        f"not a {type(new_name).__name__}")

    filepath = self.path.joinpath(f"{new_name}.{self.__extension}")

    if filepath.exists():
        raise FileExistsError(f"Cannot use '{new_name}' as new table name "
                              f"since this name is already taken")

    table_object.path.rename(filepath)

    self.__tables[new_name] = table_object

    del self.__tables[table]

    return table_object

def select(self, table, *rows)

Select specific rows from specified table

Parameters

table : str

Table name to find in local cache.

rows : tuple of str, optional

Rows identifiers to check in founded table object (Default: all).

Returns

dict

Founded rows from specified table with row identifier as keys.

Expand source code

def select(self, table, *rows):
    """ Select specific rows from specified table

    Parameters
    ----------
    table : str
        Table name to find in local cache.
    rows : tuple of str, optional
        Rows identifiers to check in founded table object (Default: all).

    Returns
    -------
    dict
        Founded rows from specified table with row identifier as keys.
    """

    table_object = self.__get_table(table)

    if not rows:
        rows = table_object.sections()

    results = dict()

    for row in rows:
        if type(row) is not str:
            raise TypeError(f"Row identifier needs to be a str, "
                            f"not a {type(row).__name__}")

        if not table_object.has_section(row):
            continue

        results[row] = dict(table_object.items(row))

    return results

def update(self, table, row, **data)

Update a specific row in specified table

Parameters

table : str

Table name to find in local cache.

row : str

Row identifier key to check in founded table object.

data : dict

New row data.

Raises

TypeError

When the specified row identifier type is not a str.

KeyError

When the specified row identifier did not exists in table object.

Expand source code

def update(self, table, row, **data):
    """ Update a specific row in specified table

    Parameters
    ----------
    table : str
        Table name to find in local cache.
    row : str
        Row identifier key to check in founded table object.
    data : dict
        New row data.

    Raises
    ------
    TypeError
        When the specified row identifier type is not a str.
    KeyError
        When the specified row identifier did not exists in table object.
    """

    if type(row) is not str:
        raise TypeError(f"Row identifier needs to be a str, "
                        f"not a {type(row).__name__}")

    table_object = self.__get_table(table)

    if not table_object.has_section(row):
        raise KeyError(f"Cannot update row '{row}' since this one did not "
                       f"exists in table '{table}'")

    for key, value in data.items():
        table_object.set(row, key, value)

    table_object.write()

Inherited members

• CommonPath:
  • exists
  • path
  • set_path