Package yacman Documentation

Package Overview

Yacman is a YAML configuration manager that provides convenience tools for dealing with YAML configuration files. It's designed for safe, concurrent access to configuration data with file locking support and a flexible attribute-based access pattern.

Key Features

• Attribute-Based Access: Access YAML data as object attributes
• File Locking: Race-free reading and writing in multi-user contexts
• Flexible Construction: Create from files, strings, or dictionaries
• Path Expansion: Automatically expand environment variables and paths
• Alias Support: Define custom aliases for configuration keys
• Context Managers: Safe read and write operations with locking

Installation

pip install yacman

Quick Example

from yacman import YAMLConfigManager

# Create from a file
ym = YAMLConfigManager.from_yaml_file("config.yaml")

# Access values
print(ym["my_key"])

# Update and write safely
ym["new_key"] = "new_value"
from yacman import write_lock
with write_lock(ym) as locked_ym:
    locked_ym.rebase()
    locked_ym.write()

API Reference

YAMLConfigManager Class

The main class for managing YAML configuration files with locking support:

YAMLConfigManager

YAMLConfigManager(entries=None, filepath=None, yamldata=None, locked=False, wait_max=DEFAULT_WAIT_TIME, strict_ro_locks=False, skip_read_lock=False, schema_source=None, validate_on_write=False, create_file=False)

Bases: MutableMapping

A YAML configuration manager, providing file locking, loading, writing, etc. for YAML configuration files. Without the requirement of attmap (and without providing attribute-style access).

Object constructor

:param Iterable[(str, object)] | Mapping[str, object] entries: YAML collection of key-value pairs. :param str filepath: Path to the YAML config file. :param str yamldata: YAML-formatted string :param bool locked: Whether to initialize as locked (providing write capability) :param int wait_max: how long to wait for creating an object when the file that data will be read from is locked :param bool strict_ro_locks: By default, we allow RO filesystems that can't be locked. Turn on strict_ro_locks to error if locks cannot be enforced on readonly filesystems. :param bool skip_read_lock: whether the file should not be locked for reading when object is created in read only mode :param str schema_source: path or a URL to a jsonschema in YAML format to use for optional config validation. If this argument is provided the object is always validated at least once, at the object creation stage. :param bool validate_on_write: a boolean indicating whether the object should be validated every time the write method is executed, which is a way of preventing invalid config writing :param str create_file: Create an empty file at filepath upon data load.

Source code in yacman/yacman1.py

def __init__(
    self,
    entries=None,
    filepath=None,
    yamldata=None,
    locked=False,
    wait_max=DEFAULT_WAIT_TIME,
    strict_ro_locks=False,
    skip_read_lock=False,
    schema_source=None,
    validate_on_write=False,
    create_file=False,
):
    """
    Object constructor

    :param Iterable[(str, object)] | Mapping[str, object] entries: YAML collection
        of key-value pairs.
    :param str filepath: Path to the YAML config file.
    :param str yamldata: YAML-formatted string
    :param bool locked: Whether to initialize as locked (providing write capability)
    :param int wait_max: how long to wait for creating an object when the file
        that data will be read from is locked
    :param bool strict_ro_locks: By default, we allow RO filesystems that can't be locked.
        Turn on strict_ro_locks to error if locks cannot be enforced on readonly filesystems.
    :param bool skip_read_lock: whether the file should not be locked for reading
        when object is created in read only mode
    :param str schema_source: path or a URL to a jsonschema in YAML format to use
        for optional config validation. If this argument is provided the object
        is always validated at least once, at the object creation stage.
    :param bool validate_on_write: a boolean indicating whether the object should be
        validated every time the `write` method is executed, which is
        a way of preventing invalid config writing
    :param str create_file: Create an empty file at filepath upon data load.
    """

    # Settings for this config object
    if filepath:
        self.filepath = mkabs(filepath)
    else:
        self.filepath = None
    self.wait_max = wait_max
    self.skip_read_lock = skip_read_lock
    self.schema_source = schema_source
    self.validate_on_write = validate_on_write
    self.locked = locked
    self.strict_ro_locks = strict_ro_locks
    self.already_locked = locked

    if self.locked:
        if filepath:
            create_lock(filepath, wait_max)
        else:
            self.locked = False
            locked = False
            _LOGGER.warning(
                "Argument 'locked' is disregarded when the object is created "
                "with 'entries' rather than 'filepath'"
            )

    if self.filepath and not skip_read_lock:
        with self as _:
            entries = self.load(filepath, entries, yamldata, create_file)
    else:
        entries = self.load(filepath, entries, yamldata, create_file)

    # We store the values in a dict under .data
    if isinstance(entries, list):
        self.data = entries
    else:
        self.data = dict(entries or {})
    if schema_source is not None:
        assert isinstance(schema_source, str), TypeError(
            f"Path to the schema to validate the config must be a string"
        )
        sp = expandpath(schema_source)
        assert os.path.exists(sp), FileNotFoundError(
            f"Provided schema file does not exist: {schema_source}."
            f" Also tried: {sp}"
        )
        # validate config
        setattr(self, SCHEMA_KEY, load_yaml(sp))
        self.validate()

