|
5 | 5 | from copy import deepcopy |
6 | 6 | from dataclasses import InitVar, dataclass, field |
7 | 7 | from itertools import product |
8 | | -from typing import Any, Dict, Iterable, List, Mapping, Optional, Union |
| 8 | +from typing import Any, Dict, Iterable, List, Mapping, Optional, Union, Tuple |
9 | 9 |
|
10 | 10 | import dpath |
11 | 11 | from typing_extensions import deprecated |
@@ -94,25 +94,22 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None: |
94 | 94 | ) |
95 | 95 |
|
96 | 96 | @property |
97 | | - def _stream_config(self): |
98 | | - def resolve_path(pointer): |
| 97 | + def _stream_config(self) -> List[Dict[str, Any]]: |
| 98 | + def resolve_path(pointer: List[Union[InterpolatedString, str]]) -> List[str]: |
99 | 99 | return [ |
100 | 100 | node.eval(self.config) if not isinstance(node, str) else node for node in pointer |
101 | 101 | ] |
102 | 102 |
|
103 | | - def normalize_configs(configs): |
104 | | - return configs if isinstance(configs, list) else [configs] |
105 | | - |
106 | | - def prepare_streams(): |
| 103 | + def prepare_streams() -> Iterable[List[Tuple[int, Any]]]: |
107 | 104 | for stream_config in self.stream_configs: |
108 | 105 | path = resolve_path(stream_config.configs_pointer) |
109 | 106 | stream_configs = dpath.get(dict(self.config), path, default=[]) |
110 | | - stream_configs = normalize_configs(stream_configs) |
| 107 | + stream_configs = stream_configs if isinstance(stream_configs, list) else [stream_configs] |
111 | 108 | if stream_config.default_values: |
112 | 109 | stream_configs.extend(stream_config.default_values) |
113 | 110 | yield [(i, item) for i, item in enumerate(stream_configs)] |
114 | 111 |
|
115 | | - def merge_combination(combo): |
| 112 | + def merge_combination(combo: Iterable[Tuple[int, Any]]) -> Dict[str, Any]: |
116 | 113 | result = {} |
117 | 114 | for config_index, (elem_index, elem) in enumerate(combo): |
118 | 115 | if isinstance(elem, dict): |
|
0 commit comments