database

DataStore

DataStore(location: str = None, extension: str = '.txt', glob_extension: str = None, load_file: Callable = None, save_file: Callable = None, sql=None, log: Logger = logger)

Stores data (currently in a directory) for a singular type of data with particular load and save behaviors

Parameters:

• location (str, default: None ) –

  The location to store/retrieve data if directories are used

• extension (str, default: '.txt' ) –

  The extension of files to use during file handling. If extension is other than .txt or .json, the arguments load_file/save_file must be provided to handle storage

• load_file (Callable, default: None ) –

  Callable taking the file_name as first argument

• save_file (Callable, default: None ) –

  Callable taking the object to save as first argument and file_name as second argument

• sql –

  The database to use if the storage is based on a SQL database

• log (Logger, default: logger ) –

  The Logger to handle operation reporting

Source code in symdesign/resources/database.py

def __init__(self, location: str = None, extension: str = '.txt', glob_extension: str = None,
             load_file: Callable = None, save_file: Callable = None, sql=None, log: Logger = logger):
    """Construct the instance

    Args:
        location: The location to store/retrieve data if directories are used
        extension: The extension of files to use during file handling. If extension is other than .txt or .json, the
            arguments load_file/save_file must be provided to handle storage
        load_file: Callable taking the file_name as first argument
        save_file: Callable taking the object to save as first argument and file_name as second argument
        sql: The database to use if the storage is based on a SQL database
        log: The Logger to handle operation reporting
    """
    self.log = log
    if sql is not None:
        self.sql = sql
    else:
        self.sql = sql
        self.location = location
        self.extension = extension
        if glob_extension:
            self.glob_extension = glob_extension
        else:
            self.glob_extension = extension

        if '.txt' in extension:  # '.txt' read the file and return the lines
            load_file = read_file
            save_file = write_list_to_file
        elif '.json' in extension:
            load_file = utils.read_json
            save_file = utils.write_json
        else:
            if load_file is None:
                load_file = not_implemented
            if save_file is None:
                save_file = not_implemented

        self.load_file = load_file
        self.save_file = save_file

make_path

make_path(condition: bool = True)

Make all required directories in specified path if it doesn't exist, and optional condition is True

Parameters:

• condition (bool, default: True ) –

  A condition to check before the path production is executed

Source code in symdesign/resources/database.py

def make_path(self, condition: bool = True):
    """Make all required directories in specified path if it doesn't exist, and optional condition is True

    Args:
        condition: A condition to check before the path production is executed
    """
    if condition:
        os.makedirs(self.location, exist_ok=True)

path_to

path_to(name: str = '*') -> AnyStr

Return the path_to the storage location given an entity name

Source code in symdesign/resources/database.py

def path_to(self, name: str = '*') -> AnyStr:
    """Return the path_to the storage location given an entity name"""
    return os.path.join(self.location, f'{name}{self.extension}')

glob_path

glob_path(name: str = '*') -> AnyStr

Return the glob_path of the storage location given an entity name

Source code in symdesign/resources/database.py

def glob_path(self, name: str = '*') -> AnyStr:
    """Return the glob_path of the storage location given an entity name"""
    return os.path.join(self.location, f'{name}{self.glob_extension}')

retrieve_file

retrieve_file(name: str) -> AnyStr | None

Returns the actual location by combining the requested name with the stored .location

Source code in symdesign/resources/database.py

def retrieve_file(self, name: str) -> AnyStr | None:
    """Returns the actual location by combining the requested name with the stored .location"""
    path = self.glob_path(name)
    files = sorted(glob(path))
    if files:
        file = files[0]
        if len(files) > 1:
            self.log.warning(f'Found more than one file with glob "{path}". Grabbing the first one: {file}')
        return file
    else:
        self.log.debug(f'No file found for "{path}"')
        return None

retrieve_files

retrieve_files() -> list

Returns the actual location of all files in the stored .location

Source code in symdesign/resources/database.py

def retrieve_files(self) -> list:
    """Returns the actual location of all files in the stored .location"""
    path = self.glob_path()
    files = sorted(glob(path))
    if not files:
        self.log.debug(f'No files found in "{path}"')
    return files

