diff --git a/keep/functions/__init__.py b/keep/functions/__init__.py index cb690ee1cb..8b3aa1e617 100644 --- a/keep/functions/__init__.py +++ b/keep/functions/__init__.py @@ -604,3 +604,56 @@ def dictget(data: str | dict, key: str, default: any = None) -> any: return default return data.get(key, default) + + +def dict_set(data: str | dict, key: str, value: any) -> dict: + """ + Sets a key-value pair in a dictionary, returning a new dictionary. + + Args: + data (str | dict): The dictionary to update. Can be a JSON string or dict. + key (str): The key to set. + value (any): The value to set. + + Returns: + dict: A new dictionary with the key set to the value. + """ + if isinstance(data, str): + try: + data = json.loads(data) + except Exception: + data = {} + + if not isinstance(data, dict): + data = {} + + import copy + dict_copy = copy.deepcopy(data) + dict_copy[key] = value + return dict_copy + + +def dict_merge(*args) -> dict: + """ + Merges multiple dictionaries into one, returning a new dictionary. + Dictionaries passed later override earlier ones. + + Args: + *args: Dictionaries or JSON strings to merge. + + Returns: + dict: A new dictionary containing all merged keys and values. + """ + result = {} + for data in args: + if isinstance(data, str): + try: + data = json.loads(data) + except Exception: + continue + + if isinstance(data, dict): + result.update(data) + + return result + diff --git a/tests/test_functions.py b/tests/test_functions.py index b68356db85..c5866fcfba 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -1217,3 +1217,48 @@ def test_from_timestamp_with_timezone(): timestamp = dt.timestamp() result = functions.from_timestamp(timestamp, "Europe/Berlin") assert result == dt + + +def test_dict_set(): + """ + Test dict_set function + """ + # dict input + d = {"a": 1, "b": 2} + result = functions.dict_set(d, "c", 3) + assert result == {"a": 1, "b": 2, "c": 3} + # original dict should not be mutated + assert d == {"a": 1, "b": 2} + + # string input + d_str = '{"a": 1, "b": 2}' + result = functions.dict_set(d_str, "c", 3) + assert result == {"a": 1, "b": 2, "c": 3} + + # override existing key + result = functions.dict_set(d, "b", 4) + assert result == {"a": 1, "b": 4} + + # invalid string input + result = functions.dict_set("invalid", "a", 1) + assert result == {"a": 1} + + +def test_dict_merge(): + """ + Test dict_merge function + """ + d1 = {"a": 1, "b": 2} + d2 = {"b": 3, "c": 4} + d3 = '{"c": 5, "d": 6}' + + result = functions.dict_merge(d1, d2, d3) + assert result == {"a": 1, "b": 3, "c": 5, "d": 6} + + # original dicts should not be mutated + assert d1 == {"a": 1, "b": 2} + + # invalid inputs + result = functions.dict_merge(d1, "invalid", {"e": 7}) + assert result == {"a": 1, "b": 2, "e": 7} +