Coverage for src/dataknobs_llm/prompts/adapters/dict_adapter.py: 12%

1"""Dictionary-based resource adapters. 

2 

3This module provides adapters that wrap Python dictionaries, enabling them to be 

4used as resource providers in the prompt library system. Supports both flat and 

5nested dictionaries with dot-notation key access. 

6""" 

7 

8from typing import Any, Dict, List, Optional 

9from .resource_adapter import ResourceAdapter, AsyncResourceAdapter, BaseSearchLogic 

10 

11 

12class DictResourceAdapter(ResourceAdapter): 

13 """Synchronous adapter for Python dictionary resources. 

14 

15 Features: 

16 - Nested key access using dot notation (e.g., "user.name") 

17 - Simple text-based search across values 

18 - Optional case-insensitive search 

19 - Filtering and deduplication via BaseSearchLogic 

20 

21 Example: 

22 >>> data = { 

23 ... "user": {"name": "Alice", "age": 30}, 

24 ... "settings": {"theme": "dark"} 

25 ... } 

26 >>> adapter = DictResourceAdapter(data, name="config") 

27 >>> adapter.get_value("user.name") 

28 "Alice" 

29 >>> adapter.search("Alice") 

30 [{"content": "Alice", "key": "user.name", "score": 1.0}] 

31 """ 

32 

33 def __init__( 

34 self, 

35 data: Dict[str, Any], 

36 name: str = "dict_adapter", 

37 case_sensitive: bool = False 

38 ): 

39 """Initialize dictionary adapter. 

40 

41 Args: 

42 data: Dictionary to wrap as a resource 

43 name: Name identifier for this adapter 

44 case_sensitive: Whether search should be case-sensitive (default: False) 

45 """ 

46 super().__init__(name=name) 

47 self._data = data 

48 self._case_sensitive = case_sensitive 

49 

50 def get_value( 

51 self, 

52 key: str, 

53 default: Any = None, 

54 context: Optional[Dict[str, Any]] = None 

55 ) -> Any: 

56 """Retrieve a value by key from the dictionary. 

57 

58 Supports nested key access using dot notation: 

59 - "simple_key" -> data["simple_key"] 

60 - "nested.key" -> data["nested"]["key"] 

61 - "deep.nested.key" -> data["deep"]["nested"]["key"] 

62 

63 Args: 

64 key: Key to look up (supports dot notation for nested access) 

65 default: Value to return if key is not found 

66 context: Optional context (unused in dict adapter) 

67 

68 Returns: 

69 Value at the key, or default if not found 

70 """ 

71 try: 

72 # Handle dot notation for nested keys 

73 if '.' in key: 

74 parts = key.split('.') 

75 value = self._data 

76 for part in parts: 

77 if isinstance(value, dict): 

78 value = value[part] 

79 else: 

80 return default 

81 return value 

82 else: 

83 return self._data.get(key, default) 

84 except (KeyError, TypeError): 

85 return default 

86 

87 def search( 

88 self, 

89 query: str, 

90 k: int = 5, 

91 filters: Optional[Dict[str, Any]] = None, 

92 **kwargs 

93 ) -> List[Dict[str, Any]]: 

94 """Perform text-based search across dictionary values. 

95 

96 Searches through all values in the dictionary (including nested values) 

97 and returns items where the query string appears in the value. 

98 

99 Args: 

100 query: Search query string 

101 k: Maximum number of results to return 

102 filters: Optional filters to apply (passed to BaseSearchLogic) 

103 **kwargs: Additional search options: 

104 - min_score: Minimum score threshold (default: 0.0) 

105 - deduplicate: Whether to deduplicate results (default: False) 

106 

107 Returns: 

108 List of search results with structure: 

109 { 

110 "content": <value>, 

111 "key": <key path>, 

112 "score": <relevance score>, 

113 "metadata": {<additional metadata>} 

114 } 

115 """ 

116 results = [] 

117 

118 # Normalize query for case-insensitive search 

119 search_query = query if self._case_sensitive else query.lower() 

120 

121 # Flatten dictionary and search 

122 for key, value in self._flatten_dict(self._data).items(): 

123 value_str = str(value) 

124 search_value = value_str if self._case_sensitive else value_str.lower() 

125 

126 if search_query in search_value: 

127 # Simple scoring: exact match = 1.0, contains = 0.8 

128 score = 1.0 if search_query == search_value else 0.8 

129 

130 result = BaseSearchLogic.format_search_result( 

131 value_str, 

132 score=score, 

133 metadata={"key": key} 

134 ) 

135 result["key"] = key # Add key to top level for easier access 

136 results.append(result) 

137 

138 if len(results) >= k: 

139 break 

140 

141 # Apply filters if provided 

142 if filters: 

143 results = BaseSearchLogic.filter_results(results, filters=filters) 

144 

145 # Apply min_score filter if provided 

146 min_score = kwargs.get('min_score', 0.0) 

147 if min_score > 0: 

148 results = BaseSearchLogic.filter_results(results, min_score=min_score) 

149 

150 # Deduplicate if requested 

151 if kwargs.get('deduplicate', False): 

152 results = BaseSearchLogic.deduplicate_results(results, key='content') 