exp property

exp

Returns a copy of the object's data elements with env vars and user vars expanded. Use it like: object.exp["item"]

__getitem__

__getitem__(item)

Fetch the value of given key.

:param hashable item: key for which to fetch value :return object: value mapped to given key, if available :raise KeyError: if the requested key is unmapped.

Source code in yacman/yacman1.py

def __getitem__(self, item):
    """
    Fetch the value of given key.

    :param hashable item: key for which to fetch value
    :return object: value mapped to given key, if available
    :raise KeyError: if the requested key is unmapped.
    """
    return self.data[item]

priority_get

priority_get(arg_name, env_var=None, default=None, override=None, strict=False)

Helper function to select a value from a config, or, if missing, then go to an env var.

:param str arg_name: Argument to retrieve :param bool strict: Should missing args raise an error? False=warning :param env_var: Env var to retrieve from should it be missing from the cfg

Source code in yacman/yacman1.py

def priority_get(
    self,
    arg_name: str,
    env_var: str = None,
    default: str = None,
    override: str = None,
    strict: bool = False,
):
    """
    Helper function to select a value from a config, or, if missing, then go to an env var.

    :param str arg_name: Argument to retrieve
    :param bool strict: Should missing args raise an error? False=warning
    :param env_var: Env var to retrieve from should it be missing from the cfg
    """
    if override:
        return override
    if self.data.get(arg_name) is not None:
        return self.data[arg_name]
    if env_var is not None:
        arg = os.getenv(env_var, None)
        if arg is not None:
            _LOGGER.debug(f"Value '{arg}' sourced from '{env_var}' env var")
            return expandpath(arg)
    if default is not None:
        return default
    if strict:
        message = (
            "Value for required argument '{arg_name}' could not be determined."
        )
        _LOGGER.warning(message)
        raise Exception(message)

rebase

rebase(filepath=None)

Reload the object from file, then update with current information

:param str filepath: path to the file that should be read

Source code in yacman/yacman1.py

@ensure_locked
def rebase(self, filepath=None):
    """
    Reload the object from file, then update with current information

    :param str filepath: path to the file that should be read
    """
    fp = filepath or self.filepath
    if fp is not None:
        local_data = self.data
        self.data = self.load(filepath=fp)
        deep_update(self.data, local_data)
        # self.data.update(local_data)
    else:
        _LOGGER.warning("Rebase has no effect if no filepath given")

    return self

reset

reset(filepath=None)

Reset dict contents to file contents, or to empty dict if no filepath found.

Source code in yacman/yacman1.py

@ensure_locked
def reset(self, filepath=None):
    """
    Reset dict contents to file contents, or to empty dict if no filepath found.
    """
    fp = filepath or self.filepath
    if fp is not None:
        self.data = self.load(filepath=fp, skip_read_lock=True)
    else:
        self.data = self.load(entries={}, skip_read_lock=True)
    return self

to_yaml

to_yaml(trailing_newline=False, expand=False)

Get text for YAML representation.

:param bool trailing_newline: whether to add trailing newline :param bool expand: whether to expand paths in values :return str: YAML text representation of this instance.

Source code in yacman/yacman1.py

def to_yaml(self, trailing_newline=False, expand=False):
    """
    Get text for YAML representation.

    :param bool trailing_newline: whether to add trailing newline
    :param bool expand: whether to expand paths in values
    :return str: YAML text representation of this instance.
    """

    if expand:
        return yaml.dump(self.exp, default_flow_style=False)
    return yaml.dump(self.data, default_flow_style=False) + (
        "\n" if trailing_newline else ""
    )

validate

validate(schema=None, exclude_case=False)

Validate the object against a schema

:param dict schema: a schema object to use to validate, it overrides the one that has been provided at object construction stage :param bool exclude_case: whether to exclude validated objects from the error. Useful when used with large configs

Source code in yacman/yacman1.py

