Coverage for src/qdrant_loader/connectors/confluence/config.py: 90%

1"""Configuration for Confluence connector.""" 

2 

3import os 

4from enum import Enum 

5from typing import Self 

6 

7from pydantic import ConfigDict, Field, field_validator, model_validator 

8 

9from qdrant_loader.config.source_config import SourceConfig 

10 

11 

12class ConfluenceDeploymentType(str, Enum): 

13 """Confluence deployment types.""" 

14 

15 CLOUD = "cloud" 

16 DATACENTER = "datacenter" 

17 SERVER = "server" # Legacy, treated same as datacenter 

18 

19 

20class ConfluenceSpaceConfig(SourceConfig): 

21 """Configuration for a Confluence space.""" 

22 

23 model_config = ConfigDict(arbitrary_types_allowed=True) 

24 

25 space_key: str = Field(..., description="Key of the Confluence space") 

26 content_types: list[str] = Field( 

27 default=["page", "blogpost"], description="Types of content to process" 

28 ) 

29 deployment_type: ConfluenceDeploymentType = Field( 

30 default=ConfluenceDeploymentType.CLOUD, 

31 description="Confluence deployment type (cloud, datacenter, or server)", 

32 ) 

33 token: str | None = Field( 

34 ..., description="Confluence API token or Personal Access Token" 

35 ) 

36 email: str | None = Field( 

37 default=None, 

38 description="Email associated with the Confluence account (Cloud only)", 

39 ) 

40 

41 include_labels: list[str] = Field( 

42 default=[], 

43 description="List of labels to include (empty list means include all)", 

44 ) 

45 exclude_labels: list[str] = Field( 

46 default=[], description="List of labels to exclude" 

47 ) 

48 

49 @field_validator("content_types") 

50 @classmethod 

51 def validate_content_types(cls, v: list[str]) -> list[str]: 

52 """Validate content types.""" 

53 valid_types = ["page", "blogpost", "comment"] 

54 for content_type in v: 

55 if content_type.lower() not in valid_types: 

56 raise ValueError(f"Content type must be one of {valid_types}") 

57 return [t.lower() for t in v] 

58 

59 @field_validator("deployment_type", mode="before") 

60 @classmethod 

61 def auto_detect_deployment_type( 

62 cls, v: str | ConfluenceDeploymentType 

63 ) -> ConfluenceDeploymentType: 

64 """Auto-detect deployment type if not specified.""" 

65 if isinstance(v, str): 

66 return ConfluenceDeploymentType(v.lower()) 

67 return v 

68 

69 @field_validator("token", mode="after") 

70 @classmethod 

71 def load_token_from_env(cls, v: str | None) -> str | None: 

72 """Load token from environment variable if not provided.""" 

73 return v or os.getenv("CONFLUENCE_TOKEN") 

74 

75 @field_validator("email", mode="after") 

76 @classmethod 

77 def load_email_from_env(cls, v: str | None) -> str | None: 

78 """Load email from environment variable if not provided.""" 

79 return v or os.getenv("CONFLUENCE_EMAIL") 

80 

81 @model_validator(mode="after") 

82 def validate_auth_config(self) -> Self: 

83 """Validate authentication configuration based on deployment type.""" 

84 if self.deployment_type == ConfluenceDeploymentType.CLOUD: 

85 # Cloud requires email and token 

86 if not self.email: 

87 raise ValueError("Email is required for Confluence Cloud deployment") 

88 if not self.token: 

89 raise ValueError( 

90 "API token is required for Confluence Cloud deployment" 

91 ) 

92 else: 

93 # Data Center/Server requires Personal Access Token 

94 if not self.token: 

95 raise ValueError( 

96 "Personal Access Token is required for Confluence Data Center/Server deployment" 

97 ) 

98 

99 return self