Utils

A collection of more general utility functions.

LoggerUtility

Utility class for setting up logging consistently.

Source code in maestrowf/utils.py

class LoggerUtility:
    """Utility class for setting up logging consistently."""

    def __init__(self, logger):
        """
        Initialize a new LoggerUtility class instance.

        :param logger: An instance of a logger to configure.
        """
        self._logger = logger

    def configure(self, log_format, log_lvl=2, colors=True):
        """
        Configures the general logging facility.

        :param log_format: String containing the desired logging format.
        :param log_lvl: Integer level (1-5) to set the logger to.
        """
        logging.basicConfig(level=self.map_level(log_lvl), format=log_format)
        if colors:
            coloredlogs.install(level=self.map_level(log_lvl),
                                logger=self._logger, fmt=log_format)

    def add_stream_handler(self, log_format, log_lvl=2):
        """
        Add a stream handler to logging.

        :param log_format: String containing the desired logging format.
        :param log_lvl: Integer level (1-5) to set the logger to.
        """
        # Create the FileHandler and add it to the logger.
        sh = logging.StreamHandler()
        sh.setLevel(self.map_level(log_lvl))
        sh.setFormatter(logging.Formatter(log_format))
        self._logger.addHandler(sh)

    def add_file_handler(self, log_path, log_format, log_lvl=2):
        """
        Add a file handler to logging.

        :param log_path: String containing the file path to store logging.
        :param log_format: String containing the desired logging format.
        :param log_lvl: Integer level (1-5) to set the logger to.
        """
        # Create the FileHandler and add it to the logger.
        formatter = logging.Formatter(log_format)

        fh = logging.FileHandler(log_path)
        fh.setLevel(self.map_level(log_lvl))
        fh.setFormatter(formatter)
        self._logger.addHandler(fh)

    @staticmethod
    def map_level(log_lvl):
        """
        Map level 1-5 to their respective logging enumerations.

        :param log_lvl: Integer level (1-5) representing logging verbosity.
        """
        if log_lvl == 1:
            return logging.DEBUG
        elif log_lvl == 2:
            return logging.INFO
        elif log_lvl == 3:
            return logging.WARNING
        elif log_lvl == 4:
            return logging.ERROR
        else:
            return logging.CRITICAL

__init__(logger)

Initialize a new LoggerUtility class instance.

Parameters:

Name	Type	Description	Default
logger		An instance of a logger to configure.	required

Source code in maestrowf/utils.py

def __init__(self, logger):
    """
    Initialize a new LoggerUtility class instance.

    :param logger: An instance of a logger to configure.
    """
    self._logger = logger

add_file_handler(log_path, log_format, log_lvl=2)

Add a file handler to logging.

Parameters:

Name	Type	Description	Default
log_path		String containing the file path to store logging.	required
log_format		String containing the desired logging format.	required
log_lvl		Integer level (1-5) to set the logger to.	2

Source code in maestrowf/utils.py

def add_file_handler(self, log_path, log_format, log_lvl=2):
    """
    Add a file handler to logging.

    :param log_path: String containing the file path to store logging.
    :param log_format: String containing the desired logging format.
    :param log_lvl: Integer level (1-5) to set the logger to.
    """
    # Create the FileHandler and add it to the logger.
    formatter = logging.Formatter(log_format)

    fh = logging.FileHandler(log_path)
    fh.setLevel(self.map_level(log_lvl))
    fh.setFormatter(formatter)
    self._logger.addHandler(fh)

add_stream_handler(log_format, log_lvl=2)

Add a stream handler to logging.

Parameters:

Name	Type	Description	Default
log_format		String containing the desired logging format.	required
log_lvl		Integer level (1-5) to set the logger to.	2

Source code in maestrowf/utils.py

