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, wait_max=DEFAULT_WAIT_TIME, strict_ro_locks=False)

Bases: MutableMapping

A YAML configuration manager.

Provides file locking, loading, writing, etc. for YAML configuration files.

Object constructor.

Parameters:

Name	Type	Description	Default
entries	dict[str, Any] | list[Any] | None	YAML collection of key-value pairs.	None
wait_max	int	How long to wait for creating an object when the file that data will be read from is locked.	DEFAULT_WAIT_TIME
strict_ro_locks	bool	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.	False

Source code in yacman/yacman.py

def __init__(
    self,
    entries: dict[str, Any] | list[Any] | None = None,
    wait_max: int = DEFAULT_WAIT_TIME,
    strict_ro_locks: bool = False,
) -> None:
    """Object constructor.

    Args:
        entries: YAML collection of key-value pairs.
        wait_max: How long to wait for creating an object when the file
            that data will be read from is locked.
        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.
    """

    # Settings for this config object
    self.filepath: str | None = None
    self.wait_max: int = wait_max
    self.strict_ro_locks: bool = strict_ro_locks
    self.locker: Any = None  # ThreeLocker type not available

    # We store the values in a dict under .data
    # Note: entries can be list/dict/etc but data is always dict for MutableMapping protocol
    self.data: dict[str, Any]
    if isinstance(entries, list):
        self.data = {
            str(i): v for i, v in enumerate(entries)
        }  # Convert list to dict
    else:
        self.data = dict(entries or {})

exp property

exp

Get data with environment and user variables expanded.

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

Returns:

Type	Description
dict[str, Any]	Dictionary with expanded paths and variables.

locked property

locked

Check if the file is currently locked.

Returns:

Type	Description
bool	True if the locker exists and is locked, False otherwise.

settings property

settings

Get the configuration settings for this object.

Returns:

Type	Description
dict[str, Any]	Dictionary containing wait_max, locked, and strict_ro_locks settings.

__del__

__del__()

Destructor that cleans up the locker if it exists.

Source code in yacman/yacman.py

def __del__(self) -> None:
    """Destructor that cleans up the locker if it exists."""
    if hasattr(self, "locker"):
        del self.locker

__delitem__

__delitem__(key)

Delete a key-value pair from the configuration.

Parameters:

Name	Type	Description	Default
key	str	The key to delete.	required

Source code in yacman/yacman.py

def __delitem__(self, key: str) -> None:
    """Delete a key-value pair from the configuration.

    Args:
        key: The key to delete.
    """
    del self.data[key]

__enter__

__enter__()

Context manager entry not supported.

Raises:

Type	Description
NotImplementedError	Always raised; use 'read_lock' and 'write_lock' context managers instead.

Source code in yacman/yacman.py

def __enter__(self):
    """Context manager entry not supported.

    Raises:
        NotImplementedError: Always raised; use 'read_lock' and 'write_lock'
            context managers instead.
    """
    raise NotImplementedError(
        "Use the 'read_lock' and 'write_lock' context managers."
    )

__exit__

__exit__()

Context manager exit not supported.

Raises:

Type	Description
NotImplementedError	Always raised; use 'read_lock' and 'write_lock' context managers instead.

Source code in yacman/yacman.py

def __exit__(self):
    """Context manager exit not supported.

    Raises:
        NotImplementedError: Always raised; use 'read_lock' and 'write_lock'
            context managers instead.
    """
    raise NotImplementedError(
        "Use the 'read_lock' and 'write_lock' context managers."
    )

__getitem__

__getitem__(item)

Fetch the value of given key.

Parameters:

Name	Type	Description	Default
item	str	Key for which to fetch value.	required

Returns:

Type	Description
object	Value mapped to given key, if available.

Raises:

Type	Description
KeyError	If the requested key is unmapped.

