|
| 1 | +#!/usr/bin/env python |
| 2 | +# -*- coding: utf-8 -*-- |
| 3 | + |
| 4 | +# Copyright (c) 2023 Oracle and/or its affiliates. |
| 5 | +# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/ |
| 6 | + |
| 7 | +import copy |
| 8 | +from typing import Any, Dict, List |
| 9 | + |
| 10 | + |
| 11 | +class DictionaryMerger: |
| 12 | + """ |
| 13 | + A class to update dictionary values for specified keys and |
| 14 | + then merge these updates back into the original dictionary. |
| 15 | +
|
| 16 | + Example |
| 17 | + ------- |
| 18 | + >>> updates = { |
| 19 | + ... "infrastructure.blockStorageSize": "20", |
| 20 | + ... "infrastructure.projectId": "my_new_project_id" |
| 21 | + ... "runtime.conda": "my_conda" |
| 22 | + ... } |
| 23 | + >>> updater = DictionaryMerger(updates) |
| 24 | + >>> source_data = { |
| 25 | + ... "infrastructure": { |
| 26 | + ... "blockStorageSize": "10", |
| 27 | + ... "projectId": "old_project_id", |
| 28 | + ... }, |
| 29 | + ... "runtime": { |
| 30 | + ... "conda": "conda", |
| 31 | + ... }, |
| 32 | + ... } |
| 33 | + >>> result = updater.dispatch(source_data) |
| 34 | + ... { |
| 35 | + ... "infrastructure": { |
| 36 | + ... "blockStorageSize": "20", |
| 37 | + ... "projectId": "my_new_project_id", |
| 38 | + ... }, |
| 39 | + ... "runtime": { |
| 40 | + ... "conda": "my_conda", |
| 41 | + ... }, |
| 42 | + ... } |
| 43 | +
|
| 44 | + Attributes |
| 45 | + ---------- |
| 46 | + updates: Dict[str, Any] |
| 47 | + A dictionary containing the keys with their new values for the update. |
| 48 | + """ |
| 49 | + |
| 50 | + _SYSTEM_KEYS = set(("kind", "type", "spec", "infrastructure", "runtime")) |
| 51 | + |
| 52 | + def __init__(self, updates: Dict[str, Any], system_keys: List[str] = None): |
| 53 | + """ |
| 54 | + Initializes the DictionaryMerger with a dictionary of updates. |
| 55 | +
|
| 56 | + Parameters |
| 57 | + ---------- |
| 58 | + updates Dict[str, Any] |
| 59 | + A dictionary with keys that need to be updated and their new values. |
| 60 | + system_keys: List[str] |
| 61 | + The list of keys that cannot be replaced in the source dictionary. |
| 62 | + """ |
| 63 | + self.updates = updates |
| 64 | + self.system_keys = set(system_keys or []).union(self._SYSTEM_KEYS) |
| 65 | + |
| 66 | + def _update_keys( |
| 67 | + self, dict_to_update: Dict[str, Any], parent_key: str = "" |
| 68 | + ) -> None: |
| 69 | + """ |
| 70 | + Recursively updates the values of given keys in a dictionary. |
| 71 | +
|
| 72 | + Parameters |
| 73 | + ---------- |
| 74 | + dict_to_update: Dict[str, Any] |
| 75 | + The dictionary whose values are to be updated. |
| 76 | + parent_key: (str, optional) |
| 77 | + The current path in the dictionary being processed, used for nested dictionaries. |
| 78 | +
|
| 79 | + Returns |
| 80 | + ------- |
| 81 | + None |
| 82 | + The method updates the dict_to_update in place. |
| 83 | + """ |
| 84 | + for key, value in dict_to_update.items(): |
| 85 | + new_key = f"{parent_key}.{key}" if parent_key else key |
| 86 | + if isinstance(value, dict): |
| 87 | + self._update_keys(value, new_key) |
| 88 | + elif new_key in self.updates and key not in self.system_keys: |
| 89 | + dict_to_update[key] = self.updates[new_key] |
| 90 | + |
| 91 | + def _merge_updates( |
| 92 | + self, |
| 93 | + original_dict: Dict[str, Any], |
| 94 | + updated_dict: Dict[str, Any], |
| 95 | + parent_key: str = "", |
| 96 | + ) -> None: |
| 97 | + """ |
| 98 | + Merges updated values from the updated_dict into the original_dict based on the provided keys. |
| 99 | +
|
| 100 | + Parameters |
| 101 | + ---------- |
| 102 | + original_dict: Dict[str, Any] |
| 103 | + The original dictionary to merge updates into. |
| 104 | + updated_dict: Dict[str, Any] |
| 105 | + The updated dictionary with new values. |
| 106 | + parent_key: str |
| 107 | + The base key path for recursive merging. |
| 108 | +
|
| 109 | + Returns |
| 110 | + ------- |
| 111 | + None |
| 112 | + The method updates the original_dict in place. |
| 113 | + """ |
| 114 | + for key, value in updated_dict.items(): |
| 115 | + new_key = f"{parent_key}.{key}" if parent_key else key |
| 116 | + if isinstance(value, dict) and key in original_dict: |
| 117 | + self._merge_updates(original_dict[key], value, new_key) |
| 118 | + elif new_key in self.updates: |
| 119 | + original_dict[key] = value |
| 120 | + |
| 121 | + def merge(self, src_dict: Dict[str, Any]) -> Dict[str, Any]: |
| 122 | + """ |
| 123 | + Updates the dictionary with new values for specified keys and merges |
| 124 | + these changes back into the original dictionary. |
| 125 | +
|
| 126 | + Parameters |
| 127 | + ---------- |
| 128 | + src_dict: Dict[str, Any] |
| 129 | + The dictionary to be updated and merged. |
| 130 | +
|
| 131 | + Returns |
| 132 | + ------- |
| 133 | + Dict[str, Any] |
| 134 | + The updated and merged dictionary. |
| 135 | + """ |
| 136 | + if not self.updates: |
| 137 | + return src_dict |
| 138 | + |
| 139 | + original_dict = copy.deepcopy(src_dict) |
| 140 | + updated_dict = copy.deepcopy(src_dict) |
| 141 | + |
| 142 | + # Update the dictionary with the new values |
| 143 | + self._update_keys(updated_dict) |
| 144 | + |
| 145 | + # Merge the updates back into the original dictionary |
| 146 | + self._merge_updates(original_dict, updated_dict) |
| 147 | + |
| 148 | + return original_dict |
0 commit comments