Coverage for src/qdrant_loader/config/sources.py: 81%

1"""Sources configuration. 

2 

3This module defines the configuration for all data sources, including Git repositories, 

4Confluence spaces, Jira projects, and public documentation. 

5""" 

6 

7from typing import TYPE_CHECKING, Any 

8from pydantic import BaseModel, ConfigDict, Field 

9 

10from qdrant_loader.config.source_config import SourceConfig 

11from qdrant_loader.config.types import SourceType 

12 

13# Use TYPE_CHECKING to avoid circular imports at runtime 

14if TYPE_CHECKING: 

15 from qdrant_loader.connectors.confluence.config import ConfluenceSpaceConfig 

16 from qdrant_loader.connectors.git.config import GitRepoConfig 

17 from qdrant_loader.connectors.jira.config import JiraProjectConfig 

18 from qdrant_loader.connectors.localfile.config import LocalFileConfig 

19 from qdrant_loader.connectors.publicdocs.config import PublicDocsSourceConfig 

20 

21 

22def _get_connector_config_classes(): 

23 """Lazy import connector config classes to avoid circular dependencies.""" 

24 from qdrant_loader.connectors.confluence.config import ConfluenceSpaceConfig 

25 from qdrant_loader.connectors.git.config import GitRepoConfig 

26 from qdrant_loader.connectors.jira.config import JiraProjectConfig 

27 from qdrant_loader.connectors.localfile.config import LocalFileConfig 

28 from qdrant_loader.connectors.publicdocs.config import PublicDocsSourceConfig 

29 

30 return { 

31 "PublicDocsSourceConfig": PublicDocsSourceConfig, 

32 "GitRepoConfig": GitRepoConfig, 

33 "ConfluenceSpaceConfig": ConfluenceSpaceConfig, 

34 "JiraProjectConfig": JiraProjectConfig, 

35 "LocalFileConfig": LocalFileConfig, 

36 } 

37 

38 

39class SourcesConfig(BaseModel): 

40 """Configuration for all available data sources.""" 

41 

42 publicdocs: dict[str, Any] = Field( 

43 default_factory=dict, description="Public documentation sources" 

44 ) 

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

46 default_factory=dict, description="Git repository sources" 

47 ) 

48 confluence: dict[str, Any] = Field( 

49 default_factory=dict, description="Confluence space sources" 

50 ) 

51 jira: dict[str, Any] = Field( 

52 default_factory=dict, description="Jira project sources" 

53 ) 

54 localfile: dict[str, Any] = Field( 

55 default_factory=dict, description="Local file sources" 

56 ) 

57 

58 model_config = ConfigDict(arbitrary_types_allowed=False, extra="forbid") 

59 

60 def __init__(self, **data): 

61 """Initialize SourcesConfig with proper connector config objects.""" 

62 # Convert dictionaries to proper config objects 

63 processed_data = {} 

64 

65 for field_name, field_value in data.items(): 

66 if field_name in [ 

67 "publicdocs", 

68 "git", 

69 "confluence", 

70 "jira", 

71 "localfile", 

72 ] and isinstance(field_value, dict): 

73 processed_data[field_name] = self._convert_source_configs( 

74 field_name, field_value 

75 ) 

76 else: 

77 processed_data[field_name] = field_value 

78 

79 super().__init__(**processed_data) 

80 

81 def _convert_source_configs(self, source_type: str, configs: dict) -> dict: 

82 """Convert dictionary configs to proper config objects.""" 

83 config_classes = _get_connector_config_classes() 

84 converted = {} 

85 

86 for name, config_data in configs.items(): 

87 if isinstance(config_data, dict): 

88 # Get the appropriate config class 

89 if source_type == "publicdocs": 

90 config_class = config_classes["PublicDocsSourceConfig"] 

91 elif source_type == "git": 

92 config_class = config_classes["GitRepoConfig"] 

93 elif source_type == "confluence": 

94 config_class = config_classes["ConfluenceSpaceConfig"] 

95 elif source_type == "jira": 

96 config_class = config_classes["JiraProjectConfig"] 

97 elif source_type == "localfile": 

98 config_class = config_classes["LocalFileConfig"] 

99 else: 

100 # Unknown source type, keep as dict 

101 converted[name] = config_data 

102 continue 

103 

104 # Create the config object - let validation errors propagate 

105 try: 

106 converted[name] = config_class(**config_data) 

107 except (ImportError, AttributeError, TypeError) as e: 

108 # Only catch import/type errors, not validation errors 

109 # These indicate missing dependencies or code issues 

110 converted[name] = config_data 

111 # Let ValidationError and other Pydantic errors propagate 

112 else: 

113 # Already a config object or other type 

114 converted[name] = config_data 

115 

116 return converted 

117 

118 def get_source_config(self, source_type: str, source: str) -> SourceConfig | None: 

119 """Get the configuration for a specific source. 

120 

121 Args: 

122 source_type: Type of the source (publicdocs, git, confluence, jira) 

123 source: Name of the specific source configuration 

124 

125 Returns: 

126 Optional[BaseModel]: The source configuration if it exists, None otherwise 

127 """ 

128 source_dict = getattr(self, source_type, {}) 

129 return source_dict.get(source) 

130 

131 def to_dict(self) -> dict: 

132 """Convert the configuration to a dictionary.""" 

133 return { 

134 SourceType.PUBLICDOCS: { 

135 name: config.model_dump() if hasattr(config, "model_dump") else config 

136 for name, config in self.publicdocs.items() 

137 }, 

138 SourceType.GIT: { 

139 name: config.model_dump() if hasattr(config, "model_dump") else config 

140 for name, config in self.git.items() 

141 }, 

142 SourceType.CONFLUENCE: { 

143 name: config.model_dump() if hasattr(config, "model_dump") else config 

144 for name, config in self.confluence.items() 

145 }, 

146 SourceType.JIRA: { 

147 name: config.model_dump() if hasattr(config, "model_dump") else config 

148 for name, config in self.jira.items() 

149 }, 

150 SourceType.LOCALFILE: { 

151 name: config.model_dump() if hasattr(config, "model_dump") else config 

152 for name, config in self.localfile.items() 

153 }, 

154 }