Coverage for presidio_analyzer / llm_utils / config_loader.py: 94%

1"""Configuration loading utilities for LLM recognizers.""" 

2import logging 

3from pathlib import Path 

4from typing import Dict, List, Union 

5 

6import yaml 

7 

8logger = logging.getLogger("presidio-analyzer") 

9 

10__all__ = [ 

11 "get_conf_path", 

12 "load_yaml_file", 

13 "resolve_config_path", 

14 "get_model_config", 

15 "validate_config_fields", 

16] 

17 

18 

19def get_conf_path(filename: str, conf_subdir: str = "conf") -> Path: 

20 """Get absolute path to file in configuration directory. 

21 

22 :param filename: Name of the file to locate. 

23 :param conf_subdir: Subdirectory name within package (default: "conf"). 

24 :return: Absolute path to the configuration file. 

25 """ 

26 return Path(__file__).parent.parent / conf_subdir / filename 

27 

28 

29def resolve_config_path(config_path: Union[str, Path]) -> Path: 

30 """Resolve configuration file path to absolute path. 

31 

32 Handles paths in multiple formats (checked in order): 

33 1. Absolute paths: returned as-is 

34 2. Relative paths that exist from CWD: returned as-is 

35 3. Paths resolved from the package conf/ directory 

36 4. Relative paths resolved from repository root 

37 

38 :param config_path: Configuration file path (string or Path object). 

39 :return: Resolved absolute path. 

40 """ 

41 config_path_obj = Path(config_path) 

42 

43 if config_path_obj.is_absolute(): 

44 return config_path_obj 

45 

46 if config_path_obj.exists(): 

47 return config_path_obj 

48 

49 conf_resolved = get_conf_path(str(config_path)) 

50 if conf_resolved.exists(): 

51 return conf_resolved 

52 

53 presidio_analyzer_root = Path(__file__).parent.parent 

54 repo_root = presidio_analyzer_root.parent.parent 

55 repo_resolved = repo_root / config_path 

56 

57 return repo_resolved 

58 

59 

60def load_yaml_file(filepath: Union[str, Path]) -> Dict: 

61 """Load and parse YAML configuration file. 

62 

63 Automatically resolves relative paths from presidio_analyzer package root. 

64 

65 :param filepath: Path to YAML file (string or Path object). 

66 :return: Parsed YAML content as dictionary. 

67 :raises FileNotFoundError: If file doesn't exist. 

68 :raises ValueError: If YAML parsing fails. 

69 """ 

70 resolved_path = resolve_config_path(filepath) 

71 

72 if not resolved_path.exists(): 

73 raise FileNotFoundError(f"File not found: {resolved_path}") 

74 

75 try: 

76 with open(resolved_path) as f: 

77 return yaml.safe_load(f) 

78 except yaml.YAMLError as e: 

79 raise ValueError(f"Failed to parse YAML: {e}") 

80 

81 

82def get_model_config(config: Dict, provider_key: str) -> Dict: 

83 """Extract and validate model configuration from provider config. 

84 

85 :param config: Full configuration dictionary. 

86 :param provider_key: Provider key (e.g., "openai", "ollama"). 

87 :return: Model configuration dictionary. 

88 :raises ValueError: If required model fields are missing. 

89 """ 

90 validate_config_fields( 

91 config, 

92 [ 

93 (provider_key,), 

94 (provider_key, "model"), 

95 (provider_key, "model", "model_id"), 

96 ] 

97 ) 

98 

99 return config[provider_key]["model"] 

100 

101 

102def validate_config_fields( 

103 config: Dict, 

104 required_fields: List[Union[str, tuple]], 

105 config_name: str = "Configuration" 

106) -> None: 

107 """Validate that required fields exist in configuration. 

108 

109 :param config: Configuration dictionary to validate. 

110 :param required_fields: List of required field names (str) or nested paths (tuple). 

111 :param config_name: Name of config for error messages (default: "Configuration"). 

112 :raises ValueError: If any required field is missing or empty. 

113 """ 

114 for field in required_fields: 

115 if isinstance(field, str): 

116 if not config.get(field): 

117 raise ValueError(f"{config_name} must contain '{field}'") 

118 elif isinstance(field, tuple): 

119 current = config 

120 for i, key in enumerate(field): 

121 if not isinstance(current, dict) or key not in current: 

122 path = ".".join(field[:i+1]) 

123 raise ValueError(f"{config_name} must contain '{path}'") 

124 current = current[key] 

125 if i == len(field) - 1 and not current: 

126 path = ".".join(field) 

127 raise ValueError(f"{config_name} must contain '{path}'")