Skip to content

taps.run.utils

taps/run/utils.py

change_cwd() 

change_cwd(dest: Path | str) -> Generator[Path, None, None]

Context manager that changes the current working directory.

Parameters:

  • dest (Path | str) –

    Path to temporarily change the working directory to.

Yields:

  • Path –

    Destination directory as an absolute Path.

Source code in taps/run/utils.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@contextlib.contextmanager
def change_cwd(
    dest: pathlib.Path | str,
) -> Generator[pathlib.Path, None, None]:
    """Context manager that changes the current working directory.

    Args:
        dest: Path to temporarily change the working directory to.

    Yields:
        Destination directory as an absolute [`Path`][pathlib.Path].
    """
    origin = pathlib.Path.cwd().absolute()
    dest = pathlib.Path(dest).absolute()
    dest.mkdir(parents=True, exist_ok=True)
    os.chdir(dest)

    try:
        yield dest
    finally:
        os.chdir(origin)

flatten_mapping() 

flatten_mapping(
    mapping: MutableMapping[str, Any],
    parent_key: str = "",
    separator: str = ".",
) -> dict[str, Any]

Flatten the keys of nested mappings/dicts.

Parameters:

  • mapping (MutableMapping[str, Any]) –

    Nested mapping to flatten.

  • parent_key (str, default: '' ) –

    Prefix of parent keys when this function is called recursively.

  • separator (str, default: '.' ) –

    Separator between key parts.

Returns:

Source code in taps/run/utils.py 
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
def flatten_mapping(
    mapping: MutableMapping[str, Any],
    parent_key: str = '',
    separator: str = '.',
) -> dict[str, Any]:
    """Flatten the keys of nested mappings/dicts.

    Args:
        mapping: Nested mapping to flatten.
        parent_key: Prefix of parent keys when this function is called
            recursively.
        separator: Separator between key parts.

    Returns:
        Flattened dictionary.
    """
    items: list[tuple[str, Any]] = []

    for key, value in mapping.items():
        new_key = parent_key + separator + key if parent_key else key
        if isinstance(value, MutableMapping):
            items.extend(
                flatten_mapping(value, new_key, separator=separator).items(),
            )
        else:
            items.append((new_key, value))
    return dict(items)

prettify_mapping() 

prettify_mapping(
    mapping: Mapping[str, Any],
    level: int = 0,
    indent: int = 2,
) -> str

Turn a mapping into a nicely formatted string.

Example 
>>> from taps.run.utils import prettify_mapping
>>> data = {'a': {'b': [1, 2, 3], 'name': 'foo'}, 'c': 'baz', 'b': 'bar'}
>>> print(prettify_mapping(data))
a:
  name: 'foo'
  b: [1, 2, 3]
b: 'bar'
c: 'baz'
Source code in taps/run/utils.py 
 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
105
106
107
108
109
def prettify_mapping(
    mapping: Mapping[str, Any],
    level: int = 0,
    indent: int = 2,
) -> str:
    """Turn a mapping into a nicely formatted string.

    Example:
        ```python
        >>> from taps.run.utils import prettify_mapping
        >>> data = {'a': {'b': [1, 2, 3], 'name': 'foo'}, 'c': 'baz', 'b': 'bar'}
        >>> print(prettify_mapping(data))
        a:
          name: 'foo'
          b: [1, 2, 3]
        b: 'bar'
        c: 'baz'
        ```
    """  # noqa: E501
    lines: list[str] = []
    space = ' ' * indent * level

    keys = sorted(mapping.keys())

    # Move the 'name' key to the front
    if 'name' in keys:
        keys.remove('name')
        keys = ['name', *keys]

    for key in keys:
        value = mapping[key]

        if isinstance(value, Mapping):
            lines.append(f'{space}{key}:')
            if len(value) > 0:
                lines.append(prettify_mapping(value, level + 1, indent))
        else:
            lines.append(f'{space}{key}: {value!r}')

    return '\n'.join(lines)

prettify_validation_error() 

prettify_validation_error(
    error: ValidationError,
    model: type[BaseModel] | None = None,
) -> ValueError

