Coverage for src / tracekit / config / loader.py: 99%

1"""Configuration file loading utilities. 

2 

3This module provides unified configuration loading from YAML and JSON files 

4with support for schema validation, default injection, and path resolution. 

5 

6 

7Example: 

8 >>> from tracekit.config.loader import load_config_file 

9 >>> config = load_config_file("pipeline.yaml", schema="pipeline") 

10""" 

11 

12from __future__ import annotations 

13 

14import json 

15from pathlib import Path 

16from typing import Any 

17 

18from tracekit.core.exceptions import ConfigurationError 

19 

20# Try to import yaml 

21try: 

22 import yaml 

23 

24 YAML_AVAILABLE = True 

25except ImportError: 

26 YAML_AVAILABLE = False 

27 

28 

29def load_config_file( 

30 path: str | Path, 

31 *, 

32 schema: str | None = None, 

33 validate: bool = True, 

34 inject_defaults: bool = True, 

35) -> dict[str, Any]: 

36 """Load configuration from YAML or JSON file. 

37 

38 Automatically detects file format from extension. 

39 Optionally validates against schema and injects defaults. 

40 

41 Args: 

42 path: Path to configuration file. 

43 schema: Schema name to validate against (e.g., "protocol"). 

44 validate: If True and schema provided, validate configuration. 

45 inject_defaults: If True, inject default values from schema. 

46 

47 Returns: 

48 Configuration dictionary. 

49 

50 Raises: 

51 ConfigurationError: If file cannot be loaded or parsed. 

52 

53 Example: 

54 >>> config = load_config_file("uart.yaml", schema="protocol") 

55 >>> print(config["name"]) 

56 'uart' 

57 """ 

58 path = Path(path).expanduser().resolve() 

59 

60 if not path.exists(): 

61 raise ConfigurationError( 

62 "Configuration file not found", 

63 config_key=str(path), 

64 ) 

65 

66 # Determine format from extension 

67 ext = path.suffix.lower() 

68 

69 if ext in (".yaml", ".yml"): 

70 config = _load_yaml(path) 

71 elif ext == ".json": 

72 config = _load_json(path) 

73 else: 

74 # Try YAML first, then JSON 

75 try: 

76 config = _load_yaml(path) 

77 except ConfigurationError: 

78 try: 

79 config = _load_json(path) 

80 except ConfigurationError: 

81 raise ConfigurationError( # noqa: B904 

82 f"Unsupported configuration format: {ext}", 

83 config_key=str(path), 

84 fix_hint="Use .yaml, .yml, or .json extension", 

85 ) 

86 

87 # Validate against schema if requested 

88 if validate and schema is not None: 

89 from tracekit.config.schema import validate_against_schema 

90 

91 validate_against_schema(config, schema) 

92 

93 # Inject defaults if requested 

94 if inject_defaults and schema is not None: 

95 from tracekit.config.defaults import inject_defaults as do_inject 

96 

97 config = do_inject(config, schema) 

98 

99 return config 

100 

101 

102def _load_yaml(path: Path) -> dict[str, Any]: 

103 """Load YAML configuration file. 

104 

105 Args: 

106 path: Path to YAML file. 

107 

108 Returns: 

109 Parsed configuration dictionary. 

110 

111 Raises: 

112 ConfigurationError: If YAML parsing fails. 

113 """ 

114 if not YAML_AVAILABLE: 

115 raise ConfigurationError( 

116 "YAML support not available", 

117 fix_hint="Install PyYAML: pip install pyyaml", 

118 ) 

119 

120 try: 

121 with open(path, encoding="utf-8") as f: 

122 content = yaml.safe_load(f) 

123 

124 if content is None: 

125 return {} 

126 

127 if not isinstance(content, dict): 

128 raise ConfigurationError( 

129 "Configuration must be a dictionary", 

130 config_key=str(path), 

131 expected_type="object", 

132 actual_value=type(content).__name__, 

133 ) 

134 

135 return content 

136 

137 except yaml.YAMLError as e: 

138 # Extract line number from YAML error 

139 line = None 

140 if hasattr(e, "problem_mark") and e.problem_mark is not None: 140 ↛ 143line 140 didn't jump to line 143 because the condition on line 140 was always true

141 line = e.problem_mark.line + 1 

142 

143 raise ConfigurationError( 

144 "Failed to parse YAML configuration", 

145 config_key=str(path), 

146 details=f"Line {line}: {e}" if line else str(e), 

147 ) from e 

148 except OSError as e: 

149 raise ConfigurationError( 

150 "Failed to read configuration file", 

151 config_key=str(path), 

152 details=str(e), 

153 ) from e 

154 

155 

156def _load_json(path: Path) -> dict[str, Any]: 

157 """Load JSON configuration file. 

158 

159 Args: 

160 path: Path to JSON file. 

161 

162 Returns: 

163 Parsed configuration dictionary. 

164 

165 Raises: 

166 ConfigurationError: If JSON parsing fails. 

167 """ 

168 try: 

169 with open(path, encoding="utf-8") as f: 

170 content = json.load(f) 

171 

