Coverage for src/qdrant_loader/cli/commands/init_cmd.py: 0%
91 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-11 07:21 +0000
1from __future__ import annotations
3from pathlib import Path
5from click.exceptions import ClickException
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
21from . import run_init as _commands_run_init
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.
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)
42 # Setup logging first (workspace-aware later). Use core reconfigure if available.
43 if getattr(LoggingConfig, "reconfigure", None): # type: ignore[attr-defined]
44 if getattr(LoggingConfig, "_initialized", False): # type: ignore[attr-defined]
45 LoggingConfig.reconfigure(file="qdrant-loader.log") # type: ignore[attr-defined]
46 else:
47 LoggingConfig.setup(
48 level=log_level, format="console", file="qdrant-loader.log"
49 )
50 else:
51 import logging as _py_logging
53 _py_logging.getLogger().handlers = []
54 LoggingConfig.setup(
55 level=log_level, format="console", file="qdrant-loader.log"
56 )
57 logger = LoggingConfig.get_logger(__name__)
59 # Check for updates (non-blocking semantics preserved by immediate return)
60 try:
61 _check_updates_helper(_get_version_str())
62 except Exception:
63 # Update check failures should not break the command
64 pass
66 # Setup workspace if provided
67 workspace_config = None
68 if workspace:
69 workspace_config = _setup_workspace_impl(workspace)
70 logger.info(
71 "Using workspace", workspace=str(workspace_config.workspace_path)
72 )
73 if getattr(workspace_config, "env_path", None):
74 logger.info(
75 "Environment file found",
76 env_path=str(workspace_config.env_path),
77 )
78 if getattr(workspace_config, "config_path", None):
79 logger.info(
80 "Config file found",
81 config_path=str(workspace_config.config_path),
82 )
84 # Setup logging again with workspace-aware file path
85 log_file = (
86 str(workspace_config.logs_path)
87 if workspace_config
88 else "qdrant-loader.log"
89 )
90 if getattr(LoggingConfig, "reconfigure", None): # type: ignore[attr-defined]
91 LoggingConfig.reconfigure(file=log_file) # type: ignore[attr-defined]
92 else:
93 import logging as _py_logging
95 _py_logging.getLogger().handlers = []
96 LoggingConfig.setup(level=log_level, format="console", file=log_file)
98 # Load configuration
99 _load_config_with_workspace(workspace_config, config, env)
101 # Fetch settings
102 settings = get_settings()
103 if settings is None:
104 logger.error("settings_not_available")
105 raise ClickException("Settings not available")
107 # Delete and recreate the database file if it exists
108 db_path_str = settings.global_config.state_management.database_path
109 if db_path_str != ":memory:":
110 db_path = Path(db_path_str)
111 db_dir = db_path.parent
112 if not db_dir.exists():
113 if not _create_database_directory(db_dir):
114 raise ClickException(
115 "Database directory creation declined. Exiting."
116 )
118 if db_path.exists() and force:
119 logger.info("Resetting state database", database_path=str(db_path))
120 db_path.unlink()
121 logger.info(
122 "State database reset completed", database_path=str(db_path)
123 )
124 elif force:
125 logger.info(
126 "State database reset skipped (no existing database)",
127 database_path=str(db_path),
128 )
130 # Run initialization via command helper
131 await _commands_run_init(settings, force)
132 if force:
133 logger.info(
134 "Collection recreated successfully",
135 collection=settings.qdrant_collection_name,
136 )
137 else:
138 logger.info(
139 "Collection initialized successfully",
140 collection=settings.qdrant_collection_name,
141 )
143 # Completed successfully
144 return
146 except DatabaseDirectoryError as e:
147 # Mirror original behavior for directory creation prompts, then retry if allowed
148 if workspace is None:
149 # derive from current directory
150 target = Path(e.path).resolve()
151 else:
152 target = e.path.resolve()
153 if not _create_database_directory(target):
154 raise ClickException(
155 "Database directory creation declined. Exiting."
156 ) from e
158 if attempts >= max_retries:
159 raise ClickException(
160 "Initialization aborted after exhausting retries for database directory creation."
161 ) from e
162 attempts += 1
163 continue
165 except ClickException:
166 raise
167 except Exception as e:
168 logger = LoggingConfig.get_logger(__name__)
169 logger.error("init_failed", error=str(e))
170 raise ClickException(f"Failed to initialize collection: {str(e)!s}") from e
173def _create_database_directory(path: Path) -> bool:
174 """Create database directory with logging (non-interactive)."""
175 try:
176 abs_path = path.resolve()
177 LoggingConfig.get_logger(__name__).info(
178 "The database directory does not exist", path=str(abs_path)
179 )
180 created = _create_db_dir_helper(abs_path)
181 if created:
182 LoggingConfig.get_logger(__name__).info(f"Created directory: {abs_path}")
183 return created
184 except Exception as e: # pragma: no cover - error path
185 raise ClickException(f"Failed to create directory: {str(e)!s}") from e