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

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 

9 

10def _get_logger(): 

11 return LoggingConfig.get_logger(__name__) 

12 

13 

14@dataclass 

15class WorkspaceConfig: 

16 """Configuration for workspace mode.""" 

17 

18 workspace_path: Path 

19 config_path: Path 

20 env_path: Path | None 

21 logs_path: Path 

22 metrics_path: Path 

23 database_path: Path 

24 

25 def __post_init__(self): 

26 """Validate workspace configuration after initialization.""" 

27 # Ensure workspace path is absolute 

28 self.workspace_path = self.workspace_path.resolve() 

29 

30 # Validate workspace directory exists 

31 if not self.workspace_path.exists(): 

32 raise ValueError( 

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

34 ) 

35 

36 if not self.workspace_path.is_dir(): 

37 raise ValueError( 

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

39 ) 

40 

41 # Validate config.yaml exists 

42 if not self.config_path.exists(): 

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

44 

45 # Validate workspace is writable 

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

47 raise ValueError( 

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

49 ) 

50 

51 _get_logger().debug( 

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

53 ) 

54 

55 

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

57 """Setup and validate workspace configuration. 

58 

59 Args: 

60 workspace_path: Path to the workspace directory 

61 

62 Returns: 

63 WorkspaceConfig: Validated workspace configuration 

64 

65 Raises: 

66 ValueError: If workspace validation fails 

67 """ 

68 _get_logger().debug("Setting up workspace", path=str(workspace_path)) 

69 

70 # Resolve to absolute path 

71 workspace_path = workspace_path.resolve() 

72 

73 # Define workspace file paths 

74 config_path = workspace_path / "config.yaml" 

75 env_path = workspace_path / ".env" 

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

77 metrics_path = workspace_path / "metrics" 

78 data_path = workspace_path / "data" 

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

80 

81 # Check if .env file exists (optional) 

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

83 

84 # Create workspace config 

85 workspace_config = WorkspaceConfig( 

86 workspace_path=workspace_path, 

87 config_path=config_path, 

88 env_path=env_path_final, 

89 logs_path=logs_path, 

90 metrics_path=metrics_path, 

91 database_path=database_path, 

92 ) 

93 

94 _get_logger().debug("Workspace setup completed", workspace=str(workspace_path)) 

95 return workspace_config 

96 

97 

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

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

100 

101 Args: 

102 workspace_path: Path to validate 

103 

104 Returns: 

105 bool: True if valid workspace, False otherwise 

106 """ 

107 try: 

108 setup_workspace(workspace_path) 

109 return True 

110 except ValueError as e: 

111 _get_logger().debug( 

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

113 ) 

114 return False 

115 

116 

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

118 """Create the basic workspace directory structure. 

119 

120 Args: 

121 workspace_path: Path to the workspace directory 

122 

123 Raises: 

124 OSError: If directory creation fails 

125 """ 

126 _get_logger().debug("Creating workspace structure", path=str(workspace_path)) 

127 

128 # Create workspace directory if it doesn't exist 

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

130 

131 # Create subdirectories 

132 logs_dir = workspace_path / "logs" 

133 logs_dir.mkdir(exist_ok=True) 

134 

135 metrics_dir = workspace_path / "metrics" 

136 metrics_dir.mkdir(exist_ok=True) 

137 

138 data_dir = workspace_path / "data" 

139 data_dir.mkdir(exist_ok=True) 

140 

141 _get_logger().debug("Workspace structure created", workspace=str(workspace_path)) 

142 

143 

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

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

146 

147 Args: 

148 workspace_config: Workspace configuration 

149 

150 Returns: 

151 dict: Environment variable overrides 

152 """ 

153 overrides = { 

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

155 } 

156 

157 _get_logger().debug( 

158 "Generated workspace environment overrides", overrides=overrides 

159 ) 

160 return overrides 

161 

162 

163def validate_workspace_flags( 

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

165) -> None: 

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

167 

168 Args: 

169 workspace: Workspace path (if provided) 

170 config: Config path (if provided) 

171 env: Env path (if provided) 

172 

173 Raises: 

174 ValueError: If conflicting flags are used 

175 """ 

176 if workspace is not None: 

177 if config is not None: 

178 raise ValueError( 

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

180 ) 

181 

182 if env is not None: 

183 raise ValueError( 

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

185 ) 

186 

187 _get_logger().debug("Workspace flag validation passed")