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