Module pipettin-piper.piper.config.config_helper

Functions

def default_accessor(key=None)
Expand source code
def default_accessor(key=None):
    """Default accessor function, always raises KeyError."""
    raise KeyError("Key not found in accessor.")

Default accessor function, always raises KeyError.

def make_accessor(obj, attributes: list = None, selectors: list = None)
Expand source code
def make_accessor(obj, attributes: list = None, selectors: list = None):
    """
    Create an accessor function that retrieves a value from an object, 
    optionally using a specified attribute and selectors to navigate through nested elements.

    Args:
        obj: The base object to access.
        attributes (list, optional): An optional list of attribute names to retrieve from the object using `getattr`.
        selectors (list, optional): A list of keys or indexes to access nested elements within the object.

    Returns:
        function: A partially applied accessor function that takes a `key` argument to further access
                  the structure within the object.

    Example:
        class foo:
            bar = {
                "baz": [
                    0,
                    1,
                    {"yay": 3}
                ]
            }
        acc = make_accessor(foo, attribute=["bar"], selectors=["baz", 2])
        acc("yay")
    """
    def accessor_prototype(key: str, obj, attributes: str, selectors: list):
        # Sub-selection.
        try:
            # Optionally select an attribute of object.
            if attributes is not None:
                for a in attributes:
                    obj = getattr(obj, a)
            # Optionally select sub-objects by a list of indexes or key names.
            if selectors is not None:
                for s in selectors:
                    obj = obj[s]
        except Exception as e:
            raise ValueError("Error in accessor attribute or selectors.") from e
        # Main selection.
        try:
            res = obj[key]
        except (IndexError, KeyError) as e:
            msg = f"Could not access key {key} in object {type(obj).__name__}, using attributes={attributes} and selectors {selectors}."
            raise KeyError(msg) from e
        return res
    return functools.partial(accessor_prototype, obj=obj, attributes=attributes, selectors=selectors)

Create an accessor function that retrieves a value from an object, optionally using a specified attribute and selectors to navigate through nested elements.

Args

obj
The base object to access.
attributes : list, optional
An optional list of attribute names to retrieve from the object using getattr.
selectors : list, optional
A list of keys or indexes to access nested elements within the object.

Returns

function
A partially applied accessor function that takes a key argument to further access the structure within the object.

Example

class foo: bar = { "baz": [ 0, 1, {"yay": 3} ] } acc = make_accessor(foo, attribute=["bar"], selectors=["baz", 2]) acc("yay")

def read_config_file(file_path)
Expand source code
def read_config_file(file_path):
    """Read a configuration file and determine the format automatically."""
    ext = Path(file_path).suffix.lower()
    if ext in {'.yaml', '.yml', '.json'}:
        return read_yaml_json_config(file_path)
    elif ext in {'.ini', '.cfg'}:
        return read_ini_cfg_config(file_path)
    else:
        raise ValueError(f"Unsupported config file format: {ext}")

Read a configuration file and determine the format automatically.

def read_ini_cfg_config(file_path)
Expand source code
def read_ini_cfg_config(file_path):
    """Read an INI or CFG file, including unnamed top-level options.

    TODO: Unfortunately configparser does not "guess" variable types, everything is a string.
          This means that many types will differ relative to the original YAML file, so I'm not
          using this method for now. Another issue is that top-level keys are supproted only
          for reading, but not for writing...
    """
    config = configparser.ConfigParser(allow_unnamed_section=True)
    config.read(file_path, encoding='utf-8')
    result = {section: dict(config.items(section)) for section in config.sections()}
    if configparser.UNNAMED_SECTION in config:
        result.update(dict(config.items(configparser.UNNAMED_SECTION)))
    return result

Read an INI or CFG file, including unnamed top-level options.

TODO: Unfortunately configparser does not "guess" variable types, everything is a string. This means that many types will differ relative to the original YAML file, so I'm not using this method for now. Another issue is that top-level keys are supproted only for reading, but not for writing…

def read_yaml_json_config(file_path)
Expand source code
def read_yaml_json_config(file_path):
    """Read a JSON or YAML file."""
    with open(file_path, 'r', encoding='utf-8') as file:
        return yaml.safe_load(file)

Read a JSON or YAML file.

def save_dict_as_ini(data, file_path)
Expand source code
def save_dict_as_ini(data, file_path):
    """Save a Python dictionary as an INI file."""
    config = configparser.ConfigParser()

    for section, parameters in data.items():
        if isinstance(parameters, dict):
            config.add_section(section)
            for key, value in parameters.items():
                config.set(section, key, str(value))
        else:
            logging.warning(f"Failed to save unsupported top-level config section '{key}' to INI file.")

    with open(file_path, 'w') as configfile:
        config.write(configfile)

Save a Python dictionary as an INI file.

Classes

