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 

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_auth_config(self) -> Self: 

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

100 if self.deployment_type == JiraDeploymentType.CLOUD: 

101 # Cloud requires email and token 

102 if not self.email: 

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

104 if not self.token: 

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

106 else: 

107 # Data Center/Server requires Personal Access Token 

108 if not self.token: 

109 raise ValueError( 

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

111 ) 

112 

113 return self 

114 

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

116 @classmethod 

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

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

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

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

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