Coverage for src / invariant / ops / stdlib.py: 100.00%

1"""Standard operations library for basic data manipulation.""" 

2 

3from typing import Any 

4 

5 

6def identity(value: Any) -> Any: 

7 """Identity operation: returns the input unchanged. 

8 

9 Args: 

10 value: Any cacheable value. 

11 

12 Returns: 

13 The input value unchanged. 

14 """ 

15 return value 

16 

17 

18def add(a: int, b: int) -> int: 

19 """Add two integers. 

20 

21 Args: 

22 a: First integer. 

23 b: Second integer. 

24 

25 Returns: 

26 Sum of a and b. 

27 """ 

28 return a + b 

29 

30 

31def multiply(a: int, b: int) -> int: 

32 """Multiply two integers. 

33 

34 Args: 

35 a: First integer. 

36 b: Second integer. 

37 

38 Returns: 

39 Product of a and b. 

40 """ 

41 return a * b 

42 

43 

44def dict_get(dict_obj: dict[str, Any], key: str) -> Any: 

45 """Extract a value from a dictionary. 

46 

47 Args: 

48 dict_obj: Dictionary object. 

49 key: String key to look up. 

50 

51 Returns: 

52 The value at the specified key in the dictionary. 

53 

54 Raises: 

55 KeyError: If key not in dictionary. 

56 TypeError: If dict_obj is not a dict. 

57 """ 

58 if not isinstance(dict_obj, dict): 

59 raise TypeError(f"dict_get op requires dict, got {type(dict_obj)}") 

60 

61 if key not in dict_obj: 

62 raise KeyError(f"Key '{key}' not found in dictionary") 

63 

64 return dict_obj[key] 

65 

66 

67def make_dict(**kwargs: Any) -> dict[str, Any]: 

68 """Construct a dictionary from resolved parameters. 

69 

70 This operation collects all resolved parameters (which may have been 

71 constructed using ref() and cel() markers) into a new dictionary artifact. 

72 

73 Args: 

74 **kwargs: Any number of key-value pairs. Keys must be strings. 

75 Values can be any cacheable type (resolved from ref/cel markers). 

76 

77 Returns: 

78 A dictionary containing all the key-value pairs from kwargs. 

79 

80 Example: 

81 Node( 

82 op_name="stdlib:make_dict", 

83 params={"width": cel("bg.width"), "color": ref("fg_color")}, 

84 deps=["bg", "fg_color"], 

85 ) 

86 # Returns: {"width": 144, "color": "#ff0000"} 

87 """ 

88 return dict(kwargs) 

89 

90 

91def make_list(items: list[Any]) -> list[Any]: 

92 """Construct a list from resolved items. 

93 

94 This operation takes a list parameter (which may contain ref() and cel() 

95 markers that get resolved during parameter resolution) and returns it as 

96 a new list artifact. 

97 

98 Args: 

99 items: A list of cacheable values (resolved from ref/cel markers). 

100 

101 Returns: 

102 A list containing all the resolved items. 

103 

104 Example: 

105 Node( 

106 op_name="stdlib:make_list", 

107 params={"items": [ref("a"), ref("b"), cel("c + 1")]}, 

108 deps=["a", "b", "c"], 

109 ) 

110 # Returns: [<resolved a>, <resolved b>, <resolved c+1>] 

111 """ 

112 return list(items) 

113 

114 

115def coalesce(values: list[Any]) -> Any: 

116 """Return the first non-None value from a list of candidates. 

117 

118 Args: 

119 values: Candidate values in priority order. 

120 

121 Returns: 

122 The first value that is not None, or None when all candidates are None. 

123 """ 

124 if not isinstance(values, list): 

125 raise TypeError(f"coalesce op requires list, got {type(values)}") 

126 

127 for value in values: 

128 if value is not None: 

129 return value 

130 return None 

131 

132 

133# Package of standard operations 

134OPS: dict[str, Any] = { 

135 "identity": identity, 

136 "add": add, 

137 "multiply": multiply, 

138 "dict_get": dict_get, 

139 "make_dict": make_dict, 

140 "make_list": make_list, 

141 "coalesce": coalesce, 

142}