153 

154 return results[:k] 

155 

156 def _flatten_dict( 

157 self, 

158 data: Dict[str, Any], 

159 parent_key: str = '', 

160 separator: str = '.' 

161 ) -> Dict[str, Any]: 

162 """Flatten nested dictionary with dot notation keys. 

163 

164 Args: 

165 data: Dictionary to flatten 

166 parent_key: Parent key prefix 

167 separator: Separator for nested keys 

168 

169 Returns: 

170 Flattened dictionary with dot-notation keys 

171 """ 

172 items = [] 

173 for key, value in data.items(): 

174 new_key = f"{parent_key}{separator}{key}" if parent_key else key 

175 

176 if isinstance(value, dict): 

177 items.extend(self._flatten_dict(value, new_key, separator).items()) 

178 else: 

179 items.append((new_key, value)) 

180 

181 return dict(items) 

182 

183 

184class AsyncDictResourceAdapter(AsyncResourceAdapter): 

185 """Asynchronous adapter for Python dictionary resources. 

186 

187 Provides the same functionality as DictResourceAdapter but with async methods. 

188 Useful for consistency in async codebases or when mixing with other async adapters. 

189 

190 Example: 

191 >>> data = {"user": {"name": "Alice", "age": 30}} 

192 >>> adapter = AsyncDictResourceAdapter(data) 

193 >>> await adapter.get_value("user.name") 

194 "Alice" 

195 """ 

196 

197 def __init__( 

198 self, 

199 data: Dict[str, Any], 

200 name: str = "async_dict_adapter", 

201 case_sensitive: bool = False 

202 ): 

203 """Initialize async dictionary adapter. 

204 

205 Args: 

206 data: Dictionary to wrap as a resource 

207 name: Name identifier for this adapter 

208 case_sensitive: Whether search should be case-sensitive (default: False) 

209 """ 

210 super().__init__(name=name) 

211 self._data = data 

212 self._case_sensitive = case_sensitive 

213 

214 async def get_value( 

215 self, 

216 key: str, 

217 default: Any = None, 

218 context: Optional[Dict[str, Any]] = None 

219 ) -> Any: 

220 """Retrieve a value by key from the dictionary (async). 

221 

222 See DictResourceAdapter.get_value for details. 

223 """ 

224 try: 

225 # Handle dot notation for nested keys 

226 if '.' in key: 

227 parts = key.split('.') 

228 value = self._data 

229 for part in parts: 

230 if isinstance(value, dict): 

231 value = value[part] 

232 else: 

233 return default 

234 return value 

235 else: 

236 return self._data.get(key, default) 

237 except (KeyError, TypeError): 

238 return default 

239 

240 async def search( 

241 self, 

242 query: str, 

243 k: int = 5, 

244 filters: Optional[Dict[str, Any]] = None, 

245 **kwargs 

246 ) -> List[Dict[str, Any]]: 

247 """Perform text-based search across dictionary values (async). 

248 

249 See DictResourceAdapter.search for details. 

250 """ 

251 results = [] 

252 

253 # Normalize query for case-insensitive search 

254 search_query = query if self._case_sensitive else query.lower() 

255 

256 # Flatten dictionary and search 

257 for key, value in self._flatten_dict(self._data).items(): 

258 value_str = str(value) 

259 search_value = value_str if self._case_sensitive else value_str.lower() 

260 

261 if search_query in search_value: 

262 # Simple scoring: exact match = 1.0, contains = 0.8 

263 score = 1.0 if search_query == search_value else 0.8 

264 

265 result = BaseSearchLogic.format_search_result( 

266 value_str, 

267 score=score, 

268 metadata={"key": key} 

269 ) 

270 result["key"] = key 

271 results.append(result) 

272 

273 if len(results) >= k: 

274 break 

275 

276 # Apply filters if provided 

277 if filters: 

278 results = BaseSearchLogic.filter_results(results, filters=filters) 

279 

280 # Apply min_score filter if provided 

281 min_score = kwargs.get('min_score', 0.0) 

282 if min_score > 0: 

283 results = BaseSearchLogic.filter_results(results, min_score=min_score) 

284 

285 # Deduplicate if requested 

286 if kwargs.get('deduplicate', False): 

287 results = BaseSearchLogic.deduplicate_results(results, key='content') 

288 

289 return results[:k] 

290 

291 def _flatten_dict( 

292 self, 

293 data: Dict[str, Any], 

294 parent_key: str = '', 

295 separator: str = '.' 

296 ) -> Dict[str, Any]: 

297 """Flatten nested dictionary with dot notation keys. 

298 

299 Args: 

300 data: Dictionary to flatten 

301 parent_key: Parent key prefix 

302 separator: Separator for nested keys 

303 

304 Returns: 

305 Flattened dictionary with dot-notation keys 

306 """ 

307 items = [] 

308 for key, value in data.items(): 

309 new_key = f"{parent_key}{separator}{key}" if parent_key else key 

310 

311 if isinstance(value, dict): 

312 items.extend(self._flatten_dict(value, new_key, separator).items()) 

313 else: 

314 items.append((new_key, value)) 

315 

316 return dict(items)