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

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(..., description="Path to SQLite database file") 

38 table_prefix: str = Field( 

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

40 ) 

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

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

43 description="Connection pool settings", 

44 ) 

45 

46 @field_validator("database_path") 

47 @classmethod 

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

49 """Validate database path.""" 

50 # Expand environment variables, including $HOME 

51 path = Path(os.path.expanduser(os.path.expandvars(v))) 

52 parent_dir = path.parent 

53 

54 if not parent_dir.exists(): 

55 raise DatabaseDirectoryError(parent_dir) 

56 

57 if not parent_dir.is_dir(): 

58 raise ValueError("Database path is not a directory") 

59 

60 if not os.access(parent_dir, os.W_OK): 

61 raise ValueError("Database directory is not writable") 

62 

63 return v 

64 

65 @field_validator("table_prefix") 

66 @classmethod 

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

68 """Validate table prefix format.""" 

69 if not v: 

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

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

72 raise ValueError( 

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

74 ) 

75 return v 

76 

77 @field_validator("connection_pool") 

78 @classmethod 

79 def validate_connection_pool( 

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

81 ) -> dict[str, Any]: 

82 """Validate connection pool settings.""" 

83 if "size" not in v: 

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

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

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

87 

88 if "timeout" not in v: 

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

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

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

92 

93 return v 

94 

95 def __init__(self, **data): 

96 """Initialize state management configuration.""" 

97 # If database_path is not provided, use in-memory database 

98 if "database_path" not in data: 

99 data["database_path"] = ":memory:" 

100 super().__init__(**data)