Coverage for src / qdrant_loader / cli / commands / config.py: 40%

1from __future__ import annotations 

2 

3import json 

4from pathlib import Path 

5 

6from click.exceptions import ClickException 

7 

8from qdrant_loader.utils.sensitive import sanitize_exception_message 

9 

10 

11def run_show_config( 

12 workspace: Path | None, 

13 config: Path | None, 

14 env: Path | None, 

15 log_level: str, 

16) -> str: 

17 """Load configuration and return a JSON string representation. 

18 

19 This function is a command helper used by the CLI wrapper to display configuration. 

20 It performs validation, optional workspace setup, logging setup, and settings loading. 

21 """ 

22 try: 

23 from qdrant_loader.cli.config_loader import ( 

24 load_config_with_workspace as _load_config_with_workspace, 

25 ) 

26 from qdrant_loader.cli.config_loader import ( 

27 setup_workspace as _setup_workspace_impl, 

28 ) 

29 from qdrant_loader.config import get_settings 

30 from qdrant_loader.config.workspace import validate_workspace_flags 

31 from qdrant_loader.utils.logging import LoggingConfig 

32 

33 # Validate flag combinations 

34 validate_workspace_flags(workspace, config, env) 

35 

36 # Setup workspace if provided 

37 workspace_config = None 

38 if workspace: 

39 workspace_config = _setup_workspace_impl(workspace) 

40 

41 # Setup/reconfigure logging once with workspace support 

42 log_file = ( 

43 str(workspace_config.logs_path / "config.log") 

44 if workspace_config 

45 else "qdrant-loader.log" 

46 ) 

47 if getattr(LoggingConfig, "reconfigure", None): # Core supports reconfigure 

48 if getattr(LoggingConfig, "_initialized", False): # type: ignore[attr-defined] 

49 LoggingConfig.reconfigure(file=log_file, level=log_level) # type: ignore[attr-defined] 

50 else: 

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

52 else: 

53 # Compatibility path when running with an older core: clear root handlers 

54 # to avoid duplicates, then perform a full setup once. 

55 import logging as _py_logging 

56 

57 _py_logging.getLogger().handlers = [] 

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

59 

60 # Emit a single set of workspace-related info logs (no duplicates) 

61 if workspace_config: 

62 logger = LoggingConfig.get_logger(__name__) 

63 logger.info( 

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

65 ) 

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

67 logger.info( 

68 "Environment file found", env_path=str(workspace_config.env_path) 

69 ) 

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

71 logger.info( 

72 "Config file found", config_path=str(workspace_config.config_path) 

73 ) 

74 

75 # Load configuration (skip validation to avoid directory prompts) 

76 _load_config_with_workspace(workspace_config, config, env, skip_validation=True) 

77 settings = get_settings() 

78 if settings is None: 

79 raise ClickException("Settings not available") 

80 

81 # Redact sensitive values before dumping to JSON 

82 def _redact_secrets(value): 

83 """Recursively redact sensitive keys within nested structures. 

84 

85 Keys are compared case-insensitively and redacted if their name contains 

86 any common secret-like token. 

87 """ 

88 sensitive_tokens = { 

89 "password", 

90 "secret", 

91 "api_key", 

92 "token", 

93 "key", 

94 "credentials", 

95 "access_token", 

96 "private_key", 

97 "client_secret", 

98 } 

99 

100 mask = "****" 

101 

102 if isinstance(value, dict): 

103 redacted: dict = {} 

104 for k, v in value.items(): 

105 key_lc = str(k).lower() 

106 if any(token in key_lc for token in sensitive_tokens): 

107 redacted[k] = mask 

108 else: 

109 redacted[k] = _redact_secrets(v) 

110 return redacted 

111 if isinstance(value, list): 

112 return [_redact_secrets(item) for item in value] 

113 # Primitive or unknown types are returned as-is 

114 return value 

115 

116 config_dict = settings.model_dump(mode="json") 

117 safe_config = _redact_secrets(config_dict) 

118 return json.dumps(safe_config, indent=2, ensure_ascii=False) 

119 except ClickException: 

120 raise 

121 except Exception as e: 

122 safe_error = sanitize_exception_message(e) 

123 raise ClickException(f"Failed to display configuration: {safe_error}") from e