Coverage for src/qdrant_loader/config/workspace.py: 69%

1"""Workspace configuration management for QDrant Loader CLI.""" 

2 

3import os 

4from dataclasses import dataclass 

5from pathlib import Path 

6 

7from qdrant_loader.utils.logging import LoggingConfig 

8 

9logger = LoggingConfig.get_logger(__name__) 

10 

11 

12@dataclass 

13class WorkspaceConfig: 

14 """Configuration for workspace mode.""" 

15 

16 workspace_path: Path 

17 config_path: Path 

18 env_path: Path | None 

19 logs_path: Path 

20 metrics_path: Path 

21 database_path: Path 

22 

23 def __post_init__(self): 

24 """Validate workspace configuration after initialization.""" 

25 # Ensure workspace path is absolute 

26 self.workspace_path = self.workspace_path.resolve() 

27 

28 # Validate workspace directory exists 

29 if not self.workspace_path.exists(): 

30 raise ValueError( 

31 f"Workspace directory does not exist: {self.workspace_path}" 

32 ) 

33 

34 if not self.workspace_path.is_dir(): 

35 raise ValueError( 

36 f"Workspace path is not a directory: {self.workspace_path}" 

37 ) 

38 

39 # Validate config.yaml exists 

40 if not self.config_path.exists(): 

41 raise ValueError(f"config.yaml not found in workspace: {self.config_path}") 

42 

43 # Validate workspace is writable 

44 if not os.access(self.workspace_path, os.W_OK): 

45 raise ValueError( 

46 f"Cannot write to workspace directory: {self.workspace_path}" 

47 ) 

48 

49 logger.debug( 

50 "Workspace configuration validated", workspace=str(self.workspace_path) 

51 ) 

52 

53 

54def setup_workspace(workspace_path: Path) -> WorkspaceConfig: 

55 """Setup and validate workspace configuration. 

56 

57 Args: 

58 workspace_path: Path to the workspace directory 

59 

60 Returns: 

61 WorkspaceConfig: Validated workspace configuration 

62 

63 Raises: 

64 ValueError: If workspace validation fails 

65 """ 

66 logger.debug("Setting up workspace", path=str(workspace_path)) 

67 

68 # Resolve to absolute path 

69 workspace_path = workspace_path.resolve() 

70 

71 # Define workspace file paths 

72 config_path = workspace_path / "config.yaml" 

73 env_path = workspace_path / ".env" 

74 logs_path = workspace_path / "logs" / "qdrant-loader.log" 

75 metrics_path = workspace_path / "metrics" 

76 data_path = workspace_path / "data" 

77 database_path = data_path / "qdrant-loader.db" 

78 

79 # Check if .env file exists (optional) 

80 env_path_final = env_path if env_path.exists() else None 

81 

82 # Create workspace config 

83 workspace_config = WorkspaceConfig( 

84 workspace_path=workspace_path, 

85 config_path=config_path, 

86 env_path=env_path_final, 

87 logs_path=logs_path, 

88 metrics_path=metrics_path, 

89 database_path=database_path, 

90 ) 

91 

92 logger.debug("Workspace setup completed", workspace=str(workspace_path)) 

93 return workspace_config 

94 

95 

96def validate_workspace(workspace_path: Path) -> bool: 

97 """Validate if a directory can be used as a workspace. 

98 

99 Args: 

100 workspace_path: Path to validate 

101 

102 Returns: 

103 bool: True if valid workspace, False otherwise 

104 """ 

105 try: 

106 setup_workspace(workspace_path) 

107 return True 

108 except ValueError as e: 

109 logger.debug( 

110 "Workspace validation failed", path=str(workspace_path), error=str(e) 

111 ) 

112 return False 

113 

114 

115def create_workspace_structure(workspace_path: Path) -> None: 

116 """Create the basic workspace directory structure. 

117 

118 Args: 

119 workspace_path: Path to the workspace directory 

120 

121 Raises: 

122 OSError: If directory creation fails 

123 """ 

124 logger.debug("Creating workspace structure", path=str(workspace_path)) 

125 

126 # Create workspace directory if it doesn't exist 

127 workspace_path.mkdir(parents=True, exist_ok=True) 

128 

129 # Create subdirectories 

130 logs_dir = workspace_path / "logs" 

131 logs_dir.mkdir(exist_ok=True) 

132 

133 metrics_dir = workspace_path / "metrics" 

134 metrics_dir.mkdir(exist_ok=True) 

135 

136 data_dir = workspace_path / "data" 

137 data_dir.mkdir(exist_ok=True) 

138 

139 logger.debug("Workspace structure created", workspace=str(workspace_path)) 

140 

141 

142def get_workspace_env_override(workspace_config: WorkspaceConfig) -> dict[str, str]: 

143 """Get environment variable overrides for workspace mode. 

144 

145 Args: 

146 workspace_config: Workspace configuration 

147 

148 Returns: 

149 dict: Environment variable overrides 

150 """ 

151 overrides = { 

152 "STATE_DB_PATH": str(workspace_config.database_path), 

153 } 

154 

155 logger.debug("Generated workspace environment overrides", overrides=overrides) 

156 return overrides 

157 

158 

159def validate_workspace_flags( 

160 workspace: Path | None, config: Path | None, env: Path | None 

161) -> None: 

162 """Validate that workspace flag is not used with conflicting flags. 

163 

164 Args: 

165 workspace: Workspace path (if provided) 

166 config: Config path (if provided) 

167 env: Env path (if provided) 

168 

169 Raises: 

170 ValueError: If conflicting flags are used 

171 """ 

172 if workspace is not None: 

173 if config is not None: 

174 raise ValueError( 

175 "Cannot use --workspace with --config flag. Use either workspace mode or individual file flags." 

176 ) 

177 

178 if env is not None: 

179 raise ValueError( 

180 "Cannot use --workspace with --env flag. Use either workspace mode or individual file flags." 

181 ) 

182 

183 logger.debug("Workspace flag validation passed")