Source code in yacman/yacman.py

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

    Args:
        item: Key for which to fetch value.

    Returns:
        Value mapped to given key, if available.

    Raises:
        KeyError: If the requested key is unmapped.
    """
    return self.data[item]

__iter__

__iter__()

Return an iterator over the configuration keys.

Source code in yacman/yacman.py

def __iter__(self) -> Iterator[str]:
    """Return an iterator over the configuration keys."""
    return iter(self.data)

__len__

__len__()

Return the number of configuration entries.

Source code in yacman/yacman.py

def __len__(self) -> int:
    """Return the number of configuration entries."""
    return len(self.data)

__repr__

__repr__()

Return string representation of the object.

Returns:

Type	Description
str	YAML representation of the object's data.

Source code in yacman/yacman.py

def __repr__(self) -> str:
    """Return string representation of the object.

    Returns:
        YAML representation of the object's data.
    """
    # Render the data in a nice way
    return self.to_yaml()

__setitem__

__setitem__(item, value)

Set a key-value pair in the configuration.

Parameters:

Name	Type	Description	Default
item	str	The key to set.	required
value	object	The value to set for the key.	required

Source code in yacman/yacman.py

def __setitem__(self, item: str, value: object) -> None:
    """Set a key-value pair in the configuration.

    Args:
        item: The key to set.
        value: The value to set for the key.
    """
    self.data[item] = value

from_obj classmethod

from_obj(entries, **kwargs)

Initialize from a Python object (dict, list, or primitive).

Parameters:

Name	Type	Description	Default
entries	dict[str, Any] | list[Any] | None	Object to initialize from.	required
**kwargs		Keyword arguments to pass to the constructor.	{}

Returns:

Type	Description
YAMLConfigManager	New instance of the class.

Source code in yacman/yacman.py

@classmethod
def from_obj(
    cls, entries: dict[str, Any] | list[Any] | None, **kwargs
) -> "YAMLConfigManager":
    """Initialize from a Python object (dict, list, or primitive).

    Args:
        entries: Object to initialize from.
        **kwargs: Keyword arguments to pass to the constructor.

    Returns:
        New instance of the class.
    """
    return cls(entries, **kwargs)

from_yaml_data classmethod

from_yaml_data(yamldata, **kwargs)

Initialize from a YAML string.

Parameters:

Name	Type	Description	Default
yamldata	str	YAML-formatted string.	required
**kwargs		Keyword arguments to pass to the constructor.	{}

Returns:

Type	Description
YAMLConfigManager	New instance of the class.

Source code in yacman/yacman.py

@classmethod
def from_yaml_data(cls, yamldata: str, **kwargs) -> "YAMLConfigManager":
    """Initialize from a YAML string.

    Args:
        yamldata: YAML-formatted string.
        **kwargs: Keyword arguments to pass to the constructor.

    Returns:
        New instance of the class.
    """
    entries = yaml.load(yamldata, YacmanLoader)
    return cls(entries, **kwargs)

from_yaml_file classmethod

from_yaml_file(filepath, create_file=False, **kwargs)

Initialize from a YAML file.

Parameters:

Name	Type	Description	Default
filepath	str | Path	Path to the YAML config file.	required
create_file	bool	Create a file at filepath if it doesn't exist.	False
**kwargs		Keyword arguments to pass to the constructor.	{}

Returns:

Type	Description
YAMLConfigManager	New instance of the class.

Source code in yacman/yacman.py

@classmethod
def from_yaml_file(
    cls, filepath: str | Path, create_file: bool = False, **kwargs
) -> "YAMLConfigManager":
    """Initialize from a YAML file.

    Args:
        filepath: Path to the YAML config file.
        create_file: Create a file at filepath if it doesn't exist.
        **kwargs: Keyword arguments to pass to the constructor.

    Returns:
        New instance of the class.
    """

    file_contents = locked_read_file(filepath, create_file=create_file)
    entries = yaml.load(file_contents, YacmanLoader)
    ref = cls(entries, **kwargs)
    ref.locker = ThreeLocker(filepath)
    ref.filepath = str(filepath)
    return ref

priority_get

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

Select a value with priority: override > config > env_var > default.

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

Parameters:

Name	Type	Description	Default
arg_name	str	Argument to retrieve from config.	required
env_var	str | None	Environment variable to retrieve from if missing from config.	None
default	str | None	Default value if not found in config or environment.	None
override	str | None	Override value that takes precedence over all other sources.	None
strict	bool	Should missing args raise an error? If False, shows warning.	False

Returns:

Type	Description
str | None	The value from the highest priority source, or None if not found
str | None	and strict is False.

Raises:

Type	Description
Exception	If strict is True and the value cannot be determined.

Source code in yacman/yacman.py

def priority_get(
    self,
    arg_name: str,
    env_var: str | None = None,
    default: str | None = None,
    override: str | None = None,
    strict: bool = False,
) -> str | None:
    """Select a value with priority: override > config > env_var > default.

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

    Args:
        arg_name: Argument to retrieve from config.
        env_var: Environment variable to retrieve from if missing from config.
        default: Default value if not found in config or environment.
        override: Override value that takes precedence over all other sources.
        strict: Should missing args raise an error? If False, shows warning.

    Returns:
        The value from the highest priority source, or None if not found
        and strict is False.

    Raises:
        Exception: If strict is True and the value cannot be determined.
    """
    if override:
        return override
    if isinstance(self.data, dict) and self.data.get(arg_name) is not None:
        result = self.data[arg_name]
        if not isinstance(result, str):
            raise TypeError(
                f"Config value for '{arg_name}' must be a string, got {type(result).__name__}"
            )
        return result
    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 = (
            f"Value for required argument '{arg_name}' could not be determined."
        )
        _LOGGER.warning(message)
        raise Exception(message)
    return None

rebase

rebase(filepath=None)

Reload the object from file, then update with current information.

Parameters:

Name	Type	Description	Default
filepath	str | Path | None	Path to the file that should be read. If not provided, uses the object's current filepath.	None

Returns:

Type	Description
YAMLConfigManager	Self for chaining.

Source code in yacman/yacman.py

@ensure_locked(READ)
def rebase(self, filepath: str | Path | None = None) -> "YAMLConfigManager":
    """Reload the object from file, then update with current information.

    Args:
        filepath: Path to the file that should be read. If not provided,
            uses the object's current filepath.

    Returns:
        Self for chaining.
    """
    assert self.locker is not None
    fp = filepath or self.locker.filepath
    if fp is not None:
        local_data = self.data
        self.data = load_yaml(fp)
        _LOGGER.debug(f"Rebased {local_data} with {self.data} from {fp}")
        if self.data is None:
            self.data = local_data
        else:
            deep_update(self.data, local_data)
    else:
        _LOGGER.warning("Rebase has no effect if no filepath given")

    return self

rebase_and_write

rebase_and_write()

Rebase from disk and write. Safe for multi-process scenarios.

This is a convenience method that combines rebase() and write() into a single call. Use this when multiple processes may have written to the file since you read it in.

Returns:

Type	Description
str	The absolute path to the written file.

Source code in yacman/yacman.py

@ensure_locked(WRITE)
def rebase_and_write(self) -> str:
    """Rebase from disk and write. Safe for multi-process scenarios.

    This is a convenience method that combines rebase() and write() into
    a single call. Use this when multiple processes may have written to
    the file since you read it in.

    Returns:
        The absolute path to the written file.
    """
    self.rebase()
    return self.write()

reset

reset(filepath=None)

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

Parameters:

Name	Type	Description	Default
filepath	str | Path | None	Path to the file that should be read. If not provided, uses the object's current filepath.	None

Returns:

Type	Description
YAMLConfigManager	Self for chaining.

Source code in yacman/yacman.py

@ensure_locked(READ)
def reset(self, filepath: str | Path | None = None) -> "YAMLConfigManager":
    """Reset dict contents to file contents, or to empty dict if no filepath found.

    Args:
        filepath: Path to the file that should be read. If not provided,
            uses the object's current filepath.

    Returns:
        Self for chaining.
    """
    assert self.locker is not None
    fp = filepath or self.locker.filepath
    if fp is not None:
        self.data = load_yaml(fp)
    else:
        self.data = {}
    return self

to_dict

to_dict(expand=True)

Convert the object to a dictionary.

Parameters:

Name	Type	Description	Default
expand	bool	Whether to expand paths in values (currently unused, kept for backwards compatibility).	True

Returns:

Type	Description
dict[str, Any]	The object's data as a dictionary.

Source code in yacman/yacman.py

def to_dict(self, expand: bool = True) -> dict[str, Any]:
    """Convert the object to a dictionary.

    Args:
        expand: Whether to expand paths in values (currently unused,
            kept for backwards compatibility).

    Returns:
        The object's data as a dictionary.
    """
    # Seems like it's probably not necessary; can just use the object now.
    # but for backwards compatibility.
    return self.data

to_yaml

to_yaml(trailing_newline=False, expand=False)

Get text for YAML representation.

Parameters:

Name	Type	Description	Default
trailing_newline	bool	Whether to add trailing newline.	False
expand	bool	Whether to expand paths in values.	False

Returns:

Type	Description
str	YAML text representation of this instance.

Source code in yacman/yacman.py

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

    Args:
        trailing_newline: Whether to add trailing newline.
        expand: Whether to expand paths in values.

    Returns:
        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 ""
    )

update_from_obj

update_from_obj(entries=None)

Update the object's data from a Python object.

Parameters:

Name	Type	Description	Default
entries	dict[str, Any] | None	Object (dict, list, or primitive) to update from.	None

Source code in yacman/yacman.py

def update_from_obj(
    self, entries: dict[str, Any] | None = None
) -> "YAMLConfigManager":
    """Update the object's data from a Python object.

    Args:
        entries: Object (dict, list, or primitive) to update from.
    """
    if entries is not None:
        self.data.update(entries)
    return self

update_from_yaml_data

update_from_yaml_data(yamldata=None)

Update the object's data from a YAML string.

Parameters:

Name	Type	Description	Default
yamldata	str | None	YAML-formatted string to update from.	None

Source code in yacman/yacman.py

def update_from_yaml_data(self, yamldata: str | None = None) -> "YAMLConfigManager":
    """Update the object's data from a YAML string.

    Args:
        yamldata: YAML-formatted string to update from.
    """
    if yamldata is not None:
        self.data.update(yaml.load(yamldata, YacmanLoader))
    return self

update_from_yaml_file

update_from_yaml_file(filepath=None)

Update the object's data from a YAML file.

Parameters:

Name	Type	Description	Default
filepath	str | Path | None	Path to the YAML file to update from. If provided and the object's filepath is not set, sets the object's filepath.	None

Source code in yacman/yacman.py

def update_from_yaml_file(
    self, filepath: str | Path | None = None
) -> "YAMLConfigManager":
    """Update the object's data from a YAML file.

    Args:
        filepath: Path to the YAML file to update from. If provided and
            the object's filepath is not set, sets the object's filepath.
    """
    if filepath is not None:  # set filepath to update filepath if uninitialized
        if self.filepath is None:
            self.filepath = str(filepath)
        self.data.update(load_yaml(filepath))
    return self

write

write()

Write the contents to the file backing this object.

Returns:

Type	Description
str	The absolute path to the written file.

Raises:

Type	Description
OSError	When the object has been created in a read only mode or other process has locked the file, or 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.
TypeError	When the filepath cannot be determined.

Source code in yacman/yacman.py

@ensure_locked(WRITE)
def write(self) -> str:
    """Write the contents to the file backing this object.

    Returns:
        The absolute path to the written file.

    Raises:
        OSError: When the object has been created in a read only mode or
            other process has locked the file, or 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.
        TypeError: When the filepath cannot be determined.
    """
    assert self.locker is not None
    if not self.locker.filepath:
        raise OSError("Must provide a filepath to write.")

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

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

write_copy

write_copy(filepath)

Write the contents to an external file.

Parameters:

Name	Type	Description	Default
filepath	str | Path	A file path to write to.	required

Returns:

Type	Description
str	The filepath that was written to.

Source code in yacman/yacman.py

def write_copy(self, filepath: str | Path) -> str:
    """Write the contents to an external file.

    Args:
        filepath: A file path to write to.

    Returns:
        The filepath that was written to.
    """

    _LOGGER.debug(f"Writing to file '{filepath}'")
    with open(filepath, "w") as f:
        f.write(self.to_yaml())
    return str(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.