Skip to content

env_tools

CycleError

Bases: ValueError

Raised when a cycle is detected in dynamic env variables compute.

Source code in client/ayon_core/lib/env_tools.py
17
18
19
class CycleError(ValueError):
    """Raised when a cycle is detected in dynamic env variables compute."""
    pass

DynamicKeyClashError

Bases: Exception

Raised when dynamic key clashes with an existing key.

Source code in client/ayon_core/lib/env_tools.py
22
23
24
class DynamicKeyClashError(Exception):
    """Raised when dynamic key clashes with an existing key."""
    pass

compute_env_variables_structure(env, fill_dynamic_keys=True)

Compute the result from recursive dynamic environment.

Keys that are not present in the data will remain unformatted as the

original keys. So they can be formatted against the current user environment when merging. So {"A": "{key}"} will remain {key} if not present in the dynamic environment.

Source code in client/ayon_core/lib/env_tools.py
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def compute_env_variables_structure(
    env: dict[str, str],
    fill_dynamic_keys: bool = True,
) -> dict[str, str]:
    """Compute the result from recursive dynamic environment.

    Note: Keys that are not present in the data will remain unformatted as the
        original keys. So they can be formatted against the current user
        environment when merging. So {"A": "{key}"} will remain {key} if not
        present in the dynamic environment.

    """
    env = env.copy()

    # Collect dependencies
    dependencies = collections.defaultdict(set)
    for key, value in env.items():
        dependent_keys = re.findall("{(.+?)}", value)
        for dependent_key in dependent_keys:
            # Ignore reference to itself or key is not in env
            if dependent_key != key and dependent_key in env:
                dependencies[key].add(dependent_key)

    ordered, cyclic = _topological_sort(dependencies)

    # Check cycle
    if cyclic:
        raise CycleError(f"A cycle is detected on: {cyclic}")

    # Format dynamic values
    for key in reversed(ordered):
        if key in env:
            if not isinstance(env[key], str):
                continue
            data = env.copy()
            data.pop(key)    # format without itself
            env[key] = _partial_format(env[key], data=data)

    # Format dynamic keys
    if fill_dynamic_keys:
        formatted = {}
        for key, value in env.items():
            if not isinstance(value, str):
                formatted[key] = value
                continue

            new_key = _partial_format(key, data=env)
            if new_key in formatted:
                raise DynamicKeyClashError(
                    f"Key clashes on: {new_key} (source: {key})"
                )

            formatted[new_key] = value
        env = formatted

    return env

env_value_to_bool(env_key=None, value=None, default=False)

Convert environment variable value to boolean.

Function is based on value of the environemt variable. Value is lowered so function is not case sensitive.

Returns:

Name Type Description
bool bool

If value match to one of ["true", "yes", "1"] result if True but if value match to ["false", "no", "0"] result is False else default value is returned.

Source code in client/ayon_core/lib/env_tools.py
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
def env_value_to_bool(
    env_key: Optional[str] = None,
    value: Optional[str] = None,
    default: bool = False,
) -> bool:
    """Convert environment variable value to boolean.

    Function is based on value of the environemt variable. Value is lowered
    so function is not case sensitive.

    Returns:
        bool: If value match to one of ["true", "yes", "1"] result if True
            but if value match to ["false", "no", "0"] result is False else
            default value is returned.

    """
    if value is None and env_key is None:
        return default

    if value is None:
        value = os.environ.get(env_key)

    if value is not None:
        value = str(value).lower()
        if value in ("true", "yes", "1", "on"):
            return True
        elif value in ("false", "no", "0", "off"):
            return False
    return default

get_paths_from_environ(env_key=None, env_value=None, return_first=False)

Return existing paths from specific environment variable.

Parameters:

Name Type Description Default
env_key Optional[str]

Environment key where should look for paths.

None
env_value Optional[str]

Value of environment variable. Argument env_key is skipped if this argument is entered.

None
return_first bool

Return first found value or return list of found paths. None or empty list returned if nothing found.

False

Returns:

Type Description
Optional[Union[str, list[str]]]

Optional[Union[str, list[str]]]: Result of found path/s.

