Skip to content

yaml_utils

YAML Utilities for Camply

read_yaml(path=None) #

Read a YAML File

Load a yaml configuration file_path (path) or data object (data) and resolve any environment variables. The environment variables must be in this format to be parsed: ${VAR_NAME}.

Parameters:

Name Type Description Default
path Optional[str]

File Path of YAML Object to Read

 None

Examples:

database: host: ${HOST} port: ${PORT} ${KEY}: ${VALUE} app: log_path: "/var/${LOG_PATH}" something_else: "${AWESOME_ENV_VAR}/var/${A_SECOND_AWESOME_VAR}"

Source code in camply/utils/yaml_utils.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def read_yaml(path: Optional[str] = None):
    """
    Read a YAML File

    Load a yaml configuration file_path (path) or data object (data)
    and resolve any environment variables. The environment
    variables must be in this format to be parsed: ${VAR_NAME}.

    Parameters
    ----------
    path: Optional[str]
        File Path of YAML Object to Read

    Examples
    --------
    database:
        host: ${HOST}
        port: ${PORT}
        ${KEY}: ${VALUE}
    app:
        log_path: "/var/${LOG_PATH}"
        something_else: "${AWESOME_ENV_VAR}/var/${A_SECOND_AWESOME_VAR}"
    """
    path = os.path.abspath(path)
    pattern = compile(r".*?\${(\w+)}.*?")

    safe_loader = SafeLoader
    safe_loader.add_implicit_resolver(tag=None, regexp=pattern, first=None)

    def env_var_constructor(safe_loader: yaml.Loader, node: Any) -> Any:
        """
        Extracts the environment variable from the node's value

        Parameters
        ----------
        safe_loader: yaml.Loader
        node: Any
            The current node in the yaml

        Returns
        -------
        Any
            the parsed string that contains the value of the environment variable
        """
        value = safe_loader.construct_scalar(node=node)
        match = pattern.findall(string=value)
        if match:
            full_value = value
            for item in match:
                full_value = full_value.replace(
                    "${{{key}}}".format(key=item), os.getenv(key=item, default=item)
                )
            return full_value
        return value

    safe_loader.add_constructor(tag=None, constructor=env_var_constructor)
    with open(path) as conf_data:
        return load(stream=conf_data, Loader=safe_loader)

yaml_file_to_arguments(file_path) #

Convert YAML File into A Dictionary to be used as **kwargs

Parameters:

Name Type Description Default
file_path str

File Path to YAML

 required

Returns:

Type Description
Tuple[str, Dict[str, object], Dict[str, object]]

Tuple containing provider string, provider kwargs, and search kwargs
Source code in camply/utils/yaml_utils.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def yaml_file_to_arguments(
    file_path: str,
) -> Tuple[str, Dict[str, object], Dict[str, object]]:
    """
    Convert YAML File into A Dictionary to be used as **kwargs

    Parameters
    ----------
    file_path: str
        File Path to YAML

    Returns
    -------
    Tuple[str, Dict[str, object], Dict[str, object]]
        Tuple containing provider string, provider **kwargs, and search **kwargs
    """
    yaml_search = read_yaml(path=file_path)
    logger.info(f"YAML File Parsed: {Path(file_path).name}")
    yaml_model = YamlSearchFile(**yaml_search)
    if isinstance(yaml_model.provider, Enum):
        provider = yaml_model.provider.value
    else:
        provider = yaml_model.provider
    search_window = handle_search_windows(
        start_date=yaml_model.start_date, end_date=yaml_model.end_date
    )
    days_of_the_week = yaml_model.days
    if days_of_the_week is not None:
        lower_mapping = {
            key.lower(): value for key, value in days_of_the_week_mapping.items()
        }
        days_of_the_week = [lower_mapping[item.lower()] for item in days_of_the_week]
    equipment = make_list(yaml_model.equipment)
    if isinstance(equipment, list):
        equipment = [tuple(equip) for equip in equipment]
    provider_kwargs = {
        "search_window": search_window,
        "recreation_area": yaml_model.recreation_area,
        "campgrounds": yaml_model.campgrounds,
        "campsites": yaml_model.campsites,
        "weekends_only": yaml_model.weekends,
        "days_of_the_week": days_of_the_week,
        "nights": yaml_model.nights,
        "equipment": equipment,
        "offline_search": yaml_model.offline_search,
        "offline_search_path": yaml_model.offline_search_path,
    }
    search_kwargs = {
        "log": True,
        "verbose": True,
        "continuous": yaml_model.continuous,
        "polling_interval": yaml_model.polling_interval,
        "notify_first_try": yaml_model.notify_first_try,
        "notification_provider": yaml_model.notifications,
        "search_forever": yaml_model.search_forever,
        "search_once": yaml_model.search_once,
    }
    return provider, provider_kwargs, search_kwargs