Coverage for src/qdrant_loader/cli/cli.py: 91%
163 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-08 06:05 +0000
1"""CLI module for QDrant Loader."""
3from pathlib import Path
5import click
6from click.decorators import group, option
7from click.exceptions import ClickException
8from click.types import Choice
9from click.types import Path as ClickPath
10from click.utils import echo
12from qdrant_loader.cli.async_utils import cancel_all_tasks as _cancel_all_tasks_helper
13from qdrant_loader.cli.asyncio import async_command
14from qdrant_loader.cli.commands import run_init as _commands_run_init
15from qdrant_loader.cli.config_loader import setup_workspace as _setup_workspace_impl
16from qdrant_loader.cli.logging_utils import get_logger as _get_logger_impl # noqa: F401
17from qdrant_loader.cli.logging_utils import ( # noqa: F401
18 setup_logging as _setup_logging_impl,
19)
20from qdrant_loader.cli.path_utils import (
21 create_database_directory as _create_db_dir_helper,
22)
23from qdrant_loader.cli.update_check import check_for_updates as _check_updates_helper
24from qdrant_loader.cli.version import get_version_str as _get_version_str
26# Use minimal imports at startup to improve CLI responsiveness.
27logger = None # Logger will be initialized when first accessed.
30def _get_version() -> str:
31 try:
32 return _get_version_str()
33 except Exception:
34 # Maintain CLI resilience: if version lookup fails for any reason,
35 # surface as 'unknown' rather than crashing the CLI.
36 return "unknown"
39# Back-compat helpers for tests: implement wrappers that operate on this module's global logger
42def _get_logger():
43 global logger
44 if logger is None:
45 from qdrant_loader.utils.logging import LoggingConfig
47 logger = LoggingConfig.get_logger(__name__)
48 return logger
51def _setup_logging(log_level: str, workspace_config=None) -> None:
52 try:
53 from qdrant_loader.utils.logging import LoggingConfig
55 log_format = "console"
56 log_file = (
57 str(workspace_config.logs_path) if workspace_config else "qdrant-loader.log"
58 )
59 LoggingConfig.setup(level=log_level, format=log_format, file=log_file)
60 # update module-global logger
61 global logger
62 logger = LoggingConfig.get_logger(__name__)
63 except Exception as e: # pragma: no cover - exercised via tests with mock
64 from click.exceptions import ClickException
66 raise ClickException(f"Failed to setup logging: {str(e)!s}") from e
69def _check_for_updates() -> None:
70 _check_updates_helper(_get_version())
73def _setup_workspace(workspace_path: Path):
74 workspace_config = _setup_workspace_impl(workspace_path)
75 # Re-log via this module's logger to satisfy tests patching _get_logger
76 lg = _get_logger()
77 lg.info("Using workspace", workspace=str(workspace_config.workspace_path))
78 if getattr(workspace_config, "env_path", None):
79 lg.info("Environment file found", env_path=str(workspace_config.env_path))
80 if getattr(workspace_config, "config_path", None):
81 lg.info("Config file found", config_path=str(workspace_config.config_path))
82 return workspace_config
85@group(name="qdrant-loader")
86@option(
87 "--log-level",
88 type=Choice(
89 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
90 ),
91 default="INFO",
92 help="Set the logging level.",
93)
94@click.version_option(
95 version=_get_version(),
96 message="qDrant Loader v.%(version)s",
97)
98def cli(log_level: str = "INFO") -> None:
99 """QDrant Loader CLI."""
100 # Initialize basic logging configuration before other operations.
101 _setup_logging(log_level)
103 # Check for available updates in background without blocking CLI startup.
104 _check_for_updates()
107def _create_database_directory(path: Path) -> bool:
108 """Create database directory with user confirmation.
110 Args:
111 path: Path to the database directory
113 Returns:
114 bool: True if directory was created, False if user declined
115 """
116 try:
117 abs_path = path.resolve()
118 _get_logger().info("The database directory does not exist", path=str(abs_path))
119 created = _create_db_dir_helper(abs_path)
120 if created:
121 _get_logger().info(f"Created directory: {abs_path}")
122 return created
123 except ClickException:
124 # Propagate ClickException from helper directly
125 raise
126 except Exception as e:
127 # Wrap any other unexpected errors
128 raise ClickException(f"Failed to create directory: {str(e)!s}") from e
131def _load_config(
132 config_path: Path | None = None,
133 env_path: Path | None = None,
134 skip_validation: bool = False,
135) -> None:
136 """Load configuration from file.
138 Args:
139 config_path: Optional path to config file
140 env_path: Optional path to .env file
141 skip_validation: If True, skip directory validation and creation
142 """
143 try:
144 # Lazy import to avoid slow startup
145 from qdrant_loader.config import initialize_config
147 # Step 1: If config path is provided, use it
148 if config_path is not None:
149 if not config_path.exists():
150 _get_logger().error("config_not_found", path=str(config_path))
151 raise ClickException(f"Config file not found: {str(config_path)!s}")
152 initialize_config(config_path, env_path, skip_validation=skip_validation)
153 return
155 # Step 2: If no config path, look for config.yaml in current folder
156 default_config = Path("config.yaml")
157 if default_config.exists():
158 initialize_config(default_config, env_path, skip_validation=skip_validation)
159 return
161 # Step 4: If no file is found, raise an error
162 raise ClickException(
163 f"No config file found. Please specify a config file or create config.yaml in the current directory: {str(default_config)!s}"
164 )
166 except Exception as e:
167 # Handle DatabaseDirectoryError and other exceptions
168 from qdrant_loader.config.state import DatabaseDirectoryError
170 if isinstance(e, DatabaseDirectoryError):
171 if skip_validation:
172 # For config display, we don't need to create the directory
173 return
175 # Get the path from the error - it's already a Path object
176 error_path = e.path
177 # Resolve to absolute path for consistency
178 abs_path = error_path.resolve()
180 if not _create_database_directory(abs_path):
181 raise ClickException(
182 "Database directory creation declined. Exiting."
183 ) from e
185 # No need to retry _load_config since the directory is now created
186 # Just initialize the config with the expanded path
187 if config_path is not None:
188 initialize_config(
189 config_path, env_path, skip_validation=skip_validation
190 )
191 else:
192 initialize_config(
193 Path("config.yaml"), env_path, skip_validation=skip_validation
194 )
195 elif isinstance(e, ClickException):
196 raise e from None
197 else:
198 _get_logger().error("config_load_failed", error=str(e))
199 raise ClickException(f"Failed to load configuration: {str(e)!s}") from e
202def _check_settings():
203 """Check if settings are available."""
204 # Lazy import to avoid slow startup
205 from qdrant_loader.config import get_settings
207 settings = get_settings()
208 if settings is None:
209 _get_logger().error("settings_not_available")
210 raise ClickException("Settings not available")
211 return settings
214def _load_config_with_workspace(
215 workspace_config,
216 config_path: Path | None = None,
217 env_path: Path | None = None,
218 skip_validation: bool = False,
219):
220 """Compatibility wrapper used by tests and project commands.
222 Delegates to qdrant_loader.cli.config_loader.load_config_with_workspace.
223 """
224 from qdrant_loader.cli.config_loader import (
225 load_config_with_workspace as _load_with_ws,
226 )
228 _load_with_ws(
229 workspace_config,
230 config_path=config_path,
231 env_path=env_path,
232 skip_validation=skip_validation,
233 )
236async def _run_init(settings, force: bool) -> None:
237 """Run initialization process via command helper, keeping existing logging."""
238 try:
239 await _commands_run_init(settings, force)
240 if force:
241 _get_logger().info(
242 "Collection recreated successfully",
243 collection=settings.qdrant_collection_name,
244 )
245 else:
246 _get_logger().info(
247 "Collection initialized successfully",
248 collection=settings.qdrant_collection_name,
249 )
250 except Exception as e:
251 _get_logger().error("init_failed", error=str(e))
252 raise ClickException(f"Failed to initialize collection: {str(e)!s}") from e
255@cli.command()
256@option(
257 "--workspace",
258 type=ClickPath(path_type=Path),
259 help="Workspace directory containing config.yaml and .env files. All output will be stored here.",
260)
261@option(
262 "--config", type=ClickPath(exists=True, path_type=Path), help="Path to config file."
263)
264@option("--env", type=ClickPath(exists=True, path_type=Path), help="Path to .env file.")
265@option("--force", is_flag=True, help="Force reinitialization of collection.")
266@option(
267 "--log-level",
268 type=Choice(
269 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
270 ),
271 default="INFO",
272 help="Set the logging level.",
273)
274@async_command
275async def init(
276 workspace: Path | None,
277 config: Path | None,
278 env: Path | None,
279 force: bool,
280 log_level: str,
281):
282 """Initialize QDrant collection."""
283 from qdrant_loader.cli.commands.init_cmd import run_init_command
285 await run_init_command(workspace, config, env, force, log_level)
288async def _cancel_all_tasks():
289 await _cancel_all_tasks_helper()
292@cli.command()
293@option(
294 "--workspace",
295 type=ClickPath(path_type=Path),
296 help="Workspace directory containing config.yaml and .env files. All output will be stored here.",
297)
298@option(
299 "--config", type=ClickPath(exists=True, path_type=Path), help="Path to config file."
300)
301@option("--env", type=ClickPath(exists=True, path_type=Path), help="Path to .env file.")
302@option(
303 "--project",
304 type=str,
305 help="Project ID to process. If specified, --source-type and --source will filter within this project.",
306)
307@option(
308 "--source-type",
309 type=str,
310 help="Source type to process (e.g., confluence, jira, git). If --project is specified, filters within that project; otherwise applies to all projects.",
311)
312@option(
313 "--source",
314 type=str,
315 help="Source name to process. If --project is specified, filters within that project; otherwise applies to all projects.",
316)
317@option(
318 "--log-level",
319 type=Choice(
320 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
321 ),
322 default="INFO",
323 help="Set the logging level.",
324)
325@option(
326 "--profile/--no-profile",
327 default=False,
328 help="Run the ingestion under cProfile and save output to 'profile.out' (for performance analysis).",
329)
330@option(
331 "--force",
332 is_flag=True,
333 help="Force processing of all documents, bypassing change detection. Warning: May significantly increase processing time and costs.",
334)
335@async_command
336async def ingest(
337 workspace: Path | None,
338 config: Path | None,
339 env: Path | None,
340 project: str | None,
341 source_type: str | None,
342 source: str | None,
343 log_level: str,
344 profile: bool,
345 force: bool,
346):
347 """Ingest documents from configured sources.
349 Examples:
350 # Ingest all projects
351 qdrant-loader ingest
353 # Ingest specific project
354 qdrant-loader ingest --project my-project
356 # Ingest specific source type from all projects
357 qdrant-loader ingest --source-type git
359 # Ingest specific source type from specific project
360 qdrant-loader ingest --project my-project --source-type git
362 # Ingest specific source from specific project
363 qdrant-loader ingest --project my-project --source-type git --source my-repo
365 # Force processing of all documents (bypass change detection)
366 qdrant-loader ingest --force
367 """
368 from qdrant_loader.cli.commands.ingest_cmd import run_ingest_command
370 await run_ingest_command(
371 workspace,
372 config,
373 env,
374 project,
375 source_type,
376 source,
377 log_level,
378 profile,
379 force,
380 )
383@cli.command()
384@option(
385 "--workspace",
386 type=ClickPath(path_type=Path),
387 help="Workspace directory containing config.yaml and .env files. All output will be stored here.",
388)
389@option(
390 "--log-level",
391 type=Choice(
392 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
393 ),
394 default="INFO",
395 help="Set the logging level.",
396)
397@option(
398 "--config", type=ClickPath(exists=True, path_type=Path), help="Path to config file."
399)
400@option("--env", type=ClickPath(exists=True, path_type=Path), help="Path to .env file.")
401def config(
402 workspace: Path | None, log_level: str, config: Path | None, env: Path | None
403):
404 """Display current configuration."""
405 try:
406 # Maintain test expectation: call _setup_logging again for the command
407 workspace_config = _setup_workspace(workspace) if workspace else None
408 _setup_logging(log_level, workspace_config)
410 echo("Current Configuration:")
411 from qdrant_loader.cli.commands.config import (
412 run_show_config as _run_show_config,
413 )
415 output = _run_show_config(workspace, config, env, log_level)
416 echo(output)
417 except Exception as e:
418 from qdrant_loader.utils.logging import LoggingConfig
420 LoggingConfig.get_logger(__name__).error("config_failed", error=str(e))
421 raise ClickException(f"Failed to display configuration: {str(e)!s}") from e
424# Add project management commands with lazy import
425def _add_project_commands():
426 """Lazily add project commands to avoid slow startup."""
427 from qdrant_loader.cli.project_commands import project_cli
429 cli.add_command(project_cli)
432# Only add project commands when CLI is actually used
433if __name__ == "__main__":
434 _add_project_commands()
435 cli()
436else:
437 # For when imported as a module, add commands on first access
438 import atexit
440 atexit.register(_add_project_commands)