Changeset View
Changeset View
Standalone View
Standalone View
python_modules/dagster/dagster/utils/merger.py
| import copy | import copy | ||||
| from typing import Mapping | |||||
| from dagster import check | from dagster import check | ||||
| def _deep_merge_dicts(onto_dict, from_dict): | def _deep_merge_dicts(onto_dict: Mapping, from_dict: Mapping) -> dict: | ||||
| check.dict_param(from_dict, "from_dict") | check.dict_param(from_dict, "from_dict") | ||||
| check.dict_param(onto_dict, "onto_dict") | check.dict_param(onto_dict, "onto_dict") | ||||
| for from_key, from_value in from_dict.items(): | for from_key, from_value in from_dict.items(): | ||||
| if from_key not in onto_dict: | if from_key not in onto_dict: | ||||
| onto_dict[from_key] = from_value | onto_dict[from_key] = from_value | ||||
| else: | else: | ||||
| onto_value = onto_dict[from_key] | onto_value = onto_dict[from_key] | ||||
| if isinstance(from_value, dict) and isinstance(onto_value, dict): | if isinstance(from_value, dict) and isinstance(onto_value, dict): | ||||
| onto_dict[from_key] = _deep_merge_dicts(onto_value, from_value) | onto_dict[from_key] = _deep_merge_dicts(onto_value, from_value) | ||||
| else: | else: | ||||
| onto_dict[from_key] = from_value # smash | onto_dict[from_key] = from_value # smash | ||||
| return onto_dict | return onto_dict | ||||
| def deep_merge_dicts(onto_dict: dict, from_dict: dict) -> dict: | def deep_merge_dicts(onto_dict: Mapping, from_dict: Mapping) -> dict: | ||||
| """ | """ | ||||
| Returns a recursive union of two input dictionaries: | Returns a recursive union of two input dictionaries: | ||||
| * The returned dictionary has an entry for any key that's in either of the inputs. | * The returned dictionary has an entry for any key that's in either of the inputs. | ||||
| * For any key whose value is a dictionary in both of the inputs, the returned value will | * For any key whose value is a dictionary in both of the inputs, the returned value will | ||||
| be the result of deep-merging the two input sub-dictionaries. | be the result of deep-merging the two input sub-dictionaries. | ||||
| If from_dict and onto_dict have different values for the same key, and the values are not both | If from_dict and onto_dict have different values for the same key, and the values are not both | ||||
| dictionaries, the returned dictionary contains the value from from_dict. | dictionaries, the returned dictionary contains the value from from_dict. | ||||
| """ | """ | ||||
| onto_dict = copy.deepcopy(onto_dict) | onto_dict = copy.deepcopy(onto_dict) | ||||
| return _deep_merge_dicts(onto_dict, from_dict) | return _deep_merge_dicts(onto_dict, from_dict) | ||||
| def merge_dicts(*args) -> dict: | def merge_dicts(*args: Mapping) -> dict: | ||||
| """ | |||||
| Returns a dictionary with with all the keys in all of the input dictionaries. | |||||
| If multiple input dictionaries have different values for the same key, the returned dictionary | |||||
| contains the value from the dictionary that comes latest in the list. | |||||
| """ | |||||
| check.is_tuple(args, of_type=dict) | check.is_tuple(args, of_type=dict) | ||||
| if len(args) < 2: | if len(args) < 2: | ||||
| check.failed(f"Expected 2 or more args to merge_dicts, found {len(args)}") | check.failed(f"Expected 2 or more args to merge_dicts, found {len(args)}") | ||||
| result = args[0].copy() | result = args[0].copy() | ||||
| for arg in args[1:]: | for arg in args[1:]: | ||||
| result.update(arg) | result.update(arg) | ||||
| return result | return result | ||||