def add_stream_handler(self, log_format, log_lvl=2):
    """
    Add a stream handler to logging.

    :param log_format: String containing the desired logging format.
    :param log_lvl: Integer level (1-5) to set the logger to.
    """
    # Create the FileHandler and add it to the logger.
    sh = logging.StreamHandler()
    sh.setLevel(self.map_level(log_lvl))
    sh.setFormatter(logging.Formatter(log_format))
    self._logger.addHandler(sh)

configure(log_format, log_lvl=2, colors=True)

Configures the general logging facility.

Parameters:

Name	Type	Description	Default
log_format		String containing the desired logging format.	required
log_lvl		Integer level (1-5) to set the logger to.	2

Source code in maestrowf/utils.py

def configure(self, log_format, log_lvl=2, colors=True):
    """
    Configures the general logging facility.

    :param log_format: String containing the desired logging format.
    :param log_lvl: Integer level (1-5) to set the logger to.
    """
    logging.basicConfig(level=self.map_level(log_lvl), format=log_format)
    if colors:
        coloredlogs.install(level=self.map_level(log_lvl),
                            logger=self._logger, fmt=log_format)

map_level(log_lvl) staticmethod

Map level 1-5 to their respective logging enumerations.

Parameters:

Name	Type	Description	Default
log_lvl		Integer level (1-5) representing logging verbosity.	required

Source code in maestrowf/utils.py

@staticmethod
def map_level(log_lvl):
    """
    Map level 1-5 to their respective logging enumerations.

    :param log_lvl: Integer level (1-5) representing logging verbosity.
    """
    if log_lvl == 1:
        return logging.DEBUG
    elif log_lvl == 2:
        return logging.INFO
    elif log_lvl == 3:
        return logging.WARNING
    elif log_lvl == 4:
        return logging.ERROR
    else:
        return logging.CRITICAL

apply_function(item, func)

Apply a function to items depending on type.

Parameters:

Name	Type	Description	Default
item		A Python primitive to apply a function to.	required
func		Function that returns takes item as a parameter and returns item modified in some way.	required

Source code in maestrowf/utils.py

def apply_function(item, func):
    """
    Apply a function to items depending on type.

    :param item: A Python primitive to apply a function to.
    :param func: Function that returns takes item as a parameter and returns
        item modified in some way.
    """
    if not item:
        return item
    elif isinstance(item, str):
        return func(item)
    elif isinstance(item, list):
        return [apply_function(x, func) for x in item]
    elif isinstance(item, dict):
        return {
            key: apply_function(value, func) for key, value in item.items()}
    else:
        msg = \
            "Encountered an object of type '{}'. Passing." \
            .format(type(item))
        LOGGER.debug(msg)
        return item

coerce_dict_values(obj_to_transform, transform)

Recursively apply a transformation function to all values in a nested mapping.

Traverses the input object, recursively applying the given transformation function to each leaf value. If the input is a mapping (such as a dictionary), the function will process each value in the mapping. Non-mapping values are transformed directly.

This is useful for cases where you need to sanitize or coerce certain values throughout a deeply nested dictionary structure. For example, the Flux CLI converts None values to 1 for treedict inputs that represent flags. This function can be used to replicate that behavior in the Python API.

Parameters:

Name	Type	Description	Default
obj_to_transform	Any	The object to transform. Can be a mapping or any other value.	required
transform	Callable	A callable that takes a single value and returns the transformed value.	required

Returns:

Type	Description
dict	A new mapping with the same structure as the input, but with all leaf values transformed.

Source code in maestrowf/utils.py

