Coverage for src/qdrant_loader_mcp_server/search/components/combining/flatten.py: 60%

1from __future__ import annotations 

2 

3from collections.abc import Mapping 

4from dataclasses import asdict, is_dataclass 

5from typing import Any 

6 

7 

8def flatten_metadata_components(metadata_components: dict[str, Any]) -> dict[str, Any]: 

9 flattened: dict[str, Any] = {} 

10 # Denylist of sensitive key patterns to redact 

11 sensitive_key_patterns = {"password", "secret", "token", "api_key", "apikey"} 

12 

13 def _should_include(key: Any, value: Any) -> bool: 

14 if not isinstance(key, str): 

15 return False 

16 if key.startswith("_"): 

17 return False 

18 try: 

19 if callable(value): 

20 return False 

21 except Exception: 

22 # Some objects may raise when inspected for callability 

23 return False 

24 return True 

25 

26 def _maybe_redact(key: str, value: Any) -> Any: 

27 lower_key = key.lower() 

28 for pattern in sensitive_key_patterns: 

29 if pattern in lower_key: 

30 return "[REDACTED]" 

31 return value 

32 

33 for _component_name, component in metadata_components.items(): 

34 if component is None: 

35 continue 

36 # Dataclasses (supports slots via asdict) 

37 if is_dataclass(component): 

38 for key, value in asdict(component).items(): 

39 if _should_include(key, value): 

40 flattened[key] = _maybe_redact(key, value) 

41 continue 

42 

43 # Generic Mapping (handles dict and subclasses like MappingProxyType) 

44 try: 

45 is_mapping = isinstance(component, Mapping) 

46 except Exception: 

47 # Some mocked objects with custom __dict__ can raise during isinstance checks 

48 is_mapping = False 

49 if is_mapping: 

50 try: 

51 items_iter = component.items() # type: ignore[assignment] 

52 except Exception: 

53 items_iter = [] 

54 for key, value in items_iter: 

55 if _should_include(key, value): 

56 flattened[key] = _maybe_redact(key, value) 

57 continue 

58 

59 # Fallback: inspect __dict__ but filter out private keys and callables 

60 if hasattr(component, "__dict__") and isinstance(component.__dict__, dict): 

61 for key, value in component.__dict__.items(): 

62 if _should_include(key, value): 

63 flattened[key] = _maybe_redact(key, value) 

64 

65 # Support for objects using __slots__ without a traditional __dict__ 

66 try: 

67 has_slots = hasattr(component, "__slots__") 

68 except Exception: 

69 has_slots = False 

70 if has_slots: 

71 try: 

72 slots = component.__slots__ 

73 except Exception: 

74 slots = [] 

75 # __slots__ can be a string or an iterable of strings 

76 slot_names: list[str] = [] 

77 if isinstance(slots, str): 

78 slot_names = [slots] 

79 else: 

80 try: 

81 slot_names = [s for s in slots if isinstance(s, str)] # type: ignore[assignment] 

82 except Exception: 

83 slot_names = [] 

84 for key in slot_names: 

85 if not isinstance(key, str) or key.startswith("_"): 

86 continue 

87 try: 

88 value = getattr(component, key) 

89 except Exception: 

90 continue 

91 if _should_include(key, value): 

92 flattened[key] = _maybe_redact(key, value) 

93 return flattened