class TrackedDict (file_path=None, data=None, accessor_func=None, allow_edits=False)
Expand source code
class TrackedDict(YAMLReader):
    """
    A class that represents an 'pseudo-immutable' dictionary with keys and values obtained from a YAML file.
    It tracks accessed keys internally, and modifications are not allowed.

    Attributes:
        file_path (str): The path to the YAML file.
        data (dict): The parsed data from the YAML file.
        accessed_keys (set): A set of keys that have been accessed.

    Methods:
        __init__(self, file_path): Initializes the TrackedDict instance.
        __getitem__(self, key): Retrieves the value for a given key, tracking the key internally.
        __setitem__(self, key, value): Raises TypeError, as TrackedDict is immutable.
        __delitem__(self, key): Raises TypeError, as TrackedDict is immutable.
        clear(self): Raises TypeError, as TrackedDict is immutable.
        update(self, *args, **kwargs): Raises TypeError, as TrackedDict is immutable.
    """
    def __init__(self, file_path=None, data=None, accessor_func=None, allow_edits=False):
        """
        Initializes the TrackedDict instance.

        Args:
            file_path (str): The path to the YAML file.
            data (dict): Altertative (and sufficient) data source. Used to update the YAML file if any.
        """

        # Set the dynamic accessor (defaulting to one).
        self.accessor = accessor_func or default_accessor
        # Set a placeholder data if the accessor is present,
        # to avoid errors in super init below.
        if data is None and accessor_func:
            data = {}

        # Allow setting keys only during initialization.
        self._keys_set_allowed = True
        # Initialize YAMLReader to read data from the YAML file.
        super().__init__(file_path=file_path, data=data)
        # Create a set to track accessed keys.
        self.accessed_keys = set()
        # Allow setting keys only during initialization.
        self._keys_set_allowed = allow_edits

    def __repr__(self):
        # TODO: This won't print any data behind the accessor.
        return f"TrackedDict({self.data}) [accessor data not shown]"

    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            return default

    def __getitem__(self, key):
        """
        Retrieves the value for a given key, tracking the key internally.

        Args:
            key: The key to retrieve the value for.

        Returns:
            The value associated with the key.
        """
        # Track the accessed key
        self.accessed_keys.add(key)
        # Access the data.
        try:
            # Try using the accessor first.
            res = self.accessor(key)
        except KeyError:
            # Call the __getitem__ method of the parent class (UserDict)
            res = super().__getitem__(key)

        return res

    def __setitem__(self, key, value):
        """
        Raises TypeError, as TrackedDict is immutable and does not allow setting items.

        Args:
            key: The key to set.
            value: The value to set.

        Raises:
            TypeError: Modification not allowed.
        """
        # Raise TypeError to prevent setting items
        if not self._keys_set_allowed:
            raise TypeError("TrackedDict is immutable, cannot set items")
        else:
            return super().__setitem__(key, value)

    def __delitem__(self, key):
        """
        Raises TypeError, as TrackedDict is immutable and does not allow deleting items.

        Args:
            key: The key to delete.

        Raises:
            TypeError: Modification not allowed.
        """
        # Raise TypeError to prevent deleting items
        if not self._keys_set_allowed:
            raise TypeError("TrackedDict is immutable, cannot delete items")
        else:
            return super().__delitem__(key)

    def clear(self):
        """
        Raises TypeError, as TrackedDict is immutable and does not allow clearing items.

        Raises:
            TypeError: Modification not allowed.
        """
        # Raise TypeError to prevent clearing items
        if not self._keys_set_allowed:
            raise TypeError("TrackedDict is immutable, cannot clear items")
        else:
            return super().clear()

    def update(self, *args, **kwargs):
        """
        Raises TypeError, as TrackedDict is immutable and does not allow updating items.

        Raises:
            TypeError: Modification not allowed.
        """
        # Raise TypeError to prevent updating items
        if not self._keys_set_allowed:
            raise TypeError("TrackedDict is immutable, cannot update items")
        else:
            return super().update(*args, **kwargs)

    def get_accessed_keys(self):
        """
        Returns the set of keys that have been accessed.

        Returns:
            set: The set of accessed keys.
        """
        return self.accessed_keys

    def get_non_accessed_keys(self):
        """
        Returns the set of keys that have not been accessed.

        Returns:
            set: The set of non-accessed keys.
        """
        all_keys = set(self.keys())
        non_accessed_keys = all_keys - self.accessed_keys
        return non_accessed_keys

    def __str__(self):
        """
        Returns a string representation of the TrackedDict, including data, accessed keys, and non-accessed keys.

        Returns:
            str: String representation of the TrackedDict.
        """
        pp = pprint.PrettyPrinter(indent=4)
        data_str = "Loaded data:\n" + pp.pformat(self)
        accessed_keys_str = f"\nAccessed Keys: {list(self.get_accessed_keys())}"
        non_accessed_keys_str = f"\nNon-accessed Keys: {list(self.get_non_accessed_keys())}"
        return data_str + non_accessed_keys_str + accessed_keys_str

A class that represents an 'pseudo-immutable' dictionary with keys and values obtained from a YAML file. It tracks accessed keys internally, and modifications are not allowed.