retrieve_names

retrieve_names() -> list[str]

Returns the names of all objects in the stored .location

Source code in symdesign/resources/database.py

def retrieve_names(self) -> list[str]:
    """Returns the names of all objects in the stored .location"""
    path = self.glob_path()
    names = list(map(os.path.basename, [os.path.splitext(file)[0] for file in sorted(glob(path))]))
    if not names:
        self.log.debug(f'No data found with names at "{path}"')
    return names

store_data

store_data(data: Any, name: str, **kwargs)

Store the data specfied by data and name to the DataStore. Saves the data as well

Parameters:

• data (Any) –

  The data object to be stored with name

• name (str) –

  The name of the data to be used

Sets

self.name = data

Source code in symdesign/resources/database.py

def store_data(self, data: Any, name: str, **kwargs):
    """Store the data specfied by data and name to the DataStore. Saves the data as well

    Args:
        data: The data object to be stored with name
        name: The name of the data to be used

    Sets:
        self.name = data
    """
    self.__setattr__(name, data)
    self._save_data(data, name, **kwargs)

retrieve_data

retrieve_data(name: str = None) -> Any | None

Return the data requested by name. Otherwise, load into the Database from a specified location

Parameters:

• name (str, default: None ) –

  The name of the data to be retrieved. Will be found with location and extension attributes

Returns:

• Any | None –

  If the data is available, the object requested will be returned, else None

Source code in symdesign/resources/database.py

def retrieve_data(self, name: str = None) -> Any | None:
    """Return the data requested by name. Otherwise, load into the Database from a specified location

    Args:
        name: The name of the data to be retrieved. Will be found with location and extension attributes

    Returns:
        If the data is available, the object requested will be returned, else None
    """
    if name is None:
        return None
    data = getattr(self, name, None)
    if data:
        self.log.debug(f'{name}{self.extension} was retrieved from {self.__class__.__name__}')
    else:
        data = self.load_data(name, log=None)  # Attempt to retrieve the new data
        if data:
            self.__setattr__(name, data)  # Attempt to store the new data as an attribute
            self.log.debug(f'The file {name}{self.extension} was loaded into the {self.__class__.__name__}')
        else:
            self.log.debug(f'Failed to load file {name}{self.extension} into the {self.__class__.__name__}')

    return data

load_data

load_data(name: str, **kwargs) -> Any | None

Return the data located in a particular entry specified by name

Parameters:

• name (str) –

  The name of the data to be used

Returns:

• Any | None –

  The data found at the requested name if any

Source code in symdesign/resources/database.py

def load_data(self, name: str, **kwargs) -> Any | None:
    """Return the data located in a particular entry specified by name

    Args:
        name: The name of the data to be used

    Returns:
        The data found at the requested name if any
    """
    if self.sql:
        dummy = True
    else:
        file = self.retrieve_file(name)
        if file:
            return self.load_file(file, **kwargs)
    return None

load_all_data

load_all_data(**kwargs)

Loads all data located in the particular DataStore storage location

Source code in symdesign/resources/database.py

def load_all_data(self, **kwargs):
    """Loads all data located in the particular DataStore storage location"""
    if self.sql:
        dummy = True
    else:
        for file in sorted(glob(self.glob_path())):
            # self.log.debug('Fetching %s' % file)
            self.__setattr__(os.path.splitext(os.path.basename(file))[0], self.load_file(file))

Database

Database(sql=None, log: Logger = logger)

A common interface to interact with DataStore instances

Parameters:

• sql –

  The database to use if the storage is based on a SQL database

• log (Logger, default: logger ) –

  The Logger to handle operation reporting

Source code in symdesign/resources/database.py

def __init__(self, sql=None, log: Logger = logger):
    """Construct the instance

    Args:
        sql: The database to use if the storage is based on a SQL database
        log: The Logger to handle operation reporting
    """
    super().__init__()
    if sql:
        raise NotImplementedError('SQL set up has not been completed')

    self.sql = sql
    self.log = log
    self.job: 'resources.job.JobResources' | None = None

load_all_data

load_all_data()

