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

1"""Configuration models for multi-project support. 

2 

3This module defines the data structures used for multi-project configuration, 

4including project contexts, project configurations, and related models. 

5""" 

6 

7from dataclasses import dataclass 

8from datetime import datetime 

9from typing import Any 

10 

11from pydantic import BaseModel, ConfigDict, Field, field_serializer 

12 

13from .sources import SourcesConfig 

14 

15 

16@dataclass 

17class ProjectContext: 

18 """Project context information passed through the pipeline.""" 

19 

20 project_id: str 

21 display_name: str 

22 description: str | None 

23 collection_name: str 

24 config_overrides: dict[str, Any] 

25 

26 def __post_init__(self): 

27 """Validate project context after initialization.""" 

28 if not self.project_id: 

29 raise ValueError("project_id cannot be empty") 

30 if not self.display_name: 

31 raise ValueError("display_name cannot be empty") 

32 if not self.collection_name: 

33 raise ValueError("collection_name cannot be empty") 

34 

35 

36class ProjectConfig(BaseModel): 

37 """Configuration for a single project.""" 

38 

39 project_id: str = Field(..., description="Unique project identifier") 

40 display_name: str = Field(..., description="Human-readable project name") 

41 description: str | None = Field(None, description="Project description") 

42 sources: SourcesConfig = Field( 

43 default_factory=SourcesConfig, description="Project-specific sources" 

44 ) 

45 overrides: dict[str, Any] = Field( 

46 default_factory=dict, description="Project-specific configuration overrides" 

47 ) 

48 

49 def get_effective_collection_name(self, global_collection_name: str) -> str: 

50 """Get the effective collection name for this project. 

51 

52 Args: 

53 global_collection_name: The global collection name from configuration 

54 

55 Returns: 

56 The collection name to use for this project (always the global collection name) 

57 """ 

58 # Always use the global collection name for all projects 

59 return global_collection_name 

60 

61 

62class ProjectsConfig(BaseModel): 

63 """Configuration for multiple projects.""" 

64 

65 projects: dict[str, ProjectConfig] = Field( 

66 default_factory=dict, description="Project configurations" 

67 ) 

68 

69 def get_project(self, project_id: str) -> ProjectConfig | None: 

70 """Get a project configuration by ID. 

71 

72 Args: 

73 project_id: The project identifier 

74 

75 Returns: 

76 The project configuration if found, None otherwise 

77 """ 

78 return self.projects.get(project_id) 

79 

80 def list_project_ids(self) -> list[str]: 

81 """Get a list of all project IDs. 

82 

83 Returns: 

84 List of project identifiers 

85 """ 

86 return list(self.projects.keys()) 

87 

88 def add_project(self, project_config: ProjectConfig) -> None: 

89 """Add a project configuration. 

90 

91 Args: 

92 project_config: The project configuration to add 

93 

94 Raises: 

95 ValueError: If project ID already exists 

96 """ 

97 if project_config.project_id in self.projects: 

98 raise ValueError(f"Project '{project_config.project_id}' already exists") 

99 

100 self.projects[project_config.project_id] = project_config 

101 

102 def to_dict(self) -> dict[str, Any]: 

103 """Convert the projects configuration to a dictionary. 

104 

105 Returns: 

106 Dict[str, Any]: Projects configuration as a dictionary 

107 """ 

108 return { 

109 project_id: project_config.model_dump() 

110 for project_id, project_config in self.projects.items() 

111 } 

112 

113 

114@dataclass 

115class ParsedConfig: 

116 """Result of parsing a configuration file.""" 

117 

118 global_config: Any # Will be GlobalConfig, but avoiding circular import 

119 projects_config: ProjectsConfig 

120 

121 def get_all_projects(self) -> list[ProjectConfig]: 

122 """Get all project configurations. 

123 

124 Returns: 

125 List of all project configurations 

126 """ 

127 return list(self.projects_config.projects.values()) 

128 

129 

130class ProjectStats(BaseModel): 

131 """Statistics for a project.""" 

132 

133 project_id: str = Field(..., description="Project identifier") 

134 document_count: int = Field(default=0, description="Number of documents in project") 

135 source_count: int = Field(default=0, description="Number of sources in project") 

136 last_updated: datetime | None = Field(None, description="Last update timestamp") 

137 storage_size: int | None = Field(None, description="Storage size in bytes") 

138 

139 model_config = ConfigDict() 

140 

141 @field_serializer("last_updated") 

142 def serialize_last_updated(self, value: datetime | None) -> str | None: 

143 return value.isoformat() if value else None 

144 

145 

146class ProjectInfo(BaseModel): 

147 """Detailed information about a project.""" 

148 

149 id: str = Field(..., description="Project identifier") 

150 display_name: str = Field(..., description="Project display name") 

151 description: str | None = Field(None, description="Project description") 

152 collection_name: str = Field(..., description="QDrant collection name") 

153 source_count: int = Field(default=0, description="Number of sources") 

154 document_count: int = Field(default=0, description="Number of documents") 

155 last_updated: datetime | None = Field(None, description="Last update timestamp") 

156 

157 model_config = ConfigDict() 

158 

159 @field_serializer("last_updated") 

160 def serialize_last_updated(self, value: datetime | None) -> str | None: 

161 return value.isoformat() if value else None 

162 

163 

164class ProjectDetail(ProjectInfo): 

165 """Detailed project information including sources and statistics.""" 

166 

167 sources: list[dict[str, Any]] = Field( 

168 default_factory=list, description="Source information" 

169 ) 

170 statistics: dict[str, Any] = Field( 

171 default_factory=dict, description="Project statistics" 

172 )