def coerce_dict_values(obj_to_transform: Any, transform: Callable)-> Dict:
    """
    Recursively apply a transformation function to all values in a nested mapping.

    Traverses the input object, recursively applying the given transformation function
    to each leaf value. If the input is a mapping (such as a dictionary), the function
    will process each value in the mapping. Non-mapping values are transformed directly.

    This is useful for cases where you need to sanitize or coerce certain values
    throughout a deeply nested dictionary structure. For example, the Flux CLI
    converts ``None`` values to ``1`` for treedict inputs that represent flags.
    This function can be used to replicate that behavior in the Python API.

    :param obj_to_transform: The object to transform. Can be a mapping or any other value.
    :type obj_to_transform: Any
    :param transform: A callable that takes a single value and returns the transformed value.
    :type transform: Callable
    :return: A new mapping with the same structure as the input, but with all leaf values transformed.
    :rtype: dict
    """
    if isinstance(obj_to_transform, Mapping):
        return {k: coerce_dict_values(v, transform)
                for k, v in obj_to_transform.items()}

    # Return transformed leaf
    return transform(obj_to_transform)

create_dictionary(list_keyvalues, token=':')

Create a dictionary from a list of key-value pairs.

Parameters:

Name	Type	Description	Default
list_keyvalues		List of token separates key-values.	required
token		The token to split each key-value by.	':'

Returns:

Type	Description
	A dictionary containing the key-value pairings in list_keyvalues.

Source code in maestrowf/utils.py

def create_dictionary(list_keyvalues, token=":"):
    """
    Create a dictionary from a list of key-value pairs.

    :param list_keyvalues: List of token separates key-values.
    :param token: The token to split each key-value by.
    :returns: A dictionary containing the key-value pairings in list_keyvalues.
    """
    _dict = {}
    for item in list_keyvalues:
        try:
            key, value = [i.strip() for i in item.split(token, 1)]
            _dict[key] = value
        except ValueError:
            msg = "'{}' is not capable of being split by the token '{}'. " \
                  "Verify that all other parameters are formatted properly." \
                  .format(item, token)
            LOGGER.exception(msg)
            raise ValueError(msg)

    return _dict

create_parentdir(path)

Recursively create parent directories.

Parameters:

Name	Type	Description	Default
path		Path to a directory to be created.	required

Source code in maestrowf/utils.py

def create_parentdir(path):
    """
    Recursively create parent directories.

    :param path: Path to a directory to be created.
    """
    if not os.path.exists(path):
        LOGGER.info("Directory does not exist. Creating directories to %s",
                    path)
        path = os.path.expanduser(path)
        os.makedirs(path)

csvtable_to_dict(fstream)

Convert a csv file stream into an in memory dictionary.

Parameters:

Name	Type	Description	Default
fstream		An open file stream to a csv table (with header)	required

Returns:

Type	Description
	A dictionary with a key for each column header and a list of column values for each key.

Source code in maestrowf/utils.py

def csvtable_to_dict(fstream):
    """
    Convert a csv file stream into an in memory dictionary.

    :param fstream: An open file stream to a csv table (with header)
    :returns: A dictionary with a key for each column header and a list of
        column values for each key.
    """
    # Read in the lines from the file stream.
    lines = fstream.readlines()
    # There are two pieces of information we need for the headers:
    # 1. The actual header titles.
    # 2. A map of index to header title
    _ = lines.pop(0).strip("\n").split(",")
    # Retain the order of the columns as they're added.
    table = OrderedDict()
    # A map of row index to the appropriate header.
    indices = {}
    i = 0
    # For each item in the header, mark its index and initialize its column.
    for item in _:
        indices[i] = item
        table[item] = []
        i += 1

    # Walk each line of the table, mapping the columns in the row to their key.
    for line in lines:
        # Split the csv row
        _ = line.split(",")
        # Walk each column and map it.
        for i in range(len(_)):
            table[indices[i]].append(_[i].strip("\n"))

    # Return the completed table
    return table

dict_to_dot_strings(dictionary, prefix='', result=None, print_none=True)

Helper to convert dictionaries into dot syntax strings, e.g. converting from maestro's yaml into flux's treedict style syntax for more concise batch directives.

Parameters:

