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

class YAMLConfigManager(entries: dict[str, Any] | list[Any] | None = None, wait_max: int = DEFAULT_WAIT_TIME, strict_ro_locks: bool = False)

Bases: MutableMapping

A YAML configuration manager.

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

Object constructor.

Parameters:

• entries (dict[str, Any] | list[Any] | None): YAML collection of key-value pairs. (default: None)
• wait_max (int): How long to wait for creating an object when the file that data will be read from is locked. (default: 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. (default: False)

Attributes

• data (dict[str, Any])
• filepath (str | None)
• locker (Any)
• strict_ro_locks (bool)
• wait_max (int)

Properties

exp

@property
def exp -> dict[str, Any]

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:

• dict[str, Any]: Dictionary with expanded paths and variables.

locked

@property
def locked -> bool

Check if the file is currently locked.

Returns:

• bool: True if the locker exists and is locked, False otherwise.

settings

@property
def settings -> dict[str, Any]

Get the configuration settings for this object.

Returns:

• dict[str, Any]: Dictionary containing wait_max, locked, and strict_ro_locks settings.

Class methods

from_obj

@classmethod
def from_obj(entries: dict[str, Any] | list[Any] | None, **kwargs) -> YAMLConfigManager

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

Parameters:

• entries (dict[str, Any] | list[Any] | None): Object to initialize from.
• **kwargs: Keyword arguments to pass to the constructor. (default: {})

Returns:

• YAMLConfigManager: New instance of the class.

from_yaml_data

@classmethod
def from_yaml_data(yamldata: str, **kwargs) -> YAMLConfigManager

Initialize from a YAML string.

Parameters:

• yamldata (str): YAML-formatted string.
• **kwargs: Keyword arguments to pass to the constructor. (default: {})

Returns:

• YAMLConfigManager: New instance of the class.

from_yaml_file

@classmethod
def from_yaml_file(filepath: str | Path, create_file: bool = False, **kwargs) -> YAMLConfigManager

Initialize from a YAML file.

Parameters:

• filepath (str | Path): Path to the YAML config file.
• create_file (bool): Create a file at filepath if it doesn't exist. (default: False)
• **kwargs: Keyword arguments to pass to the constructor. (default: {})

Returns:

• YAMLConfigManager: New instance of the class.

Methods

__del__

def __del__() -> None

Destructor that cleans up the locker if it exists.

__delitem__

def __delitem__(key: str) -> None

Delete a key-value pair from the configuration.

Parameters:

• key (str): The key to delete.

__enter__

def __enter__()

Context manager entry not supported.

Raises:

• NotImplementedError: Always raised; use 'read_lock' and 'write_lock' context managers instead.

__exit__

def __exit__()

Context manager exit not supported.

Raises:

• NotImplementedError: Always raised; use 'read_lock' and 'write_lock' context managers instead.

__getitem__

def __getitem__(item: str) -> object

Fetch the value of given key.

Parameters:

• item (str): Key for which to fetch value.

Returns:

• object: Value mapped to given key, if available.

Raises:

• KeyError: If the requested key is unmapped.

__iter__

def __iter__() -> Iterator[str]

Return an iterator over the configuration keys.

__len__

def __len__() -> int

Return the number of configuration entries.

__repr__

def __repr__() -> str

Return string representation of the object.

Returns:

• str: YAML representation of the object's data.

__setitem__

def __setitem__(item: str, value: object) -> None

Set a key-value pair in the configuration.

Parameters:

• item (str): The key to set.
• value (object): The value to set for the key.

priority_get

def priority_get(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.

Parameters:

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

Returns:

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

Raises:

• Exception: If strict is True and the value cannot be determined.

rebase

def rebase(filepath: str | Path | None = None) -> YAMLConfigManager

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

Parameters:

• filepath (str | Path | None): Path to the file that should be read. If not provided, uses the object's current filepath. (default: None)

Returns:

• YAMLConfigManager: Self for chaining.

rebase_and_write

def rebase_and_write() -> 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:

• str: The absolute path to the written file.

reset

def reset(filepath: str | Path | None = None) -> YAMLConfigManager

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

Parameters:

• filepath (str | Path | None): Path to the file that should be read. If not provided, uses the object's current filepath. (default: None)

Returns:

• YAMLConfigManager: Self for chaining.

to_dict

def to_dict(expand: bool = True) -> dict[str, Any]

Convert the object to a dictionary.

Parameters:

• expand (bool): Whether to expand paths in values (currently unused, kept for backwards compatibility). (default: True)

Returns:

• dict[str, Any]: The object's data as a dictionary.

to_yaml

def to_yaml(trailing_newline: bool = False, expand: bool = False) -> str

Get text for YAML representation.

Parameters:

• trailing_newline (bool): Whether to add trailing newline. (default: False)
• expand (bool): Whether to expand paths in values. (default: False)

Returns:

• str: YAML text representation of this instance.

update_from_obj

def update_from_obj(entries: dict[str, Any] | None = None) -> YAMLConfigManager

Update the object's data from a Python object.

Parameters:

• entries (dict[str, Any] | None): Object (dict, list, or primitive) to update from. (default: None)

update_from_yaml_data

def update_from_yaml_data(yamldata: str | None = None) -> YAMLConfigManager

Update the object's data from a YAML string.

Parameters:

• yamldata (str | None): YAML-formatted string to update from. (default: None)

update_from_yaml_file

def update_from_yaml_file(filepath: str | Path | None = None) -> YAMLConfigManager

Update the object's data from a YAML file.

Parameters:

• 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. (default: None)

write

def write() -> str

Write the contents to the file backing this object.

Returns:

• str: 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.

write_copy

def write_copy(filepath: str | Path) -> str

Write the contents to an external file.

Parameters:

• filepath (str | Path): A file path to write to.

Returns:

• str: The filepath that was written to.

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.