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

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

2 

3import os 

4from enum import Enum 

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(str, Enum): 

13 """Jira deployment types.""" 

14 

15 CLOUD = "cloud" 

16 DATACENTER = "datacenter" 

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

18 

19 

20class JiraProjectConfig(SourceConfig): 

21 """Configuration for a Jira project.""" 

22 

23 # Authentication 

24 token: str | None = Field( 

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

26 ) 

27 email: str | None = Field( 

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

29 ) 

30 base_url: HttpUrl = Field( 

31 ..., 

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

33 ) 

34 

35 # Project configuration 

36 project_key: str = Field( 

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

38 ) 

39 

40 # Deployment type 

41 deployment_type: JiraDeploymentType = Field( 

42 default=JiraDeploymentType.CLOUD, 

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

44 ) 

45 

46 # Rate limiting 

47 requests_per_minute: int = Field( 

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

49 ) 

50 

51 # Pagination 

52 page_size: int = Field( 

53 default=100, 

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

55 ge=1, 

56 le=100, 

57 ) 

58 

59 # Attachment handling 

60 download_attachments: bool = Field( 

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

62 ) 

63 process_attachments: bool = Field( 

64 default=True, 

65 description="Whether to process issue attachments (deprecated, use download_attachments)", 

66 ) 

67 

68 # Additional configuration 

69 issue_types: list[str] = Field( 

70 default=[], 

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

72 ) 

73 include_statuses: list[str] = Field( 

74 default=[], 

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

76 ) 

77 

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

79 

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

81 @classmethod 

82 def auto_detect_deployment_type( 

83 cls, v: str | JiraDeploymentType 

84 ) -> JiraDeploymentType: 

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

86 if isinstance(v, str): 

87 return JiraDeploymentType(v.lower()) 

88 return v 

89 

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

91 @classmethod 

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

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

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

95 

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

97 @classmethod 

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

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

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

101 

102 @model_validator(mode="after") 

103 def validate_auth_config(self) -> Self: 

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

105 if self.deployment_type == JiraDeploymentType.CLOUD: 

106 # Cloud requires email and token 

107 if not self.email: 

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

109 if not self.token: 

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

111 else: 

112 # Data Center/Server requires Personal Access Token 

113 if not self.token: 

114 raise ValueError( 

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

116 ) 

117 

118 return self 

119 

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

121 @classmethod 

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

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

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

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

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