Name	Type	Description	Default
dictionary	dict	Dictionary of args to flatten	required
prefix	str	Prefix to use at current level (used for recursion)	''
result	list | None	Accumulated result list (used for recursion)	None
print_none	bool	Flag to print =None literal if value=None, or ""	True

Returns:

Type	Description
list Example >>> dict_to_dot_strings({"foo": {"bar": 2, "foo2": 42}}) ['foo.bar=2', 'foo.foo2=42']	list of string views of dictionary in 'key.value' syntax

Source code in maestrowf/utils.py

def dict_to_dot_strings(dictionary, prefix='', result=None, print_none=True):
    """
    Helper to convert dictionaries into dot syntax strings, e.g. converting
    from maestro's yaml into flux's treedict style syntax for more concise
    batch directives.

    :param dictionary: Dictionary of args to flatten
    :type dictionary: dict
    :param prefix: Prefix to use at current level (used for recursion)
    :type prefix: str
    :param result: Accumulated result list (used for recursion)
    :type result: list or None
    :param print_none: Flag to print =None literal if value=None, or ""
    :type print_none: bool
    :returns: list of string views of dictionary in 'key.value' syntax
    :rtype: list

    Example
    >>> dict_to_dot_strings({"foo": {"bar": 2, "foo2": 42}})
    ['foo.bar=2', 'foo.foo2=42']
    """
    if result is None:
        result = []

    for key, value in dictionary.items():
        # Create the current key path
        current_key = f"{prefix}.{key}" if prefix else key

        if isinstance(value, dict):
            # Recursively process nested dictionaries
            dict_to_dot_strings(value, current_key, result, print_none)
        elif value is None and not print_none:
            result.append(f"{current_key}")
        else:
            # Add the terminal value and complete the string
            result.append(f"{current_key}={value}")

    return result

generate_filename(path, append_time=True)

Generate a non-conflicting file name.

Parameters:

Name	Type	Description	Default
path		Path to file.	required
append_time		Setting to append a timestamp.	True

Source code in maestrowf/utils.py

def generate_filename(path, append_time=True):
    """
    Generate a non-conflicting file name.

    :param path: Path to file.
    :param append_time: Setting to append a timestamp.
    """
    LOGGER.debug("Parameter path = %s", path)
    path = os.path.expanduser(path)
    root, ext = os.path.splitext(path)
    parent = os.path.dirname(root)
    fname = os.path.basename(root)

    LOGGER.debug("Expanded path = %s", path)
    LOGGER.debug("Root, Extension = (%s, %s)", root, ext)
    LOGGER.debug("Parent directory = %s", parent)
    LOGGER.debug("Filename = %s", fname)

    index = 0
    timestamp = ''
    if append_time:
        timestamp = '_{0}'.format(time.strftime("%Y%m%d-%H%M%S"))

    candidate = "{0}{1}{2}".format(fname, timestamp, ext)
    ls_files = set(os.listdir(parent))

    while candidate in ls_files:
        candidate = "{0}_{1:05d}{2}".format(root, index, ext)
        index += 1

    return os.path.join(parent, candidate)

get_duration(time_delta)

Convert durations to HH:MM:SS format.

Parameters:

Name	Type	Description	Default
time_delta		A time difference in datatime format.	required

Returns:

Type	Description
	A formatted string in HH:MM:SS

Source code in maestrowf/utils.py

def get_duration(time_delta):
    """
    Convert durations to HH:MM:SS format.

    :params time_delta: A time difference in datatime format.
    :returns: A formatted string in HH:MM:SS
    """
    duration = time_delta.total_seconds()
    days = int(duration / 86400)
    hours = int((duration % 86400) / 3600)
    minutes = int((duration % 86400 % 3600) / 60)
    seconds = int((duration % 86400 % 3600) % 60)

    return "{:d}d:{:02d}h:{:02d}m:{:02d}s" \
           .format(days, hours, minutes, seconds)

iter_dotpath_items(mapping, sep='.', prefix=None)

