Coverage for src/qdrant_loader/cli/commands/init_cmd.py: 0%

1from __future__ import annotations 

2 

3from pathlib import Path 

4 

5from click.exceptions import ClickException 

6 

7from qdrant_loader.cli.config_loader import ( 

8 load_config_with_workspace as _load_config_with_workspace, 

9) 

10from qdrant_loader.cli.config_loader import setup_workspace as _setup_workspace_impl 

11from qdrant_loader.cli.path_utils import ( 

12 create_database_directory as _create_db_dir_helper, 

13) 

14from qdrant_loader.cli.update_check import check_for_updates as _check_updates_helper 

15from qdrant_loader.cli.version import get_version_str as _get_version_str 

16from qdrant_loader.config import get_settings 

17from qdrant_loader.config.state import DatabaseDirectoryError 

18from qdrant_loader.config.workspace import validate_workspace_flags 

19from qdrant_loader.utils.logging import LoggingConfig 

20 

21from . import run_init as _commands_run_init 

22 

23 

24async def run_init_command( 

25 workspace: Path | None, 

26 config: Path | None, 

27 env: Path | None, 

28 force: bool, 

29 log_level: str, 

30 max_retries: int = 1, 

31) -> None: 

32 """Implementation for the `init` CLI command. 

33 

34 Keeps side-effects and logging identical to the previous inline implementation. 

35 """ 

36 attempts = 0 

37 while True: 

38 try: 

39 # Validate flag combinations 

40 validate_workspace_flags(workspace, config, env) 

41 

42 # Setup logging first (workspace-aware later) 

43 LoggingConfig.setup( 

44 level=log_level, format="console", file="qdrant-loader.log" 

45 ) 

46 logger = LoggingConfig.get_logger(__name__) 

47 

48 # Check for updates (non-blocking semantics preserved by immediate return) 

49 try: 

50 _check_updates_helper(_get_version_str()) 

51 except Exception: 

52 # Update check failures should not break the command 

53 pass 

54 

55 # Setup workspace if provided 

56 workspace_config = None 

57 if workspace: 

58 workspace_config = _setup_workspace_impl(workspace) 

59 logger.info( 

60 "Using workspace", workspace=str(workspace_config.workspace_path) 

61 ) 

62 if getattr(workspace_config, "env_path", None): 

63 logger.info( 

64 "Environment file found", 

65 env_path=str(workspace_config.env_path), 

66 ) 

67 if getattr(workspace_config, "config_path", None): 

68 logger.info( 

69 "Config file found", 

70 config_path=str(workspace_config.config_path), 

71 ) 

72 

73 # Setup logging again with workspace-aware file path 

74 log_file = ( 

75 str(workspace_config.logs_path) 

76 if workspace_config 

77 else "qdrant-loader.log" 

78 ) 

79 LoggingConfig.setup(level=log_level, format="console", file=log_file) 

80 

81 # Load configuration 

82 _load_config_with_workspace(workspace_config, config, env) 

83 

84 # Fetch settings 

85 settings = get_settings() 

86 if settings is None: 

87 logger.error("settings_not_available") 

88 raise ClickException("Settings not available") 

89 

90 # Delete and recreate the database file if it exists 

91 db_path_str = settings.global_config.state_management.database_path 

92 if db_path_str != ":memory:": 

93 db_path = Path(db_path_str) 

94 db_dir = db_path.parent 

95 if not db_dir.exists(): 

96 if not _create_database_directory(db_dir): 

97 raise ClickException( 

98 "Database directory creation declined. Exiting." 

99 ) 

100 

101 if db_path.exists() and force: 

102 logger.info("Resetting state database", database_path=str(db_path)) 

103 db_path.unlink() 

104 logger.info( 

105 "State database reset completed", database_path=str(db_path) 

106 ) 

107 elif force: 

108 logger.info( 

109 "State database reset skipped (no existing database)", 

110 database_path=str(db_path), 

111 ) 

112 

113 # Run initialization via command helper 

114 await _commands_run_init(settings, force) 

115 if force: 

116 logger.info( 

117 "Collection recreated successfully", 

118 collection=settings.qdrant_collection_name, 

119 ) 

120 else: 

121 logger.info( 

122 "Collection initialized successfully", 

123 collection=settings.qdrant_collection_name, 

124 ) 

125 

126 # Completed successfully 

127 return 

128 

129 except DatabaseDirectoryError as e: 

130 # Mirror original behavior for directory creation prompts, then retry if allowed 

131 if workspace is None: 

132 # derive from current directory 

133 target = Path(e.path).resolve() 

134 else: 

135 target = e.path.resolve() 

136 if not _create_database_directory(target): 

137 raise ClickException( 

138 "Database directory creation declined. Exiting." 

139 ) from e 

140 

141 if attempts >= max_retries: 

142 raise ClickException( 

143 "Initialization aborted after exhausting retries for database directory creation." 

144 ) from e 

145 attempts += 1 

146 continue 

147 

148 except ClickException: 

149 raise 

150 except Exception as e: 

151 logger = LoggingConfig.get_logger(__name__) 

152 logger.error("init_failed", error=str(e)) 

153 raise ClickException(f"Failed to initialize collection: {str(e)!s}") from e 

154 

155 

156def _create_database_directory(path: Path) -> bool: 

157 """Create database directory with logging (non-interactive).""" 

158 try: 

159 abs_path = path.resolve() 

160 LoggingConfig.get_logger(__name__).info( 

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

162 ) 

163 created = _create_db_dir_helper(abs_path) 

164 if created: 

165 LoggingConfig.get_logger(__name__).info(f"Created directory: {abs_path}") 

166 return created 

167 except Exception as e: # pragma: no cover - error path 

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