Attributes

file_path : str
The path to the YAML file.
data : dict
The parsed data from the YAML file.
accessed_keys : set
A set of keys that have been accessed.

Methods

init(self, file_path): Initializes the TrackedDict instance. getitem(self, key): Retrieves the value for a given key, tracking the key internally. setitem(self, key, value): Raises TypeError, as TrackedDict is immutable. delitem(self, key): Raises TypeError, as TrackedDict is immutable. clear(self): Raises TypeError, as TrackedDict is immutable. update(self, args, *kwargs): Raises TypeError, as TrackedDict is immutable.

Initializes the TrackedDict instance.

Args

file_path : str
The path to the YAML file.
data : dict
Altertative (and sufficient) data source. Used to update the YAML file if any.

Ancestors

  • YAMLReader
  • collections.UserDict
  • collections.abc.MutableMapping
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Subclasses

Methods

def clear(self)
Expand source code
def clear(self):
    """
    Raises TypeError, as TrackedDict is immutable and does not allow clearing items.

    Raises:
        TypeError: Modification not allowed.
    """
    # Raise TypeError to prevent clearing items
    if not self._keys_set_allowed:
        raise TypeError("TrackedDict is immutable, cannot clear items")
    else:
        return super().clear()

Raises TypeError, as TrackedDict is immutable and does not allow clearing items.

Raises

TypeError
Modification not allowed.
def get(self, key, default=None)
Expand source code
def get(self, key, default=None):
    try:
        return self[key]
    except KeyError:
        return default

D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.

def get_accessed_keys(self)
Expand source code
def get_accessed_keys(self):
    """
    Returns the set of keys that have been accessed.

    Returns:
        set: The set of accessed keys.
    """
    return self.accessed_keys

Returns the set of keys that have been accessed.

Returns

set
The set of accessed keys.
def get_non_accessed_keys(self)
Expand source code
def get_non_accessed_keys(self):
    """
    Returns the set of keys that have not been accessed.

    Returns:
        set: The set of non-accessed keys.
    """
    all_keys = set(self.keys())
    non_accessed_keys = all_keys - self.accessed_keys
    return non_accessed_keys

Returns the set of keys that have not been accessed.

Returns

set
The set of non-accessed keys.
def update(self, *args, **kwargs)
Expand source code
def update(self, *args, **kwargs):
    """
    Raises TypeError, as TrackedDict is immutable and does not allow updating items.

    Raises:
        TypeError: Modification not allowed.
    """
    # Raise TypeError to prevent updating items
    if not self._keys_set_allowed:
        raise TypeError("TrackedDict is immutable, cannot update items")
    else:
        return super().update(*args, **kwargs)

Raises TypeError, as TrackedDict is immutable and does not allow updating items.

Raises

TypeError
Modification not allowed.
class YAMLReader (file_path: str = None, data: dict = None)
Expand source code
class YAMLReader(UserDict):
    """
    A class for reading data from a YAML file.

    Attributes:
        file_path (str): The path to the YAML file.
        data (dict): The parsed data from the YAML file.

    Methods:
        __init__(self, file_path): Initializes the YAMLReader instance.
        _read_yaml(self): Reads and parses data from the YAML file.
    """
    def __init__(self, file_path:str=None, data:dict=None):
        """
        Initializes the YAMLReader instance.

        Args:
            file_path (str): The path to the YAML file.
            data (dict): Altertative (and sufficient) data source. Used to update the YAML file if any.
        """
        self.file_path = file_path
        self.data = {}

        if file_path:
            yaml_data = self._read_yaml()
            self.data.update(yaml_data)

        if data:
            self.data.update(data)

        if file_path is None and data is None:
            raise ValueError("YAMLReader: at least 'file_path' or 'data' must be specified.")

        # Initialize UserDict with the data obtained from YAMLReader.
        super().__init__(self.data)

    def _read_yaml(self):
        """
        Reads and parses data from the YAML file.

        Returns:
            dict: Parsed data from the YAML file.
        """
        try:
            yaml_data = read_yaml_json_config(self.file_path)
            if yaml_data is None:
                raise ValueError(f"Failed to load YAML data from file '{self.file_path}'. The file may be empty (got 'None' from 'safe_load').")
            return yaml_data
        except FileNotFoundError as e:
            print(f"Error: The provided '{self.file_path}' file was not found.")
            raise e
        except Exception as e:
            print(f"Error reading YAML file: {e}")
            raise e

A class for reading data from a YAML file.

Attributes

file_path : str
The path to the YAML file.
data : dict
The parsed data from the YAML file.

Methods

init(self, file_path): Initializes the YAMLReader instance. _read_yaml(self): Reads and parses data from the YAML file.

Initializes the YAMLReader instance.

Args

file_path : str
The path to the YAML file.
data : dict
Altertative (and sufficient) data source. Used to update the YAML file if any.

Ancestors

  • collections.UserDict
  • collections.abc.MutableMapping
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Subclasses