Coverage for src/qdrant_loader/cli/config_loader.py: 52%

1from __future__ import annotations 

2 

3from pathlib import Path 

4from typing import Any 

5 

6from click.exceptions import ClickException 

7 

8from qdrant_loader.config.state import DatabaseDirectoryError 

9from qdrant_loader.utils.logging import LoggingConfig 

10 

11 

12# Back-compat shim for tests that import _get_logger from cli.cli 

13def _get_logger(): 

14 return LoggingConfig.get_logger(__name__) 

15 

16 

17def setup_workspace(workspace_path: Path): 

18 try: 

19 from qdrant_loader.config.workspace import ( 

20 create_workspace_structure, 

21 ) 

22 from qdrant_loader.config.workspace import setup_workspace as _setup 

23 

24 create_workspace_structure(workspace_path) 

25 workspace_config = _setup(workspace_path) 

26 return workspace_config 

27 except ValueError as e: 

28 raise ClickException(str(e)) from e 

29 except Exception as e: # pragma: no cover - handled by CLI tests 

30 raise ClickException(f"Failed to setup workspace: {str(e)!s}") from e 

31 

32 

33def load_config_with_workspace( 

34 workspace_config: Any | None = None, 

35 config_path: Path | None = None, 

36 env_path: Path | None = None, 

37 skip_validation: bool = False, 

38) -> None: 

39 try: 

40 from qdrant_loader.config import initialize_config_with_workspace 

41 

42 if workspace_config: 

43 LoggingConfig.get_logger(__name__).debug( 

44 "Loading configuration in workspace mode" 

45 ) 

46 initialize_config_with_workspace( 

47 workspace_config, skip_validation=skip_validation 

48 ) 

49 else: 

50 LoggingConfig.get_logger(__name__).debug( 

51 "Loading configuration in traditional mode" 

52 ) 

53 load_config(config_path, env_path, skip_validation) 

54 except Exception as e: 

55 LoggingConfig.get_logger(__name__).error("config_load_failed", error=str(e)) 

56 raise ClickException(f"Failed to load configuration: {str(e)!s}") from e 

57 

58 

59def create_database_directory(path: Path) -> bool: 

60 try: 

61 abs_path = path.resolve() 

62 LoggingConfig.get_logger(__name__).info( 

63 "The database directory does not exist", path=str(abs_path) 

64 ) 

65 import click 

66 

67 if click.confirm("Would you like to create this directory?", default=True): 

68 abs_path.mkdir(parents=True, mode=0o755, exist_ok=True) 

69 LoggingConfig.get_logger(__name__).info("Created directory: %s", abs_path) 

70 return True 

71 return False 

72 except Exception as e: # pragma: no cover - interactive path 

73 raise ClickException(f"Failed to create directory: {str(e)!s}") from e 

74 

75 

76def load_config( 

77 config_path: Path | None = None, 

78 env_path: Path | None = None, 

79 skip_validation: bool = False, 

80) -> None: 

81 try: 

82 from qdrant_loader.config import initialize_config 

83 

84 if config_path is not None: 

85 if not config_path.exists(): 

86 LoggingConfig.get_logger(__name__).error( 

87 "config_not_found", path=str(config_path) 

88 ) 

89 raise ClickException(f"Config file not found: {str(config_path)!s}") 

90 initialize_config(config_path, env_path, skip_validation=skip_validation) 

91 return 

92 

93 default_config = Path("config.yaml") 

94 if default_config.exists(): 

95 initialize_config(default_config, env_path, skip_validation=skip_validation) 

96 return 

97 

98 raise ClickException( 

99 f"No config file found. Please specify a config file or create config.yaml in the current directory: {str(default_config)!s}" 

100 ) 

101 

102 except DatabaseDirectoryError as e: 

103 if skip_validation: 

104 return 

105 error_path = e.path 

106 abs_path = error_path.resolve() 

107 if not create_database_directory(abs_path): 

108 raise ClickException( 

109 "Database directory creation declined. Exiting." 

110 ) from e 

111 # After successful creation, initialize once using the original config_path if provided, 

112 # otherwise fall back to default config.yaml 

113 from qdrant_loader.config import initialize_config 

114 

115 target_config = config_path if config_path is not None else Path("config.yaml") 

116 initialize_config(target_config, env_path, skip_validation=skip_validation) 

117 except ClickException: 

118 raise 

119 except Exception as e: 

120 LoggingConfig.get_logger(__name__).error("config_load_failed", error=str(e)) 

121 raise ClickException(f"Failed to load configuration: {str(e)!s}") from e