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) if workspace_config else "qdrant-loader.log" 

44 ) 

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

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

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

48 else: 

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

50 else: 

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

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

53 import logging as _py_logging 

54 

55 _py_logging.getLogger().handlers = [] 

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

57 

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

59 if workspace_config: 

60 logger = LoggingConfig.get_logger(__name__) 

61 logger.info( 

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

63 ) 

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

65 logger.info( 

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

67 ) 

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

69 logger.info( 

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

71 ) 

72 

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

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

75 settings = get_settings() 

76 if settings is None: 

77 raise ClickException("Settings not available") 

78 

79 # Redact sensitive values before dumping to JSON 

80 def _redact_secrets(value): 

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

82 

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

84 any common secret-like token. 

85 """ 

86 sensitive_tokens = { 

87 "password", 

88 "secret", 

89 "api_key", 

90 "token", 

91 "key", 

92 "credentials", 

93 "access_token", 

94 "private_key", 

95 "client_secret", 

96 } 

97 

98 mask = "****" 

99 

100 if isinstance(value, dict): 

101 redacted: dict = {} 

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

103 key_lc = str(k).lower() 

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

105 redacted[k] = mask 

106 else: 

107 redacted[k] = _redact_secrets(v) 

108 return redacted 

109 if isinstance(value, list): 

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

111 # Primitive or unknown types are returned as-is 

112 return value 

113 

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

115 safe_config = _redact_secrets(config_dict) 

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

117 except ClickException: 

118 raise 

119 except Exception as e: 

120 safe_error = sanitize_exception_message(e) 

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