Coverage for src / qdrant_loader / connectors / jira / config.py: 95%

1"""Configuration for Jira connector.""" 

2 

3import os 

4from enum import StrEnum 

5from typing import Self 

6 

7from pydantic import ConfigDict, Field, HttpUrl, field_validator, model_validator 

8 

9from qdrant_loader.config.source_config import SourceConfig 

10 

11 

12class JiraDeploymentType(StrEnum): 

13 """Jira deployment types.""" 

14 

15 CLOUD = "cloud" 

16 DATACENTER = "datacenter" 

17 

18 

19class JiraProjectConfig(SourceConfig): 

20 """Configuration for a Jira project.""" 

21 

22 # Authentication 

23 token: str | None = Field( 

24 default=None, description="Jira API token or Personal Access Token" 

25 ) 

26 email: str | None = Field( 

27 default=None, description="Email associated with the API token (Cloud only)" 

28 ) 

29 base_url: HttpUrl = Field( 

30 ..., 

31 description="Base URL of the Jira instance (e.g., 'https://your-domain.atlassian.net')", 

32 ) 

33 

34 # Project configuration 

35 project_key: str = Field( 

36 ..., description="Project key to process (e.g., 'PROJ')", min_length=1 

37 ) 

38 

39 # Deployment type 

40 deployment_type: JiraDeploymentType = Field( 

41 default=JiraDeploymentType.CLOUD, 

42 description="Jira deployment type (cloud, datacenter, or server)", 

43 ) 

44 

45 # Rate limiting 

46 requests_per_minute: int = Field( 

47 default=60, description="Maximum number of requests per minute", ge=1, le=1000 

48 ) 

49 

50 # Pagination 

51 page_size: int = Field( 

52 default=100, 

53 description="Number of items per page for paginated requests", 

54 ge=1, 

55 le=100, 

56 ) 

57 

58 # Attachment handling 

59 download_attachments: bool = Field( 

60 default=False, description="Whether to download and process issue attachments" 

61 ) 

62 

63 # Additional configuration 

64 issue_types: list[str] = Field( 

65 default=[], 

66 description="Optional list of issue types to process (e.g., ['Bug', 'Story']). If empty, all types are processed.", 

67 ) 

68 include_statuses: list[str] = Field( 

69 default=[], 

70 description="Optional list of statuses to include (e.g., ['Open', 'In Progress']). If empty, all statuses are included.", 

71 ) 

72 

73 model_config = ConfigDict(validate_default=True, arbitrary_types_allowed=True) 

74 

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

76 @classmethod 

77 def auto_detect_deployment_type( 

78 cls, v: str | JiraDeploymentType 

79 ) -> JiraDeploymentType: 

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

81 if isinstance(v, str): 

82 return JiraDeploymentType(v.lower()) 

83 return v 

84 

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

86 @classmethod 

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

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

89 return v or os.getenv("JIRA_TOKEN") 

90 

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

92 @classmethod 

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

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

95 return v or os.getenv("JIRA_EMAIL") 

96 

97 @model_validator(mode="after") 

98 def validate_no_placeholders(self) -> Self: 

99 """Fail immediately if any required field still contains an un-substituted ${VAR} placeholder.""" 

100 import re 

101 

102 _placeholder = re.compile(r"\$\{[^}]+\}") 

103 

104 fields_to_check: dict[str, str | None] = { 

105 "project_key": self.project_key, 

106 "base_url": str(self.base_url) if self.base_url else None, 

107 "token": self.token, 

108 "email": self.email, 

109 } 

110 

111 bad: list[str] = [] 

112 for field_name, value in fields_to_check.items(): 

113 if value and _placeholder.search(value): 

114 # Extract the variable name for a helpful hint 

115 var = _placeholder.search(value).group(0) # type: ignore[union-attr] 

116 bad.append(f" - {field_name}: {var} (env var not set)") 

117 

118 if bad: 

119 raise ValueError( 

120 "Jira source config contains un-substituted environment variables.\n" 

121 "Set the following variables in your .env file or shell before running:\n" 

122 + "\n".join(bad) 

123 ) 

124 

125 return self 

126 

127 @model_validator(mode="after") 

128 def validate_auth_config(self) -> Self: 

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

130 if self.deployment_type == JiraDeploymentType.CLOUD: 

131 # Cloud requires email and token 

132 if not self.email: 

133 raise ValueError("Email is required for Jira Cloud deployment") 

134 if not self.token: 

135 raise ValueError("API token is required for Jira Cloud deployment") 

136 else: 

137 # Data Center/Server requires Personal Access Token 

138 if not self.token: 

139 raise ValueError( 

140 "Personal Access Token is required for Jira Data Center/Server deployment" 

141 ) 

142 

143 return self 

144 

145 @field_validator("issue_types", "include_statuses") 

146 @classmethod 

147 def validate_list_items(cls, v: list[str]) -> list[str]: 

148 """Validate that list items are not empty strings.""" 

149 if any(not item.strip() for item in v): 

150 raise ValueError("List items cannot be empty strings") 

151 return [item.strip() for item in v]