Source code for tamr_toolbox.utils.config

"""Tasks related to loading and using configuration files"""
from typing import Union, Optional
from pathlib import Path

import yaml
import logging
import re

import os

from tamr_toolbox.models.data_type import JsonDict

LOGGER = logging.getLogger(__name__)

# Define pattern to be used for parsing environment variables in config files
# This pattern searches for the '$' character at the start of variable name and then retrieves
# all other text in the variable name. For example: '$MY_VAR' matches and returns 'MY_VAR'
ENV_VAR_PATTERN = re.compile(r"^\$(.*)$")

def _yaml_env_variable_constructor(loader, node):
    """Replaces values that match the ENV_VAR_PATTERN with
    the value of the corresponding environment variable"""
    value = loader.construct_scalar(node)
    env_match = ENV_VAR_PATTERN.match(value)
    if env_match:
        value = env_match.groups()[0]
        env_var = os.environ.get(value)

        if env_var is None:
            LOGGER.warning(f"No environment variable for {value} found, using None.")
        return env_var
    return value

def _yaml_env_loader(path_to_file: Optional[Union[str, Path]]) -> JsonDict:
    """Reads a yaml file and creates a dictionary, retrieving environment variables as needed

        path_to_file: Path to config yaml file

        All configuration variables in a dictionary

    yaml_loader = yaml.SafeLoader
    yaml_loader.add_implicit_resolver("!env_var", ENV_VAR_PATTERN, None)
    yaml_loader.add_constructor("!env_var", _yaml_env_variable_constructor)

    with open(path_to_file, "r") as config_file:
        configs = yaml.load(, Loader=yaml_loader)"Configurations have been loaded from {path_to_file}")
    return configs

[docs]def from_yaml( path_to_file: Optional[Union[str, Path]], *, default_path_to_file: Optional[Union[str, Path]] = None, ) -> JsonDict: """Reads a yaml file and creates a dictionary. Input values can be retrieved from environment variables Args: path_to_file: Path to config yaml file default_path_to_file: Path to use if path_to_file is null or empty Returns: All configuration variables in a dictionary """ # If path_to_file is not populated, use default_path_to_file instead if path_to_file is None or (isinstance(path_to_file, str) and len(path_to_file) == 0): if default_path_to_file is None or ( isinstance(default_path_to_file, str) and len(default_path_to_file) == 0 ): not_populated_error = ( f"No target path or default path has been specified. At least one must have a " f"value not: 'path_to_file'='{path_to_file}', " f"'default_path_to_file'='{default_path_to_file}'" ) LOGGER.error(not_populated_error) raise ValueError(not_populated_error) else:"Reading configuration file from default: {default_path_to_file}") path_to_file = default_path_to_file return _yaml_env_loader(path_to_file)