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
keyargument 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 resultRead 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_strA 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 defaultD.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_keysReturns 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_keysReturns 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 eA 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