Parse a Pydantic validation error into a ValueError.

Given a ValidationError, 

pydantic_core._pydantic_core.ValidationError: 2 validation errors for GeneratedConfig
app.matrix_size
  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='100x', input_type=str]
    For further information visit https://errors.pydantic.dev/2.7/v/int_parsing
engine.executor.max_threads
  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='1.5', input_type=str]
    For further information visit https://errors.pydantic.dev/2.7/v/int_parsing
returns a ValueError with a more readable output. 
ValueError: Found 2 validation errors:
  - Input should be a valid integer, unable to parse string as an integer
    Attribute: app.matrix_size
    Input (str): '100x'
    Error type: int_parsing (https://errors.pydantic.dev/2.7/v/int_parsing)
  - Input should be a valid integer, unable to parse string as an integer
    Attribute: engine.executor.max_threads
    Input (str): '1.5'
    Error type: int_parsing (https://errors.pydantic.dev/2.7/v/int_parsing)

Source code in taps/run/utils.py 
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def prettify_validation_error(
    error: ValidationError,
    model: type[BaseModel] | None = None,
) -> ValueError:
    """Parse a Pydantic validation error into a ValueError.

    Given a [`ValidationError`][pydantic_core.ValidationError],
    ```
    pydantic_core._pydantic_core.ValidationError: 2 validation errors for GeneratedConfig
    app.matrix_size
      Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='100x', input_type=str]
        For further information visit https://errors.pydantic.dev/2.7/v/int_parsing
    engine.executor.max_threads
      Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='1.5', input_type=str]
        For further information visit https://errors.pydantic.dev/2.7/v/int_parsing
    ```
    returns a [`ValueError`][ValueError] with a more readable output.
    ```
    ValueError: Found 2 validation errors:
      - Input should be a valid integer, unable to parse string as an integer
        Attribute: app.matrix_size
        Input (str): '100x'
        Error type: int_parsing (https://errors.pydantic.dev/2.7/v/int_parsing)
      - Input should be a valid integer, unable to parse string as an integer
        Attribute: engine.executor.max_threads
        Input (str): '1.5'
        Error type: int_parsing (https://errors.pydantic.dev/2.7/v/int_parsing)
    ```
    """  # noqa: E501
    errors: list[str] = []

    for e in error.errors():
        attribute = '.'.join(str(v) for v in e['loc'])
        input_ = e['input']
        message = e['msg']
        type_ = e['type']
        url = e.get(
            'url',
            'https://docs.pydantic.dev/latest/errors/validation_errors/',
        )

        errors.append(f"""\
  - {message}
    Attribute: {attribute}
    Input ({type(input_).__name__}): {input_!r}
    Error type: {type_} ({url})\
""")

    errors_str = '\n'.join(errors)
    count = error.error_count()

    if model is not None:
        model_str = f' for {model.__module__}.{model.__name__}'
    else:
        model_str = ''

    return ValueError(f"""\
Found {count} validation error{"" if count == 1 else "s"}{model_str}
{errors_str}\
""")

update_environment() 

update_environment(
    variables: Mapping[str, str]
) -> Generator[None, None, None]

Context manager that updates environment variables.

Parameters:

  • variables (Mapping[str, str]) –

    Mapping of environment variable name to value to temporarily set.

Source code in taps/run/utils.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@contextlib.contextmanager
def update_environment(
    variables: Mapping[str, str],
) -> Generator[None, None, None]:
    """Context manager that updates environment variables.

    Args:
        variables: Mapping of environment variable name to value to
            temporarily set.
    """
    previous = {
        name: value for name, value in os.environ.items() if name in variables
    }
    os.environ.update(variables)
    logger.debug(
        f'Updated {len(variables)} environment variable(s) '
        f'({", ".join(variables.keys())})',
    )
    try:
        yield
    finally:
        # Remove the added variables then restore any old variables
        for name in variables:
            os.environ.pop(name)
        os.environ.update(previous)
        if len(previous) > 0:
            logger.debug(
                f'Restored {len(previous)} environment variable(s) '
                f'({",".join(previous.keys())})',
            )