Coverage for src/qdrant_loader/config/state.py: 77%

1"""State management configuration. 

2 

3This module defines the configuration settings for state management, 

4including database path, table prefix, and connection pool settings. 

5""" 

6 

7import os 

8from pathlib import Path 

9from typing import Any 

10 

11from pydantic import Field, ValidationInfo, field_validator 

12 

13from qdrant_loader.config.base import BaseConfig 

14 

15 

16class DatabaseDirectoryError(Exception): 

17 """Exception raised when database directory needs to be created.""" 

18 

19 def __init__(self, path: Path): 

20 self.path = path 

21 super().__init__(f"Database directory does not exist: {path}") 

22 

23 

24class IngestionStatus: 

25 """Enum-like class for ingestion status values.""" 

26 

27 SUCCESS = "success" 

28 FAILED = "failed" 

29 IN_PROGRESS = "in_progress" 

30 SKIPPED = "skipped" 

31 CANCELLED = "cancelled" 

32 

33 

34class StateManagementConfig(BaseConfig): 

35 """Configuration for state management.""" 

36 

37 database_path: str = Field( 

38 default="./state.db", description="Path to SQLite database file" 

39 ) 

40 database_url: str | None = Field( 

41 default=None, 

42 repr=False, # may embed credentials; keep out of repr/logs/tracebacks 

43 description=( 

44 "Full SQLAlchemy database URL, e.g. " 

45 "postgresql+asyncpg://user:pass@host:5432/dbname. When set, " 

46 "overrides database_path and selects the backend by dialect. Leave " 

47 "unset to use the SQLite database_path (community default)." 

48 ), 

49 ) 

50 table_prefix: str = Field( 

51 default="qdrant_loader_", description="Prefix for database tables" 

52 ) 

53 connection_pool: dict[str, Any] = Field( 

54 default_factory=lambda: {"size": 5, "timeout": 30}, 

55 description="Connection pool settings", 

56 ) 

57 

58 @field_validator("database_path") 

59 @classmethod 

60 def validate_database_path(cls, v: str, info: ValidationInfo) -> str: 

61 """Validate database path.""" 

62 # Handle in-memory database 

63 if v in (":memory:", "sqlite:///:memory:"): 

64 return v 

65 

66 # Handle SQLite URLs 

67 if v.startswith("sqlite://"): 

68 # For SQLite URLs, skip file path validation since they might be 

69 # in-memory or use special formats 

70 return v 

71 

72 # For file paths, perform basic validation but allow directory creation 

73 try: 

74 # Expand environment variables, including $HOME 

75 expanded_path = os.path.expanduser(os.path.expandvars(v)) 

76 path = Path(expanded_path) 

77 

78 # Convert to absolute path for consistent handling 

79 if not path.is_absolute(): 

80 path = path.resolve() 

81 

82 # For absolute paths, use them as-is 

83 parent_dir = path.parent 

84 

85 # Check if parent directory exists 

86 if not parent_dir.exists(): 

87 # Don't fail here - let StateManager handle directory creation 

88 # Just validate that the path structure is reasonable 

89 try: 

90 # Test if the path is valid by trying to resolve it 

91 # Don't actually create the directory here 

92 parent_dir.resolve() 

93 

94 # Basic validation: ensure the path is reasonable 

95 # Note: We removed the arbitrary depth limit as it was too restrictive 

96 # for legitimate use cases like nested project structures and Windows paths 

97 

98 except OSError as e: 

99 raise ValueError( 

100 f"Invalid database path - cannot resolve directory {parent_dir}: {e}" 

101 ) 

102 else: 

103 # Directory exists, check if it's actually a directory and writable 

104 if not parent_dir.is_dir(): 

105 raise ValueError( 

106 f"Database directory path is not a directory: {parent_dir}" 

107 ) 

108 

109 if not os.access(str(parent_dir), os.W_OK): 

110 raise ValueError( 

111 f"Database directory is not writable: {parent_dir}" 

112 ) 

113 

114 except Exception as e: 

115 # If any validation fails, still allow the path through 

116 # StateManager will provide better error handling 

117 if isinstance(e, ValueError): 

118 raise # Re-raise validation errors 

119 # For other exceptions, just log and allow the path 

120 pass 

121 

122 # Return the original value to preserve any environment variables 

123 return v 

124 

125 @field_validator("database_url") 

126 @classmethod 

127 def validate_database_url(cls, v: str | None) -> str | None: 

128 if v is not None and "://" not in v: 

129 raise ValueError( 

130 "database_url must be a full SQLAlchemy URL (scheme://...)" 

131 ) 

132 return v 

133 

134 @field_validator("table_prefix") 

135 @classmethod 

136 def validate_table_prefix(cls, v: str, info: ValidationInfo) -> str: 

137 """Validate table prefix format.""" 

138 if not v: 

139 raise ValueError("Table prefix cannot be empty") 

140 if not v.replace("_", "").isalnum(): 

141 raise ValueError( 

142 "Table prefix can only contain alphanumeric characters and underscores" 

143 ) 

144 return v 

145 

146 @field_validator("connection_pool") 

147 @classmethod 

148 def validate_connection_pool( 

149 cls, v: dict[str, Any], info: ValidationInfo 

150 ) -> dict[str, Any]: 

151 """Validate connection pool settings.""" 

152 if "size" not in v: 

153 raise ValueError("Connection pool must specify 'size'") 

154 if not isinstance(v["size"], int) or v["size"] < 1: 

155 raise ValueError("Connection pool size must be a positive integer") 

156 

157 if "timeout" not in v: 

158 raise ValueError("Connection pool must specify 'timeout'") 

159 if not isinstance(v["timeout"], int) or v["timeout"] < 1: 

160 raise ValueError("Connection pool timeout must be a positive integer") 

161 

162 return v 

163 

164 def __init__(self, **data): 

165 """Initialize state management configuration.""" 

166 # If database_path is not provided, use default file-based database 

167 if "database_path" not in data: 

168 data["database_path"] = "./state.db" 

169 super().__init__(**data)