Yield (dotpath, value) pairs for all leaves in a nested mapping/dictionary. Leaves are any non-mapping objects.

:yields: Tuples of (dotpath, value) :ytype: tuple[str, any]

Parameters:

Name	Type	Description	Default
mapping	Mapping[str, Any]	Mapping to flatten into list of dotpath strings	required
sep	str	Optional separator used to join keys. Default = '.'	'.'

Source code in maestrowf/utils.py

def iter_dotpath_items(mapping: Mapping[str, Any], sep: str = ".", prefix: Optional[str] = None):
    """
    Yield (dotpath, value) pairs for all leaves in a nested mapping/dictionary.
    Leaves are any non-mapping objects.

    :param mapping: Mapping to flatten into list of dotpath strings
    :type mapping: collections.abc.Mapping
    :param sep: Optional separator used to join keys. Default = '.'
    :type sep: str
    :yields: Tuples of (dotpath, value)
    :ytype: tuple[str, any]
    """
    if not isinstance(mapping, Mapping):
        raise TypeError("iter_dotpath_items expects a mapping at the root")

    def join(base: str, key: str) -> str:
        """Helper to join two keys if non-terminal"""
        return f"{base}{sep}{key}" if base else key

    def walk(node: Any, base: str) -> Iterator[Tuple[str, Any]]:
        """Recursively walk the mapping, depth first"""
        if isinstance(node, Mapping):
            for k, v in node.items():
                ks = str(k)
                yield from walk(v, join(base, ks))

        else:
            # Node is a leaf (anything but a mapping)
            yield (base, node)

    root = prefix or ""
    yield from walk(mapping, root)

make_safe_path(base_path, *args)

Construct a subpath that is path safe.

Parameters:

Name	Type	Description	Default
base_path		The base path to append args to.	required
args		Path components to join into a path.	()

Returns:

Type	Description
	A joined subpath with invalid characters stripped.

Source code in maestrowf/utils.py

def make_safe_path(base_path, *args):
    """
    Construct a subpath that is path safe.

    :params base_path: The base path to append args to.
    :params args: Path components to join into a path.
    :returns: A joined subpath with invalid characters stripped.
    """
    valid = "-_.() {}{}".format(string.ascii_letters, string.digits)
    path = [base_path]
    for arg in args:
        arg = "".join(c for c in arg if c in valid)
        arg = arg.replace(" ", "_")
        path.append(arg)
    return os.path.join(*path)

parse_version(version_string)

Attempts using pep440 compliant version format and then falls back to a semver format to handle things like flux's version formats which can be a combination of the two.

Note: only major/minor/patch returned from semver parser; additional modifiers currently ignored for comparison purposes.

Parameters:

Name	Type	Description	Default
version_string		version string to parse	required

Returns:

Type	Description
	Version object, or None

Source code in maestrowf/utils.py

def parse_version(version_string):
    """
    Attempts using pep440 compliant version format and then falls back to a
    semver format to handle things like flux's version formats which can be a
    combination of the two.

    Note: only major/minor/patch returned from semver parser; additional
    modifiers currently ignored for comparison purposes.

    :param version_string: version string to parse
    :returns: Version object, or None
    """
    try:
        version = pkg_ver_parse(version_string)
        return version

    except InvalidVersion:
        LOGGER.info("Encountered version string '%s' that is not pep-440 compliant.  Attempting semver match instead.")

    match = _SEMVER_REGEX.match(version_string)
    if match:
        groups = match.groupdict()
        # NOTE: should we include prerelease if found here or some other pep
        # 440 ~equivalent? May be able to add prerelease, but not build
        return Version(f"{int(groups['major'])}.{int(groups['minor'])}.{int(groups['patch'])}")

    LOGGER.info("Could not parse version '%s'", version_string)

    raise InvalidVersion

ping_url(url)

Load a webpage to test that it is accessible.

Parameters:

