Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/pass_through_endpoints/jsonpath_extractor.py: 10%

1""" 

2JSONPath Extractor Module 

3 

4Extracts field values from data using simple JSONPath-like expressions. 

5""" 

6 

7from typing import Any, Final 

8 

9from litellm._logging import verbose_proxy_logger 

10 

11 

12class JsonPathExtractor: 

13 """Extracts field values from data using JSONPath-like expressions.""" 

14 

15 @staticmethod 

16 def extract_fields( 

17 data: dict, 

18 jsonpath_expressions: list[str], 

19 ) -> str: 

20 """ 

21 Extract field values from data using JSONPath-like expressions. 

22 

23 Supports simple expressions like: 

24 - "query" -> data["query"] 

25 - "documents[*].text" -> all text fields from documents array 

26 - "messages[*].content" -> all content fields from messages array 

27 

28 Returns concatenated string of all extracted values. 

29 """ 

30 extracted_values: Final[list[str]] = [] 

31 

32 for expr in jsonpath_expressions: 

33 try: 

34 value = JsonPathExtractor.evaluate(data, expr) 

35 if value: 

36 if isinstance(value, list): 

37 extracted_values.extend([str(v) for v in value if v]) 

38 else: 

39 extracted_values.append(str(value)) 

40 except Exception as e: 

41 verbose_proxy_logger.debug("Failed to extract field %s: %s", expr, str(e)) 

42 

43 return "\n".join(extracted_values) 

44 

45 @staticmethod 

46 def evaluate(data: dict, expr: str) -> str | list[str] | None: 

47 """ 

48 Evaluate a simple JSONPath-like expression. 

49 

50 Supports: 

51 - Simple key: "query" -> data["query"] 

52 - Nested key: "foo.bar" -> data["foo"]["bar"] 

53 - Array wildcard: "items[*].text" -> [item["text"] for item in data["items"]] 

54 """ 

55 if not expr or not data: 

56 return None 

57 

58 parts: Final = expr.replace("[*]", ".[*]").split(".") 

59 current: Any = data 

60 

61 for i, part in enumerate(parts): 

62 if current is None: 

63 return None 

64 

65 if part == "[*]": 

66 # Wildcard - current should be a list 

67 if not isinstance(current, list): 

68 return None 

69 

70 # Get remaining path 

71 remaining_path = ".".join(parts[i + 1 :]) 

72 if not remaining_path: 

73 return current 

74 

75 # Recursively evaluate remaining path for each item 

76 results = [] 

77 for item in current: 

78 if isinstance(item, dict): 

79 result = JsonPathExtractor.evaluate(item, remaining_path) 

80 if result: 

81 if isinstance(result, list): 

82 results.extend(result) 

83 else: 

84 results.append(result) 

85 return results if results else None 

86 

87 elif isinstance(current, dict): 

88 current = current.get(part) 

89 else: 

90 return None 

91 

92 return current