(
per_key_opts, # type: Dict[str, Any]
opt_type, # type: OptionsBase
valid_opt_keys, # type: List[str]
durability_type # type: Optional[Union[Type[Dict], Type[int]]]
)
| 147 | |
| 148 | |
| 149 | def _get_per_key_opts( |
| 150 | per_key_opts, # type: Dict[str, Any] |
| 151 | opt_type, # type: OptionsBase |
| 152 | valid_opt_keys, # type: List[str] |
| 153 | durability_type # type: Optional[Union[Type[Dict], Type[int]]] |
| 154 | ) -> Dict[str, Any]: |
| 155 | final_key_opts = {} |
| 156 | for key, opts in per_key_opts.items(): |
| 157 | if not isinstance(opts, (opt_type, dict)): |
| 158 | raise InvalidArgumentException(message=f'Expected options to be of type Union[{opt_type.__name__}, dict]') |
| 159 | key_opts = {} |
| 160 | for opt_key, opt_value in opts.items(): |
| 161 | if opt_key not in valid_opt_keys: |
| 162 | continue |
| 163 | transform = VALID_MULTI_OPTS.get(opt_key, None) |
| 164 | if transform: |
| 165 | transformed_value = transform(opt_value) |
| 166 | if opt_key == 'durability': |
| 167 | if durability_type is None: |
| 168 | durability_type = dict if isinstance(transformed_value, dict) else int |
| 169 | if not isinstance(transformed_value, durability_type): |
| 170 | raise InvalidArgumentException('Durability must be either client or server for all operations.') |
| 171 | if isinstance(transformed_value, dict): |
| 172 | key_opts['persist_to'] = transformed_value['persist_to'] |
| 173 | key_opts['replicate_to'] = transformed_value['replicate_to'] |
| 174 | else: |
| 175 | key_opts['durability_level'] = transformed_value |
| 176 | else: |
| 177 | key_opts[opt_key] = transformed_value |
| 178 | |
| 179 | final_key_opts[key] = key_opts |
| 180 | |
| 181 | return final_key_opts |
| 182 | |
| 183 | |
| 184 | def get_valid_multi_args( |
no test coverage detected