Name	Type	Description	Default
url		URL string to be loaded.	required

Source code in maestrowf/utils.py

def ping_url(url):
    """
    Load a webpage to test that it is accessible.

    :param url: URL string to be loaded.
    """
    try:
        response = urlopen(url)
    except HTTPError as e:
        LOGGER.error("Error fulfilling HTTP request. (%s)", e.code)
        raise e
    except URLError as e:
        LOGGER.error(
            "Check specified URL (%s) and that you are connected to the "
            "internet. (%s)", url, e.code)
        raise e
    else:
        response.read()
        return

round_datetime_seconds(input_datetime)

Round datetime to the nearest whole second.

Solution referenced from: https://stackoverflow.com/questions/47792242/ rounding-time-off-to-the-nearest-second-python.

Parameters:

Name	Type	Description	Default
input_datetime		A datetime in datatime format.	required

Returns:

Type	Description
	input_datetime rounded to the nearest whole second

Source code in maestrowf/utils.py

def round_datetime_seconds(input_datetime):
    """
    Round datetime to the nearest whole second.

    Solution referenced from: https://stackoverflow.com/questions/47792242/
    rounding-time-off-to-the-nearest-second-python.

    :params input_datetime: A datetime in datatime format.
    :returns: ``input_datetime`` rounded to the nearest whole second
    """
    new_datetime = input_datetime

    if new_datetime.microsecond >= 500000:
        new_datetime = new_datetime + datetime.timedelta(seconds=1)

    return new_datetime.replace(microsecond=0)

start_process(cmd, cwd=None, env=None, shell=True)

Start a new process using a specified command.

Parameters:

Name	Type	Description	Default
cmd		A string or a list representing the command to be run.	required
cwd		Current working path that the process will be started in.	None
env		A dictionary containing the environment the process will use.	None
shell		Boolean that determines if the process will run a shell.	True

Source code in maestrowf/utils.py

def start_process(cmd, cwd=None, env=None, shell=True):
    """
    Start a new process using a specified command.

    :param cmd: A string or a list representing the command to be run.
    :param cwd: Current working path that the process will be started in.
    :param env: A dictionary containing the environment the process will use.
    :param shell: Boolean that determines if the process will run a shell.
    """
    if isinstance(cmd, list):
        shell = False

    # Define kwargs for the upcoming Popen call.
    kwargs = {
        "shell":                shell,
        "universal_newlines":   True,
        "stdout":               PIPE,
        "stderr":               PIPE,
    }

    # Individually check if cwd and env are set -- this prevents us from
    # adding parameters to the command that are only set to defaults. It
    # also insulates us from potential default value changes in the future.
    if cwd is not None:
        kwargs["cwd"] = cwd

    if env is not None:
        kwargs["env"] = env

    return Popen(cmd, **kwargs)

unflatten_dotpath_dict(dotpath_dict, sep='.')

Flatten a dictionary that may contain dotpath encoded keys into a pure nested dictionary.

Note this does not account for '=value' being in the strings as the source of this is expected to be dictionaries from the study spec in yaml, not a fully rendered cli directives.

Parameters:

Name	Type	Description	Default
dotpaths	Tuple[str, Any]	Tuple of dotpath string flattend dicts and values	required
sep	str	Separator used to split dotpath strings into nested keys. Default = "."	'.'

Returns:

Type	Description
Dict[str, Any]	nested dictionary view of the dotpath flattened input

Source code in maestrowf/utils.py