def validate(self, schema=None, exclude_case=False):
    """
    Validate the object against a schema

    :param dict schema: a schema object to use to validate, it overrides the one
        that has been provided at object construction stage
    :param bool exclude_case: whether to exclude validated objects
        from the error. Useful when used with large configs
    """
    try:
        _validate(
            self.to_dict(expand=True),
            schema or getattr(self, SCHEMA_KEY),
        )
    except ValidationError as e:
        _LOGGER.error(
            f"{self.__class__.__name__} object did not pass schema validation"
        )
        if self.filepath is not None:
            # need to unlock locked files in case of validation error so that no
            # locks are left in place
            self.make_readonly()
        if not exclude_case:
            raise
        raise ValidationError(
            f"{self.__class__.__name__} object did not pass schema validation: "
            f"{e.message}"
        )
    _LOGGER.debug("Validated successfully")

write

write(schema=None, exclude_case=False)

Write the contents to the file backing this object.

:param dict schema: a schema object to use to validate, it overrides the one that has been provided at object construction stage :raise OSError: when the object has been created in a read only mode or other process has locked the file :raise TypeError: when the filepath cannot be determined. This takes place only if YacAttMap initialized with a Mapping as an input, not read from file. :raise OSError: when the write is called on an object with no write capabilities or when writing to a file that is locked by a different object :return str: the path to the created files

Source code in yacman/yacman1.py

@ensure_locked
def write(self, schema=None, exclude_case=False):
    """
    Write the contents to the file backing this object.

    :param dict schema: a schema object to use to validate, it overrides the one
        that has been provided at object construction stage
    :raise OSError: when the object has been created in a read only mode or other
        process has locked the file
    :raise TypeError: when the filepath cannot be determined. This takes place only
        if YacAttMap initialized with a Mapping as an input, not read from file.
    :raise OSError: when the write is called on an object with no write capabilities
        or when writing to a file that is locked by a different object
    :return str: the path to the created files
    """
    if not self.filepath:
        raise OSError("Must provide a filepath to write.")

    _check_filepath(self.filepath)
    _LOGGER.debug(f"Writing to file '{self.filepath}'")
    with open(self.filepath, "w") as f:
        f.write(self.to_yaml())

    if schema is not None or self.validate_on_write:
        self.validate(schema=schema, exclude_case=exclude_case)

    abs_path = os.path.abspath(self.filepath)
    _LOGGER.debug(f"Wrote to a file: {abs_path}")
    return os.path.abspath(abs_path)

write_copy

write_copy(filepath=None)

Write the contents to an external file.

:param str filepath: a file path to write to

Source code in yacman/yacman1.py

def write_copy(self, filepath=None):
    """
    Write the contents to an external file.

    :param str filepath: a file path to write to
    """

    _LOGGER.debug(f"Writing to file '{filepath}'")
    with open(filepath, "w") as f:
        f.write(self.to_yaml())
    return filepath

Context Managers

Yacman provides context managers for safe file locking. These are re-exported from the ubiquerg package for convenience.

write_lock(config_manager)

Context manager for write operations with exclusive locking. Prevents other processes from reading or writing the file while you hold the lock.

Parameters: - config_manager (YAMLConfigManager): The configuration manager instance to lock

Returns: - YAMLConfigManager: The locked configuration manager

Usage:

from yacman import YAMLConfigManager, write_lock

ym = YAMLConfigManager.from_yaml_file("config.yaml")
ym["key"] = "value"

with write_lock(ym) as locked_ym:
    locked_ym.rebase()  # Sync with any file changes
    locked_ym.write()   # Write to disk

read_lock(config_manager)

Context manager for read operations with shared locking. Multiple processes can hold read locks simultaneously, but no process can hold a write lock while read locks exist.

Parameters:

• config_manager (YAMLConfigManager): The configuration manager instance to lock

Returns:

• YAMLConfigManager: The locked configuration manager

Usage:

from yacman import YAMLConfigManager, read_lock

ym = YAMLConfigManager.from_yaml_file("config.yaml")

with read_lock(ym) as locked_ym:
    locked_ym.rebase()  # Sync with file
    print(locked_ym.to_dict())

Note: These context managers are provided by the ubiquerg package and re-exported by yacman for convenience. For more details on the locking implementation, see the ubiquerg documentation.

See the tutorial for more examples.

Utility Functions

Yacman provides several utility functions for working with YAML files and paths:

• load_yaml(filepath): Load a YAML file and return its contents as a dictionary
• select_config(config_filepath, config_env_vars, default_config_filepath): Select a configuration file from multiple sources
• expandpath(path): Expand environment variables and user home directory in a path

These functions are available in the yacman module. See the source code or tutorial for usage examples.

Deprecated Classes (v0.x)

The following classes are deprecated in v1.0 and maintained only for backwards compatibility. Use YAMLConfigManager instead:

• YacAttMap - Replaced by YAMLConfigManager
• AliasedYacAttMap - Use YAMLConfigManager instead

See the upgrading guide for migration instructions.