Source code in client/ayon_core/lib/env_tools.py
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_paths_from_environ(
    env_key: Optional[str] = None,
    env_value: Optional[str] = None,
    return_first: bool = False,
) -> Optional[Union[str, list[str]]]:
    """Return existing paths from specific environment variable.

    Args:
        env_key (Optional[str]): Environment key where should look for paths.
        env_value (Optional[str]): Value of environment variable.
            Argument `env_key` is skipped if this argument is entered.
        return_first (bool): Return first found value or return list of found
            paths. `None` or empty list returned if nothing found.

    Returns:
        Optional[Union[str, list[str]]]: Result of found path/s.

    """
    existing_paths = []
    if not env_key and not env_value:
        if return_first:
            return None
        return existing_paths

    if env_value is None:
        env_value = os.environ.get(env_key) or ""

    path_items = env_value.split(os.pathsep)
    for path in path_items:
        # Skip empty string
        if not path:
            continue
        # Normalize path
        path = os.path.normpath(path)
        # Check if path exists
        if os.path.exists(path):
            # Return path if `return_first` is set to True
            if return_first:
                return path
            # Store path
            existing_paths.append(path)

    # Return None if none of paths exists
    if return_first:
        return None
    # Return all existing paths from environment variable
    return existing_paths

merge_env_variables(src_env, dst_env, missing_template=None)

Merge the tools environment with the 'current_env'.

This finalizes the join with a current environment by formatting the remainder of dynamic variables with that from the current environment.

Remaining missing variables result in an empty value.

Parameters:

Name Type Description Default
src_env dict

The dynamic environment

required
dst_env dict

The target environment variables mapping to merge the dynamic environment into.

required
missing_template str

Argument passed to '_partial_format' during merging. None should keep missing keys unchanged.

None

Returns:

Type Description
dict[str, str]

dict[str, str]: The resulting environment after the merge.

Source code in client/ayon_core/lib/env_tools.py
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def merge_env_variables(
    src_env: dict[str, str],
    dst_env: dict[str, str],
    missing_template: Optional[str] = None,
) -> dict[str, str]:
    """Merge the tools environment with the 'current_env'.

    This finalizes the join with a current environment by formatting the
    remainder of dynamic variables with that from the current environment.

    Remaining missing variables result in an empty value.

    Args:
        src_env (dict): The dynamic environment
        dst_env (dict): The target environment variables mapping to merge
            the dynamic environment into.
        missing_template (str): Argument passed to '_partial_format' during
            merging. `None` should keep missing keys unchanged.

    Returns:
        dict[str, str]: The resulting environment after the merge.

    """
    result = dst_env.copy()
    for key, value in src_env.items():
        result[key] = _partial_format(
            str(value), dst_env, missing_template
        )

    return result

parse_env_variables_structure(env, platform_name=None)

Parse environment for platform-specific values and paths as lists.

Parameters:

Name Type Description Default
env dict

The source environment to read.

required
platform_name Optional[PlatformName]

Name of platform to parse for. Defaults to current platform.

None

Returns:

Name Type Description
dict dict[str, str]

The flattened environment for a platform.

Source code in client/ayon_core/lib/env_tools.py
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
140
141
142
143
144
145
146
def parse_env_variables_structure(
    env: dict[str, EnvValue],
    platform_name: Optional[PlatformName] = None
) -> dict[str, str]:
    """Parse environment for platform-specific values and paths as lists.

    Args:
        env (dict): The source environment to read.
        platform_name (Optional[PlatformName]): Name of platform to parse for.
            Defaults to current platform.

    Returns:
        dict: The flattened environment for a platform.

    """
    if platform_name is None:
        platform_name = platform.system().lower()

    # Separator based on OS 'os.pathsep' is ';' on Windows and ':' on Unix
    sep = ";" if platform_name == "windows" else ":"

    result = {}
    for variable, value in env.items():
        # Platform specific values
        if isinstance(value, dict):
            value = value.get(platform_name)

        # Allow to have lists as values in the tool data
        if isinstance(value, (list, tuple)):
            value = sep.join(value)

        if not value:
            continue

        if not isinstance(value, str):
            raise TypeError(f"Expected 'str' got '{type(value)}'")

        result[variable] = value

    return result