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

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 

8 

9from pydantic import BaseModel, ConfigDict, Field 

10 

11from qdrant_loader.config.source_config import SourceConfig 

12from qdrant_loader.config.types import SourceType 

13 

14# Use TYPE_CHECKING to avoid circular imports at runtime 

15if TYPE_CHECKING: 

16 pass 

17 

18 

19def _get_connector_config_classes(): 

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

21 from qdrant_loader.connectors.confluence.config import ConfluenceSpaceConfig 

22 from qdrant_loader.connectors.git.config import GitRepoConfig 

23 from qdrant_loader.connectors.jira.config import JiraProjectConfig 

24 from qdrant_loader.connectors.localfile.config import LocalFileConfig 

25 from qdrant_loader.connectors.publicdocs.config import PublicDocsSourceConfig 

26 

27 return { 

28 "PublicDocsSourceConfig": PublicDocsSourceConfig, 

29 "GitRepoConfig": GitRepoConfig, 

30 "ConfluenceSpaceConfig": ConfluenceSpaceConfig, 

31 "JiraProjectConfig": JiraProjectConfig, 

32 "LocalFileConfig": LocalFileConfig, 

33 } 

34 

35 

36class SourcesConfig(BaseModel): 

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

38 

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

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

41 ) 

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

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

44 ) 

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

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

47 ) 

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

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

50 ) 

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

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

53 ) 

54 

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

56 

57 def __init__(self, **data): 

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

59 # Convert dictionaries to proper config objects 

60 processed_data = {} 

61 

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

63 if field_name in [ 

64 "publicdocs", 

65 "git", 

66 "confluence", 

67 "jira", 

68 "localfile", 

69 ] and isinstance(field_value, dict): 

70 processed_data[field_name] = self._convert_source_configs( 

71 field_name, field_value 

72 ) 

73 else: 

74 processed_data[field_name] = field_value 

75 

76 super().__init__(**processed_data) 

77 

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

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

80 config_classes = _get_connector_config_classes() 

81 converted = {} 

82 

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

84 if isinstance(config_data, dict): 

85 # Get the appropriate config class 

86 if source_type == "publicdocs": 

87 config_class = config_classes["PublicDocsSourceConfig"] 

88 elif source_type == "git": 

89 config_class = config_classes["GitRepoConfig"] 

90 elif source_type == "confluence": 

91 config_class = config_classes["ConfluenceSpaceConfig"] 

92 elif source_type == "jira": 

93 config_class = config_classes["JiraProjectConfig"] 

94 elif source_type == "localfile": 

95 config_class = config_classes["LocalFileConfig"] 

96 else: 

97 # Unknown source type, keep as dict 

98 converted[name] = config_data 

99 continue 

100 

101 # Create the config object - let validation errors propagate 

102 try: 

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

104 except (ImportError, AttributeError, TypeError): 

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

106 # These indicate missing dependencies or code issues 

107 converted[name] = config_data 

108 # Let ValidationError and other Pydantic errors propagate 

109 else: 

110 # Already a config object or other type 

111 converted[name] = config_data 

112 

113 return converted 

114 

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

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

117 

118 Args: 

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

120 source: Name of the specific source configuration 

121 

122 Returns: 

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

124 """ 

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

126 return source_dict.get(source) 

127 

128 def to_dict(self) -> dict: 

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

130 return { 

131 SourceType.PUBLICDOCS: { 

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

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

134 }, 

135 SourceType.GIT: { 

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

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

138 }, 

139 SourceType.CONFLUENCE: { 

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

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

142 }, 

143 SourceType.JIRA: { 

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

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

146 }, 

147 SourceType.LOCALFILE: { 

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

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

150 }, 

151 }