Coverage for src / qdrant_loader / cli / cli.py: 88%
176 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-10 09:40 +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
12# async_command is needed at module level for @async_command decorator on commands.
13from qdrant_loader.cli.asyncio import async_command # noqa: F401
14from qdrant_loader.utils.sensitive import sanitize_exception_message
16# Heavy modules are lazy-imported to keep CLI startup fast.
19def _get_version_str():
20 from qdrant_loader.cli.version import get_version_str
22 return get_version_str()
25def _check_updates_helper(version):
26 from qdrant_loader.cli.update_check import check_for_updates
28 check_for_updates(version)
31def _setup_workspace_impl(workspace_path):
32 from qdrant_loader.cli.config_loader import setup_workspace
34 return setup_workspace(workspace_path)
37def _create_db_dir_helper(abs_path):
38 from qdrant_loader.cli.path_utils import create_database_directory
40 return create_database_directory(abs_path)
43async def _commands_run_init(settings, force):
44 from qdrant_loader.cli.commands import run_init
46 return await run_init(settings, force)
49# Use minimal imports at startup to improve CLI responsiveness.
50logger = None # Logger will be initialized when first accessed.
53def _get_version() -> str:
54 try:
55 return _get_version_str()
56 except Exception:
57 # Maintain CLI resilience: if version lookup fails for any reason,
58 # surface as 'unknown' rather than crashing the CLI.
59 return "unknown"
62# Back-compat helpers for tests: implement wrappers that operate on this module's global logger
65def _get_logger():
66 global logger
67 if logger is None:
68 from qdrant_loader.utils.logging import LoggingConfig
70 logger = LoggingConfig.get_logger(__name__)
71 return logger
74def _setup_logging(log_level: str, workspace_config=None) -> None:
75 try:
76 from qdrant_loader.utils.logging import LoggingConfig
78 log_format = "console"
79 log_file = (
80 str(workspace_config.logs_path) if workspace_config else "qdrant-loader.log"
81 )
82 LoggingConfig.setup(level=log_level, format=log_format, file=log_file)
83 # update module-global logger
84 global logger
85 logger = LoggingConfig.get_logger(__name__)
86 except Exception as e: # pragma: no cover - exercised via tests with mock
87 from click.exceptions import ClickException
89 raise ClickException(f"Failed to setup logging: {str(e)!s}") from e
92def _check_for_updates() -> None:
93 try:
94 _check_updates_helper(_get_version())
95 except Exception:
96 pass
99def _setup_workspace(workspace_path: Path):
100 workspace_config = _setup_workspace_impl(workspace_path)
101 return workspace_config
104@group(name="qdrant-loader")
105@option(
106 "--log-level",
107 type=Choice(
108 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
109 ),
110 default="INFO",
111 help="Set the logging level.",
112)
113@click.version_option(
114 version=_get_version(),
115 message="qDrant Loader v.%(version)s",
116)
117def cli(log_level: str = "INFO") -> None:
118 """QDrant Loader CLI."""
119 # Check for available updates in background without blocking CLI startup.
120 _check_for_updates()
123def _create_database_directory(path: Path) -> bool:
124 """Create database directory with user confirmation.
126 Args:
127 path: Path to the database directory
129 Returns:
130 bool: True if directory was created, False if user declined
131 """
132 try:
133 abs_path = path.resolve()
134 _get_logger().info("The database directory does not exist", path=str(abs_path))
135 created = _create_db_dir_helper(abs_path)
136 if created:
137 _get_logger().info(f"Created directory: {abs_path}")
138 return created
139 except ClickException:
140 # Propagate ClickException from helper directly
141 raise
142 except Exception as e:
143 # Wrap any other unexpected errors
144 raise ClickException(f"Failed to create directory: {str(e)!s}") from e
147def _load_config(
148 config_path: Path | None = None,
149 env_path: Path | None = None,
150 skip_validation: bool = False,
151) -> None:
152 """Load configuration from file.
154 Args:
155 config_path: Optional path to config file
156 env_path: Optional path to .env file
157 skip_validation: If True, skip directory validation and creation
158 """
159 try:
160 # Lazy import to avoid slow startup
161 from qdrant_loader.config import initialize_config
163 # Step 1: If config path is provided, use it
164 if config_path is not None:
165 if not config_path.exists():
166 _get_logger().error("config_not_found", path=str(config_path))
167 raise ClickException(f"Config file not found: {str(config_path)!s}")
168 initialize_config(config_path, env_path, skip_validation=skip_validation)
169 return
171 # Step 2: If no config path, look for config.yaml in current folder
172 default_config = Path("config.yaml")
173 if default_config.exists():
174 initialize_config(default_config, env_path, skip_validation=skip_validation)
175 return
177 # Step 4: If no file is found, raise an error
178 raise ClickException(
179 f"No config file found. Please specify a config file or create config.yaml in the current directory: {str(default_config)!s}"
180 )
182 except Exception as e:
183 # Handle DatabaseDirectoryError and other exceptions
184 from qdrant_loader.config.state import DatabaseDirectoryError
186 if isinstance(e, DatabaseDirectoryError):
187 if skip_validation:
188 # For config display, we don't need to create the directory
189 return
191 # Get the path from the error - it's already a Path object
192 error_path = e.path
193 # Resolve to absolute path for consistency
194 abs_path = error_path.resolve()
196 if not _create_database_directory(abs_path):
197 raise ClickException(
198 "Database directory creation declined. Exiting."
199 ) from e
201 # No need to retry _load_config since the directory is now created
202 # Just initialize the config with the expanded path
203 if config_path is not None:
204 initialize_config(
205 config_path, env_path, skip_validation=skip_validation
206 )
207 else:
208 initialize_config(
209 Path("config.yaml"), env_path, skip_validation=skip_validation
210 )
211 elif isinstance(e, ClickException):
212 raise e from None
213 else:
214 safe_error = sanitize_exception_message(e) or type(e).__name__
215 _get_logger().error("config_load_failed", error=safe_error)
216 raise ClickException(f"Failed to load configuration: {safe_error}") from e
219def _check_settings():
220 """Check if settings are available."""
221 # Lazy import to avoid slow startup
222 from qdrant_loader.config import get_settings
224 settings = get_settings()
225 if settings is None:
226 _get_logger().error("settings_not_available")
227 raise ClickException("Settings not available")
228 return settings
231def _load_config_with_workspace(
232 workspace_config,
233 config_path: Path | None = None,
234 env_path: Path | None = None,
235 skip_validation: bool = False,
236):
237 """Compatibility wrapper used by tests and project commands.
239 Delegates to qdrant_loader.cli.config_loader.load_config_with_workspace.
240 """
241 from qdrant_loader.cli.config_loader import (
242 load_config_with_workspace as _load_with_ws,
243 )
245 _load_with_ws(
246 workspace_config,
247 config_path=config_path,
248 env_path=env_path,
249 skip_validation=skip_validation,
250 )
253async def _run_init(settings, force: bool) -> None:
254 """Run initialization process via command helper, keeping existing logging."""
255 try:
256 await _commands_run_init(settings, force)
257 if force:
258 _get_logger().info(
259 "Collection recreated successfully",
260 collection=settings.qdrant_collection_name,
261 )
262 else:
263 _get_logger().info(
264 "Collection initialized successfully",
265 collection=settings.qdrant_collection_name,
266 )
267 except Exception as e:
268 safe_error = sanitize_exception_message(e) or type(e).__name__
269 _get_logger().error("init_failed", error=safe_error)
270 raise ClickException(f"Failed to initialize collection: {safe_error}") from e
273@cli.command()
274@option(
275 "--workspace",
276 type=ClickPath(path_type=Path),
277 help="Workspace directory containing config.yaml and .env files. All output will be stored here.",
278)
279@option(
280 "--config", type=ClickPath(exists=True, path_type=Path), help="Path to config file."
281)
282@option("--env", type=ClickPath(exists=True, path_type=Path), help="Path to .env file.")
283@option("--force", is_flag=True, help="Force reinitialization of collection.")
284@option(
285 "--log-level",
286 type=Choice(
287 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
288 ),
289 default="INFO",
290 help="Set the logging level.",
291)
292@async_command
293async def init(
294 workspace: Path | None,
295 config: Path | None,
296 env: Path | None,
297 force: bool,
298 log_level: str,
299):
300 """Initialize QDrant collection."""
301 from qdrant_loader.cli.commands.init_cmd import run_init_command
303 await run_init_command(workspace, config, env, force, log_level)
306async def _cancel_all_tasks():
307 from qdrant_loader.cli.async_utils import cancel_all_tasks
309 await cancel_all_tasks()
312@cli.command()
313@option(
314 "--workspace",
315 type=ClickPath(path_type=Path),
316 help="Workspace directory containing config.yaml and .env files. All output will be stored here.",
317)
318@option(
319 "--config", type=ClickPath(exists=True, path_type=Path), help="Path to config file."
320)
321@option("--env", type=ClickPath(exists=True, path_type=Path), help="Path to .env file.")
322@option(
323 "--project",
324 type=str,
325 help="Project ID to process. If specified, --source-type and --source will filter within this project.",
326)
327@option(
328 "--source-type",
329 type=str,
330 help="Source type to process (e.g., confluence, jira, git). If --project is specified, filters within that project; otherwise applies to all projects.",
331)
332@option(
333 "--source",
334 type=str,
335 help="Source name to process. If --project is specified, filters within that project; otherwise applies to all projects.",
336)
337@option(
338 "--log-level",
339 type=Choice(
340 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
341 ),
342 default="INFO",
343 help="Set the logging level.",
344)
345@option(
346 "--profile/--no-profile",
347 default=False,
348 help="Run the ingestion under cProfile and save output to 'profile.out' (for performance analysis).",
349)
350@option(
351 "--force",
352 is_flag=True,
353 help="Force processing of all documents, bypassing change detection. Warning: May significantly increase processing time and costs.",
354)
355@async_command
356async def ingest(
357 workspace: Path | None,
358 config: Path | None,
359 env: Path | None,
360 project: str | None,
361 source_type: str | None,
362 source: str | None,
363 log_level: str,
364 profile: bool,
365 force: bool,
366):
367 """Ingest documents from configured sources.
369 Examples:
370 # Ingest all projects
371 qdrant-loader ingest
373 # Ingest specific project
374 qdrant-loader ingest --project my-project
376 # Ingest specific source type from all projects
377 qdrant-loader ingest --source-type git
379 # Ingest specific source type from specific project
380 qdrant-loader ingest --project my-project --source-type git
382 # Ingest specific source from specific project
383 qdrant-loader ingest --project my-project --source-type git --source my-repo
385 # Force processing of all documents (bypass change detection)
386 qdrant-loader ingest --force
387 """
388 from qdrant_loader.cli.commands.ingest_cmd import run_ingest_command
390 await run_ingest_command(
391 workspace,
392 config,
393 env,
394 project,
395 source_type,
396 source,
397 log_level,
398 profile,
399 force,
400 )
403@cli.command()
404@option(
405 "--output-dir",
406 type=ClickPath(path_type=Path),
407 default=None,
408 help="Workspace directory to write config.yaml and .env files to. "
409 "If omitted, you will be prompted to choose a workspace folder.",
410)
411@option(
412 "--mode",
413 type=Choice(["default", "normal", "advanced"], case_sensitive=False),
414 default=None,
415 help="Setup mode: default (quick start), normal (interactive wizard), advanced (full control). "
416 "If omitted, you will be prompted to choose.",
417)
418def setup(output_dir: Path | None, mode: str | None) -> None:
419 """Setup wizard to generate config.yaml and .env files.
421 When run without flags, presents a TUI to choose a workspace folder and
422 setup mode (Default / Normal / Advanced).
423 """
424 from qdrant_loader.cli.commands.setup_cmd import run_setup
426 run_setup(output_dir, mode=mode)
429@cli.command()
430@option(
431 "--workspace",
432 type=ClickPath(path_type=Path),
433 help="Workspace directory containing config.yaml and .env files. All output will be stored here.",
434)
435@option(
436 "--log-level",
437 type=Choice(
438 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
439 ),
440 default="INFO",
441 help="Set the logging level.",
442)
443@option(
444 "--config", type=ClickPath(exists=True, path_type=Path), help="Path to config file."
445)
446@option("--env", type=ClickPath(exists=True, path_type=Path), help="Path to .env file.")
447def config(
448 workspace: Path | None, log_level: str, config: Path | None, env: Path | None
449):
450 """Display current configuration."""
451 try:
452 echo("Current Configuration:")
453 from qdrant_loader.cli.commands.config import (
454 run_show_config as _run_show_config,
455 )
457 output = _run_show_config(workspace, config, env, log_level)
458 echo(output)
459 except ClickException:
460 raise
461 except Exception as e:
462 from qdrant_loader.utils.logging import LoggingConfig
464 safe_error = sanitize_exception_message(e) or type(e).__name__
465 LoggingConfig.get_logger(__name__).error("config_failed", error=safe_error)
466 raise ClickException(f"Failed to display configuration: {safe_error}") from e
469# Add project management commands with lazy import
470def _add_project_commands():
471 """Lazily add project commands to avoid slow startup."""
472 from qdrant_loader.cli.project_commands import project_cli
474 cli.add_command(project_cli)
477# Only add project commands when CLI is actually used
478if __name__ == "__main__":
479 _add_project_commands()
480 cli()
481else:
482 # Register project commands immediately so they are available when CLI parses args
483 _add_project_commands()