Coverage for functions \ flipdare \ util \ yaml_loader.py: 100%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from collections.abc import Callable 

15from pathlib import Path 

16from typing import IO, Any, cast 

17import ruamel.yaml as ru_yaml 

18 

19from flipdare.app_log import LOG 

20from flipdare.error.app_error import ServerError 

21 

22type YamlSourceType = IO[str] | Path 

23 

24 

25class YamlLoader: 

26 def __init__(self, source: YamlSourceType) -> None: 

27 self._source = source 

28 

29 def load(self, validator: Callable[[dict[str, Any]], None] | None = None) -> dict[str, Any]: 

30 """ 

31 Load and parse YAML from the configured source. 

32 

33 Args: 

34 validator: Optional callable that receives the parsed dict and should 

35 raise on invalid data. Runs before the data is returned. 

36 

37 """ 

38 data = ( 

39 self._load_file(self._source) 

40 if isinstance(self._source, Path) 

41 else self._load_stream(self._source) 

42 ) 

43 if validator is not None: 

44 validator(data) 

45 return data 

46 

47 def _load_file(self, yaml_path: Path) -> dict[str, Any]: 

48 if not yaml_path.exists(): 

49 raise FileNotFoundError(f"Config file not found: {yaml_path}") 

50 

51 LOG().warning(f"Loading config from {yaml_path}...") 

52 with yaml_path.open(encoding="utf-8") as f: 

53 return self._load_stream(f) 

54 

55 def _load_stream(self, cfg: IO[str]) -> dict[str, Any]: 

56 

57 from flipdare.util.yaml_error_formatter import YamlErrorFormatter 

58 

59 yaml = ru_yaml.YAML(typ="safe") 

60 try: 

61 data = yaml.load(cfg) 

62 return cast("dict[str, Any]", data) if data is not None else {} 

63 except Exception as e: 

64 msg = YamlErrorFormatter(str(self._source), e).formatted 

65 LOG().error(msg) 

66 raise ServerError(msg) from e