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 

18 

19class ConfluenceSpaceConfig(SourceConfig): 

20 """Configuration for a Confluence space.""" 

21 

22 model_config = ConfigDict(arbitrary_types_allowed=True) 

23 

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

25 content_types: list[str] = Field( 

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

27 ) 

28 deployment_type: ConfluenceDeploymentType = Field( 

29 default=ConfluenceDeploymentType.CLOUD, 

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

31 ) 

32 token: str | None = Field( 

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

34 ) 

35 email: str | None = Field( 

36 default=None, 

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

38 ) 

39 

40 # Rate limiting 

41 requests_per_minute: int = Field( 

42 default=60, 

43 description="Maximum number of requests per minute for Confluence API", 

44 ge=1, 

45 le=1000, 

46 ) 

47 

48 include_labels: list[str] = Field( 

49 default=[], 

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

51 ) 

52 exclude_labels: list[str] = Field( 

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

54 ) 

55 

56 @field_validator("content_types") 

57 @classmethod 

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

59 """Validate content types.""" 

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

61 for content_type in v: 

62 if content_type.lower() not in valid_types: 

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

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

65 

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

67 @classmethod 

68 def auto_detect_deployment_type( 

69 cls, v: str | ConfluenceDeploymentType 

70 ) -> ConfluenceDeploymentType: 

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

72 if isinstance(v, str): 

73 return ConfluenceDeploymentType(v.lower()) 

74 return v 

75 

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

77 @classmethod 

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

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

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

81 

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

83 @classmethod 

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

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

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

87 

88 @model_validator(mode="after") 

89 def validate_auth_config(self) -> Self: 

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

91 if self.deployment_type == ConfluenceDeploymentType.CLOUD: 

92 # Cloud requires email and token 

93 if not self.email: 

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

95 if not self.token: 

96 raise ValueError( 

97 "API token is required for Confluence Cloud deployment" 

98 ) 

99 else: 

100 # Data Center/Server requires Personal Access Token 

101 if not self.token: 

102 raise ValueError( 

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

104 ) 

105 

106 return self