172 if content is None: 

173 return {} 

174 

175 if not isinstance(content, dict): 

176 raise ConfigurationError( 

177 "Configuration must be a dictionary", 

178 config_key=str(path), 

179 expected_type="object", 

180 actual_value=type(content).__name__, 

181 ) 

182 

183 return content 

184 

185 except json.JSONDecodeError as e: 

186 raise ConfigurationError( 

187 "Failed to parse JSON configuration", 

188 config_key=str(path), 

189 details=f"Line {e.lineno}, column {e.colno}: {e.msg}", 

190 ) from e 

191 except OSError as e: 

192 raise ConfigurationError( 

193 "Failed to read configuration file", 

194 config_key=str(path), 

195 details=str(e), 

196 ) from e 

197 

198 

199def load_config( 

200 config_path: str | Path | None = None, 

201 *, 

202 use_defaults: bool = True, 

203) -> dict[str, Any]: 

204 """Load TraceKit configuration. 

205 

206 Searches for configuration in standard locations if no path provided. 

207 

208 Args: 

209 config_path: Path to configuration file. If None, searches 

210 standard locations. 

211 use_defaults: If True, merge with default configuration. 

212 

213 Returns: 

214 Configuration dictionary. 

215 

216 Example: 

217 >>> config = load_config() # Auto-find config 

218 >>> config = load_config("~/.tracekit/config.yaml") 

219 """ 

220 from tracekit.config.defaults import DEFAULT_CONFIG, deep_merge 

221 

222 config: dict[str, Any] = {} 

223 

224 if use_defaults: 

225 import copy 

226 

227 config = copy.deepcopy(DEFAULT_CONFIG) 

228 

229 if config_path is None: 

230 # Search standard locations 

231 search_paths = [ 

232 Path.cwd() / "tracekit.yaml", 

233 Path.cwd() / ".tracekit.yaml", 

234 Path.cwd() / "tracekit.json", 

235 Path.home() / ".tracekit" / "config.yaml", 

236 Path.home() / ".config" / "tracekit" / "config.yaml", 

237 ] 

238 

239 for path in search_paths: 

240 if path.exists(): 

241 config_path = path 

242 break 

243 

244 if config_path is not None: 

245 user_config = load_config_file( 

246 config_path, 

247 validate=False, 

248 inject_defaults=False, 

249 ) 

250 config = deep_merge(config, user_config) if use_defaults else user_config 

251 

252 return config 

253 

254 

255def save_config( 

256 config: dict[str, Any], 

257 path: str | Path, 

258 *, 

259 format: str | None = None, 

260) -> None: 

261 """Save configuration to file. 

262 

263 Args: 

264 config: Configuration dictionary to save. 

265 path: Output file path. 

266 format: Output format ("yaml" or "json"). Auto-detected from 

267 extension if not specified. 

268 

269 Raises: 

270 ConfigurationError: If configuration cannot be saved. 

271 

272 Example: 

273 >>> save_config(config, "config.yaml") 

274 >>> save_config(config, "config.json") 

275 """ 

276 path = Path(path).expanduser().resolve() 

277 

278 # Determine format 

279 if format is None: 

280 ext = path.suffix.lower() 

281 if ext in (".yaml", ".yml"): 

282 format = "yaml" 

283 elif ext == ".json": 

284 format = "json" 

285 else: 

286 format = "yaml" # Default 

287 

288 # Create parent directory 

289 path.parent.mkdir(parents=True, exist_ok=True) 

290 

291 try: 

292 if format == "yaml": 

293 if not YAML_AVAILABLE: 

294 raise ConfigurationError( 

295 "YAML support not available", 

296 fix_hint="Install PyYAML: pip install pyyaml", 

297 ) 

298 with open(path, "w", encoding="utf-8") as f: 

299 yaml.dump(config, f, default_flow_style=False, sort_keys=False) 

300 else: 

301 with open(path, "w", encoding="utf-8") as f: 

302 json.dump(config, f, indent=2) 

303 

304 except OSError as e: 

305 raise ConfigurationError( 

306 "Failed to save configuration", 

307 config_key=str(path), 

308 details=str(e), 

309 ) from e 

310 

311 

312def get_config_value( 

313 config: dict[str, Any], 

314 key_path: str, 

315 default: Any = None, 

316) -> Any: 

317 """Get configuration value by dot-separated path. 

318 

319 Args: 

320 config: Configuration dictionary. 

321 key_path: Dot-separated path (e.g., "defaults.sample_rate"). 

322 default: Default value if key not found. 

323 

324 Returns: 

325 Configuration value or default. 

326 

327 Example: 

328 >>> get_config_value(config, "defaults.sample_rate", 1e6) 

329 1000000.0 

330 """ 

331 keys = key_path.split(".") 

332 value = config 

333 

334 for key in keys: 

335 if isinstance(value, dict) and key in value: 

336 value = value[key] 

337 else: 

338 return default 

339 

340 return value 

341 

342 

343__all__ = [ 

344 "get_config_value", 

345 "load_config", 

346 "load_config_file", 

347 "save_config", 

348]