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

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 

10from qdrant_loader.utils.sensitive import sanitize_exception_message 

11 

12 

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

14def _get_logger(): 

15 return LoggingConfig.get_logger(__name__) 

16 

17 

18def setup_workspace(workspace_path: Path): 

19 try: 

20 from qdrant_loader.config.workspace import ( 

21 create_workspace_structure, 

22 ) 

23 from qdrant_loader.config.workspace import setup_workspace as _setup 

24 

25 create_workspace_structure(workspace_path) 

26 workspace_config = _setup(workspace_path) 

27 return workspace_config 

28 except ValueError as e: 

29 raise ClickException(str(e)) from e 

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

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

32 

33 

34def load_config_with_workspace( 

35 workspace_config: Any | None = None, 

36 config_path: Path | None = None, 

37 env_path: Path | None = None, 

38 skip_validation: bool = False, 

39) -> None: 

40 try: 

41 from qdrant_loader.config import initialize_config_with_workspace 

42 

43 if workspace_config: 

44 LoggingConfig.get_logger(__name__).debug( 

45 "Loading configuration in workspace mode" 

46 ) 

47 initialize_config_with_workspace( 

48 workspace_config, skip_validation=skip_validation 

49 ) 

50 else: 

51 LoggingConfig.get_logger(__name__).debug( 

52 "Loading configuration in traditional mode" 

53 ) 

54 load_config(config_path, env_path, skip_validation) 

55 except ClickException: 

56 raise 

57 except Exception as e: 

58 LoggingConfig.get_logger(__name__).error( 

59 "config_load_failed", error=sanitize_exception_message(e) 

60 ) 

61 # Lazy import to avoid slowing CLI startup 

62 from qdrant_loader.config.error_formatter import print_config_error 

63 

64 print_config_error(e) 

65 raise SystemExit(1) from e 

66 

67 

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

69 try: 

70 abs_path = path.resolve() 

71 LoggingConfig.get_logger(__name__).info( 

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

73 ) 

74 import click 

75 

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

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

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

79 return True 

80 return False 

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

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

83 

84 

85def load_config( 

86 config_path: Path | None = None, 

87 env_path: Path | None = None, 

88 skip_validation: bool = False, 

89) -> None: 

90 try: 

91 from qdrant_loader.config import initialize_config 

92 

93 if config_path is not None: 

94 if not config_path.exists(): 

95 LoggingConfig.get_logger(__name__).error( 

96 "config_not_found", path=str(config_path) 

97 ) 

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

99 initialize_config(config_path, env_path, skip_validation=skip_validation) 

100 return 

101 

102 default_config = Path("config.yaml") 

103 if default_config.exists(): 

104 initialize_config(default_config, env_path, skip_validation=skip_validation) 

105 return 

106 

107 raise ClickException( 

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

109 ) 

110 

111 except DatabaseDirectoryError as e: 

112 if skip_validation: 

113 return 

114 error_path = e.path 

115 abs_path = error_path.resolve() 

116 if not create_database_directory(abs_path): 

117 raise ClickException( 

118 "Database directory creation declined. Exiting." 

119 ) from e 

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

121 # otherwise fall back to default config.yaml 

122 from qdrant_loader.config import initialize_config 

123 

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

125 initialize_config(target_config, env_path, skip_validation=skip_validation) 

126 except ClickException: 

127 raise 

128 except Exception as e: 

129 LoggingConfig.get_logger(__name__).error( 

130 "config_load_failed", error=sanitize_exception_message(e) 

131 ) 

132 # Lazy import to avoid slowing CLI startup 

133 from qdrant_loader.config.error_formatter import print_config_error 

134 

135 print_config_error(e) 

136 raise SystemExit(1) from e