For every resource, acquire all existing data in memory

Source code in symdesign/resources/database.py

def load_all_data(self):
    """For every resource, acquire all existing data in memory"""
    for source in self.sources:
        try:
            source.load_all_data()
        except ValueError:
            raise ValueError(
                f'Issue loading data from source {source}')

source

source(name: str) -> DataStore

Return on of the various DataStores supported by the Database

Parameters:

• name (str) –

  The name of the data source to use

Source code in symdesign/resources/database.py

def source(self, name: str) -> DataStore:
    """Return on of the various DataStores supported by the Database

    Args:
        name: The name of the data source to use
    """
    try:
        return getattr(self, name)
    except AttributeError:
        raise AttributeError(
            f"There was no source named '{name}' found in the {self.__class__.__name__}. "
            f'Possible sources are: {", ".join(self.__dict__)}')

retrieve_data

retrieve_data(source: str = None, name: str = None) -> Any | None

Return the data requested by name from the specified source. Otherwise, load into the Database from a specified location

Parameters:

• source (str, default: None ) –

  The name of the data source to use

• name (str, default: None ) –

  The name of the data to be retrieved. Will be found with location and extension attributes

Returns:

• Any | None –

  If the data is available, the object requested will be returned, else None

Source code in symdesign/resources/database.py

def retrieve_data(self, source: str = None, name: str = None) -> Any | None:
    """Return the data requested by name from the specified source. Otherwise, load into the Database from a
    specified location

    Args:
        source: The name of the data source to use
        name: The name of the data to be retrieved. Will be found with location and extension attributes

    Returns:
        If the data is available, the object requested will be returned, else None
    """
    return self.source(source).retrieve_data(name)

retrieve_file

retrieve_file(source: str = None, name: str = None) -> AnyStr | None

Retrieve the file specified by the source and identifier name

Parameters:

• source (str, default: None ) –

  The name of the data source to use

• name (str, default: None ) –

  The name of the data to be retrieved. Will be found with location and extension attributes

Returns:

• AnyStr | None –

  If the file is available, it will be returned, else None

Source code in symdesign/resources/database.py

def retrieve_file(self, source: str = None, name: str = None) -> AnyStr | None:
    """Retrieve the file specified by the source and identifier name

    Args:
        source: The name of the data source to use
        name: The name of the data to be retrieved. Will be found with location and extension attributes

    Returns:
        If the file is available, it will be returned, else None
    """
    return self.source(source).retrieve_file(name)

read_file

read_file(file, **kwargs) -> list[AnyStr]

The simplest form of parsing a file encoded in ASCII characters

Source code in symdesign/resources/database.py

def read_file(file, **kwargs) -> list[AnyStr]:
    """The simplest form of parsing a file encoded in ASCII characters"""
    with open(file, 'r') as f:
        return f.readlines()

write_str_to_file

write_str_to_file(string, file_name, **kwargs) -> AnyStr

Use standard file IO to write a string to a file

Parameters:

• string –

  The string to write

• file_name –

  The location of the file to write

Returns:

• AnyStr –

  The name of the written file

Source code in symdesign/resources/database.py

def write_str_to_file(string, file_name, **kwargs) -> AnyStr:
    """Use standard file IO to write a string to a file

    Args:
        string: The string to write
        file_name: The location of the file to write

    Returns:
        The name of the written file
    """
    with open(file_name, 'w') as f_save:
        f_save.write(f'{string}\n')

    return file_name

write_list_to_file

write_list_to_file(_list, file_name, **kwargs) -> AnyStr

Use standard file IO to write a string to a file

Parameters:

• _list –

  The string to write

• file_name –

  The location of the file to write

Returns:

• AnyStr –

  The name of the written file

Source code in symdesign/resources/database.py

def write_list_to_file(_list, file_name, **kwargs) -> AnyStr:
    """Use standard file IO to write a string to a file

    Args:
        _list: The string to write
        file_name: The location of the file to write

    Returns:
        The name of the written file
    """
    with open(file_name, 'w') as f_save:
        lines = '\n'.join(map(str, _list))
        f_save.write(f'{lines}\n')

    return file_name