pyantz.infrastructure.config.variables package

Module contents

Fill in variables in the parameters.

Strings in a job’s parameters may refer to variables as %{name}. Variables may refer to other variables, up to MAX_RESOLUTION_DEPTH levels deep.

MAX_RESOLUTION_DEPTH = 32

Maximum nesting of variables referring to other variables.

VARIABLE_REGEX = re.compile('(%{[^%{}]+})')

Matches a reference to a variable, such as %{name}.

_SEQUENCE_AND_SET_TYPES = (<class 'list'>, <class 'tuple'>, <class 'set'>, <class 'frozenset'>)

Containers whose items are searched for variables.

_rebuild(original: Iterable[Any], items: list[Any])

Make a container like original from items.

Parameters:

• original (Iterable[Any]) – the container the items came from.

• items (list[Any]) – the new items.

Returns:

a container of the type of original holding items, or of its base type (list, tuple, set or frozenset) if that type can’t be built from a list, such as a namedtuple.

Return type:

Any

Raises:

StopIteration – if original isn’t a list, tuple, set or frozenset.

Example

from collections import namedtuple

from pyantz.infrastructure.config.variables import _rebuild

Point = namedtuple("Point", "x y")
print(_rebuild(frozenset(), [1]), _rebuild(Point(0, 0), [1, 2]))

frozenset({1}) (1, 2)

_resolve_var_str(v: str, variables: Mapping[str, Any])

Resolve the string by replacing variables.

If the whole string is exactly one reference ("%{n}") to a non-string variable, the variable’s value is returned as is, preserving its type. Otherwise, values are converted to strings and interpolated.

Parameters:

• v (str) – the string to resolve.

• variables (Mapping[str, Any]) – the variables to fill in.

Returns:

the string with variables interleaved (or the value of a whole reference) AND a boolean. If the boolean is true, then a variable substring was found but could not be replaced because the variable isn’t set.

Return type:

tuple[Any, bool]

Raises:

ValueError – if resolving needs more than MAX_RESOLUTION_DEPTH passes, i.e., a variable refers to itself.

Example

from pyantz.infrastructure.config.variables import _resolve_var_str

print(_resolve_var_str("%{n}", {"n": 5}))
print(_resolve_var_str("%{a}-%{b}", {"a": 1, "b": "x"}))
print(_resolve_var_str("%{a}-%{unset}", {"a": 1}))
try:
    _resolve_var_str("%{a}", {"a": "%{b}", "b": "%{a}"})
except ValueError as exc:
    print(exc)

(5, False)
('1-x', False)
('1-%{unset}', True)
Recursive variable reference while resolving '%{a}'

resolve_parameters(job_parameters: Mapping[str, Any], variables: Mapping[str, Any])

Resolve any variables in the job_parameters.

Variables are demarked with %{VARIABLE_NAME}

Parameters:

• job_parameters (Mapping[str, Any]) – parameters to be passed to the function

• variables (Mapping[str, Any]) – variables to put in the parameters

Returns:

parameters with the variables eagerly replaced, and True if some variables were detected and not replaced

Return type:

tuple[Mapping[str, Any], bool]

Raises:

ValueError – if a variable refers to itself, directly or not.

Example

from pyantz.infrastructure.config.variables import resolve_parameters

parameters, unresolved = resolve_parameters(
    {"path": "%{root}/%{name}.txt", "count": "%{n}", "other": "%{unset}"},
    {"root": "/data", "name": "run_%{n}", "n": 3},
)
print(parameters, unresolved)

{'path': '/data/run_3.txt', 'count': 3, 'other': '%{unset}'} True

resolve_var_any[T](some_val: T, variables: Mapping[str, Any])

Resolve the variable and return it.

Strings are resolved, and mappings, lists, tuples and sets are searched at any depth; anything else is returned unchanged.

Parameters:

• some_val (T) – the value to resolve.

• variables (Mapping[str, Any]) – the variables to fill in.

Returns:

value with variables interleaved and a boolean. if the boolean is true, a variable was detected but left unchanged.

Return type:

tuple[T, bool]

Raises:

ValueError – if a variable refers to itself, directly or not.

Example

from pyantz.infrastructure.config.variables import resolve_var_any

print(resolve_var_any(("%{a}", ["%{a}-%{b}"], 7), {"a": 1, "b": "x"}))

((1, ['1-x'], 7), False)