Coverage for async_durable_execution/preview.py: 97%

1"""Build compact structured previews for externally stored payloads.""" 

2 

3from __future__ import annotations 

4 

5import json 

6from dataclasses import dataclass, field 

7from enum import Enum 

8from typing import Any 

9 

10 

11class PreviewMode(str, Enum): 

12 """Controls which fields are visible by default.""" 

13 

14 INCLUDE_ALL = "INCLUDE_ALL" 

15 EXCLUDE_ALL = "EXCLUDE_ALL" 

16 

17 

18class FieldMatchMode(str, Enum): 

19 """Controls how a preview selector matches a field.""" 

20 

21 ANYWHERE = "ANYWHERE" 

22 PATH = "PATH" 

23 

24 

25@dataclass(frozen=True) 

26class PreviewField: 

27 """A field name or exact dot-separated path used by preview rules.""" 

28 

29 name: str 

30 match: FieldMatchMode = FieldMatchMode.ANYWHERE 

31 

32 def __post_init__(self) -> None: 

33 if not self.name: 

34 msg = "PreviewField.name must not be empty." 

35 raise ValueError(msg) 

36 if not isinstance(self.match, FieldMatchMode): 

37 msg = "PreviewField.match must be a FieldMatchMode." 

38 raise TypeError(msg) 

39 

40 

41@dataclass(frozen=True) 

42class PreviewConfig: 

43 """Configuration for :func:`build_preview`.""" 

44 

45 mode: PreviewMode 

46 include: tuple[PreviewField, ...] = field(default_factory=tuple) 

47 exclude: tuple[PreviewField, ...] = field(default_factory=tuple) 

48 mask: tuple[PreviewField, ...] = field(default_factory=tuple) 

49 mask_string: str = "***" 

50 max_preview_bytes: int = 4096 

51 max_traversal_nodes: int = 10_000 

52 max_depth: int = 64 

53 

54 def __post_init__(self) -> None: 

55 if not isinstance(self.mode, PreviewMode): 

56 msg = "PreviewConfig.mode must be a PreviewMode." 

57 raise TypeError(msg) 

58 if self.max_preview_bytes <= 0: 

59 msg = "max_preview_bytes must be positive." 

60 raise ValueError(msg) 

61 if self.max_traversal_nodes <= 0: 

62 msg = "max_traversal_nodes must be positive." 

63 raise ValueError(msg) 

64 if self.max_depth <= 0: 

65 msg = "max_depth must be positive." 

66 raise ValueError(msg) 

67 

68 

69def _field_matches(path: str, preview_field: PreviewField) -> bool: 

70 if preview_field.match is FieldMatchMode.PATH: 

71 return path == preview_field.name 

72 return preview_field.name in path.split(".") 

73 

74 

75def _is_matched(path: str, fields: tuple[PreviewField, ...]) -> bool: 

76 return any(_field_matches(path, preview_field) for preview_field in fields) 

77 

78 

79def build_preview( 

80 value: Any, 

81 config: PreviewConfig, 

82) -> dict[str, Any] | None: 

83 """Build a bounded nested preview from a mapping-like JSON value. 

84 

85 Exclusion wins over every other rule. Masking implies visibility unless 

86 the field is excluded. Arrays are traversed and matching object fields are 

87 merged into their containing preview path. 

88 """ 

89 if not isinstance(value, dict): 

90 return None 

91 

92 accepted: dict[str, Any] = {} 

93 visited_nodes = 0 

94 stopped = False 

95 

96 def visit_node() -> bool: 

97 nonlocal stopped, visited_nodes 

98 visited_nodes += 1 

99 if visited_nodes > config.max_traversal_nodes: 

100 stopped = True 

101 return not stopped 

102 

103 def add_value(path: str, preview_value: Any) -> None: 

104 nonlocal stopped 

105 missing = object() 

106 previous = accepted.get(path, missing) 

107 accepted[path] = preview_value 

108 candidate_preview = _paths_to_nested_dict(accepted) 

109 encoded = json.dumps( 

110 candidate_preview, 

111 ensure_ascii=False, 

112 separators=(",", ":"), 

113 ).encode("utf-8") 

114 if len(encoded) <= config.max_preview_bytes: 

115 return 

116 

117 if previous is missing: 117 ↛ 120line 117 didn't jump to line 120 because the condition on line 117 was always true

118 del accepted[path] 

119 else: 

120 accepted[path] = previous 

121 stopped = True 

122 

123 def collect(current: Any, path_prefix: str, depth: int) -> None: 

124 nonlocal stopped 

125 if stopped or depth > config.max_depth or not visit_node(): 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true

126 return 

127 if isinstance(current, list): 

128 for item in current: 

129 collect(item, path_prefix, depth + 1) 

130 if stopped: 

131 break 

132 return 

133 if not isinstance(current, dict): 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 return 

135 

136 for raw_key, child in current.items(): 

137 if stopped or not visit_node(): 

138 break 

139 key = str(raw_key) 

140 if "." in key: 

141 continue 

142 

143 path = f"{path_prefix}.{key}" if path_prefix else key 

144 excluded = _is_matched(path, config.exclude) 

145 masked = _is_matched(path, config.mask) 

146 visible = not excluded and ( 

147 masked 

148 or config.mode is PreviewMode.INCLUDE_ALL 

149 or _is_matched(path, config.include) 

150 ) 

151 

152 if not visible: 

153 if not excluded and isinstance(child, dict | list): 

154 collect(child, path, depth + 1) 

155 continue 

156 if masked: 

157 add_value(path, config.mask_string) 

158 elif isinstance(child, dict | list): 

159 collect(child, path, depth + 1) 

160 else: 

161 add_value(path, child) 

162 

163 collect(value, "", 0) 

164 if not accepted: 

165 return None 

166 return _paths_to_nested_dict(accepted) 

167 

168 

169def _paths_to_nested_dict(paths: dict[str, Any]) -> dict[str, Any]: 

170 result: dict[str, Any] = {} 

171 for path, value in paths.items(): 

172 parts = path.split(".") 

173 node = result 

174 for part in parts[:-1]: 

175 existing = node.get(part) 

176 if not isinstance(existing, dict): 

177 existing = {} 

178 node[part] = existing 

179 node = existing 

180 node[parts[-1]] = value 

181 return result