Coverage for haystack/components/agents/state/state_utils.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import inspect 

6from typing import Any, TypeVar, Union, get_origin 

7 

8from haystack.utils.type_serialization import _is_union_type 

9 

10T = TypeVar("T") 

11 

12 

13def _is_valid_type(obj: Any) -> bool: 

14 """ 

15 Check if an object is a valid type annotation. 

16 

17 Valid types include: 

18 - Normal classes (str, dict, CustomClass) 

19 - Generic types (list[str], dict[str, int]) 

20 - Union types (Union[str, int], Optional[str], str | int, str | None) 

21 

22 :param obj: The object to check 

23 :return: True if the object is a valid type annotation, False otherwise 

24 

25 Example usage: 

26 # >> _is_valid_type(str) 

27 # >> True 

28 # >> _is_valid_type(list[int]) 

29 # >> True 

30 # >> _is_valid_type(Union[str, int]) 

31 # >> True 

32 # >> _is_valid_type(str | int) 

33 # >> True 

34 # >> _is_valid_type(42) 

35 # >> False 

36 """ 

37 # Handle Union types (including Optional) 

38 if (origin := get_origin(obj)) and _is_union_type(origin): 

39 return True 

40 

41 # Bare Union type (without parameters) is not a valid type annotation 

42 # Previously handled by inspect.isclass(obj) but in python 3.14 this returns True for typing.Union 

43 if obj == Union: 

44 return False 

45 

46 # Handle normal classes and generic types 

47 return inspect.isclass(obj) or type(obj).__name__ in {"_GenericAlias", "GenericAlias"} 

48 

49 

50def _is_list_type(type_hint: Any) -> bool: 

51 """ 

52 Check if a type hint represents a list type. 

53 

54 :param type_hint: The type hint to check 

55 :return: True if the type hint represents a list, False otherwise 

56 """ 

57 return type_hint == list or (hasattr(type_hint, "__origin__") and get_origin(type_hint) == list) 

58 

59 

60def merge_lists(current: Union[list[T], T, None], new: Union[list[T], T]) -> list[T]: 

61 """ 

62 Merges two values into a single list. 

63 

64 If either `current` or `new` is not already a list, it is converted into one. 

65 The function ensures that both inputs are treated as lists and concatenates them. 

66 

67 If `current` is None, it is treated as an empty list. 

68 

69 :param current: The existing value(s), either a single item or a list. 

70 :param new: The new value(s) to merge, either a single item or a list. 

71 :return: A list containing elements from both `current` and `new`. 

72 """ 

73 current_list = [] if current is None else current if isinstance(current, list) else [current] 

74 new_list = new if isinstance(new, list) else [new] 

75 return current_list + new_list 

76 

77 

78def replace_values(current: Any, new: Any) -> Any: # noqa: ARG001 

79 """ 

80 Replace the `current` value with the `new` value. 

81 

82 :param current: The existing value 

83 :param new: The new value to replace 

84 :return: The new value 

85 """ 

86 return new