def unflatten_dotpath_dict(dotpath_dict: Dict[str, Any], sep: str = ".") -> Dict[str, Any]:
    """
    Flatten a dictionary that may contain dotpath encoded keys into a pure
    nested dictionary.

    Note this does not account for '=value' being in the strings as the source
    of this is expected to be dictionaries from the study spec in yaml, not a
    fully rendered cli directives.

    :param dotpaths: Tuple of dotpath string flattend dicts and values
    :type dotpaths: Tuple[str, Any]
    :param sep: Separator used to split dotpath strings into nested keys.
                Default = "."
    :type sep: str
    :returns: nested dictionary view of the dotpath flattened input    
    """
    root: Dict[str, Any] = {}
    for path, value in dotpath_dict.items():
        segments = str(path).split(sep) if path else [""]
        cursor: Dict[str, Any] = root
        for seg in segments[:-1]:
            if seg not in cursor or not isinstance(cursor[seg], dict):
                cursor[seg] = {}
            cursor = cursor[seg]

        cursor[segments[-1]] = value

    return root

unflatten_dotpath_tuples(dotpaths, sep='.')

Flatten a list of (dotpath_string, value) back into a nested dictionary.

Note this does not account for '=value' being in the strings as the source of this is expected to be dictionaries from the study spec in yaml, not a fully rendered cli directives.

Parameters:

Name	Type	Description	Default
dotpaths	List[Tuple[str, Any]]	Tuple of dotpath string flattend dicts and values	required
sep	str	Separator used to split dotpath strings into nested keys. Default = "."	'.'

Returns:

Type	Description
Dict[str, Any]	nested dictionary view of the dotpath flattened input

Source code in maestrowf/utils.py

def unflatten_dotpath_tuples(dotpaths: List[Tuple[str, Any]], sep: str = ".") -> Dict[str, Any]:
    """
    Flatten a list of (dotpath_string, value) back into a nested dictionary.

    Note this does not account for '=value' being in the strings as the source
    of this is expected to be dictionaries from the study spec in yaml, not a
    fully rendered cli directives.

    :param dotpaths: Tuple of dotpath string flattend dicts and values
    :type dotpaths: Tuple[str, Any]
    :param sep: Separator used to split dotpath strings into nested keys.
                Default = "."
    :type sep: str
    :returns: nested dictionary view of the dotpath flattened input

    """
    root: Dict[str, Any] = {}
    for path, value in dotpaths:
        segments = str(path).split(sep) if path else [""]
        cursor: Dict[str, Any] = root
        for seg in segments[:-1]:
            if seg not in cursor or not isinstance(cursor[seg], dict):
                cursor[seg] = {}
            cursor = cursor[seg]

        cursor[segments[-1]] = value

    return root

update_recursive(base_dict, update_dict)

Recursively update a dictionary with values from another dictionary.

For each key in update_dict, updates the corresponding value in base_dict. If both values for a given key are dictionaries, the update is performed recursively. Otherwise, the value from update_dict overwrites the value in base_dict.

Parameters:

Name	Type	Description	Default
base_dict	dict	The dictionary to update in-place.	required
update_dict	dict	The dictionary containing updates.	required

Returns:

Type	Description
dict .. note:: If a key exists in both dictionaries but the values are not both dicts, the value from ``update_dict`` will overwrite the value in ``base_dict``.	The updated base dictionary.

Source code in maestrowf/utils.py

def update_recursive(base_dict, update_dict):
    """
    Recursively update a dictionary with values from another dictionary.

    For each key in ``update_dict``, updates the corresponding value in ``base_dict``.
    If both values for a given key are dictionaries, the update is performed recursively.
    Otherwise, the value from ``update_dict`` overwrites the value in ``base_dict``.

    :param base_dict: The dictionary to update in-place.
    :type base_dict: dict
    :param update_dict: The dictionary containing updates.
    :type update_dict: dict
    :return: The updated base dictionary.
    :rtype: dict

    .. note::
        If a key exists in both dictionaries but the values are not both dicts,
        the value from ``update_dict`` will overwrite the value in ``base_dict``.
    """
    for k, v in update_dict.items():
        if k in base_dict and isinstance(base_dict[k], dict) and isinstance(v, dict):
            update_recursive(base_dict[k], v)
        else:
            base_dict[k] = v
    return base_dict