Coverage for src/qdrant_loader/connectors/jira/data_center_connector.py: 88%

1"""Jira connector implementation.""" 

2 

3from collections.abc import AsyncGenerator 

4from datetime import datetime 

5from urllib.parse import urlparse # noqa: F401 - may be used in URL handling 

6 

7from requests.auth import HTTPBasicAuth # noqa: F401 - compatibility 

8 

9from qdrant_loader.connectors.jira.connector import BaseJiraConnector 

10from qdrant_loader.connectors.jira.models import ( 

11 JiraIssue, 

12) 

13from qdrant_loader.utils.logging import LoggingConfig 

14 

15logger = LoggingConfig.get_logger(__name__) 

16 

17 

18class JiraDataCenterConnector(BaseJiraConnector): 

19 """Jira data center connector for fetching and processing issues.""" 

20 

21 SEARCH_ENDPOINT = "search" 

22 

23 def _get_api_url(self, endpoint: str) -> str: 

24 """Construct the full API URL for an endpoint. 

25 

26 Args: 

27 endpoint: API endpoint path 

28 

29 Returns: 

30 str: Full API URL 

31 """ 

32 return f"{self.base_url}/rest/api/2/{endpoint}" 

33 

34 async def get_issues( 

35 self, updated_after: datetime | None = None 

36 ) -> AsyncGenerator[JiraIssue, None]: 

37 """ 

38 Get all issues from Jira. 

39 

40 Args: 

41 updated_after: Optional datetime to filter issues updated after this time 

42 

43 Yields: 

44 JiraIssue objects 

45 """ 

46 # Resume from checkpoint if provided (WS-2 feature) 

47 # For DataCenter, checkpoint_cursor is the startAt offset as a string 

48 try: 

49 start_at = int(self._checkpoint_cursor) if self._checkpoint_cursor else 0 

50 except (ValueError, TypeError): 

51 start_at = 0 

52 

53 page_size = self.config.page_size 

54 total_issues = 0 

55 processed_count = 0 

56 

57 if self._checkpoint_cursor: 

58 logger.info( 

59 "🎫 Resuming JIRA issue retrieval from checkpoint", 

60 project_key=self.config.project_key, 

61 page_size=page_size, 

62 start_at=start_at, 

63 ) 

64 else: 

65 logger.info( 

66 "🎫 Starting JIRA issue retrieval", 

67 project_key=self.config.project_key, 

68 page_size=page_size, 

69 updated_after=updated_after.isoformat() if updated_after else None, 

70 ) 

71 

72 while True: 

73 jql = self._build_jql_filter(updated_after) 

74 

75 params = { 

76 "jql": jql, 

77 "startAt": start_at, 

78 "maxResults": page_size, 

79 "expand": "changelog", 

80 "fields": "*all", 

81 } 

82 

83 logger.debug( 

84 "Fetching JIRA issues page", 

85 start_at=start_at, 

86 page_size=page_size, 

87 jql=jql, 

88 ) 

89 

90 try: 

91 response = await self._make_request( 

92 "GET", self.SEARCH_ENDPOINT, params=params 

93 ) 

94 except Exception as e: 

95 logger.error( 

96 "Failed to fetch JIRA issues page", 

97 start_at=start_at, 

98 page_size=page_size, 

99 error=str(e), 

100 error_type=type(e).__name__, 

101 ) 

102 raise 

103 

104 if not response or not response.get("issues"): 

105 logger.debug( 

106 "No more JIRA issues found, stopping pagination", 

107 start_at=start_at, 

108 total_processed=start_at, 

109 issues_processed=processed_count, 

110 ) 

111 break 

112 

113 issues = response["issues"] 

114 # Compute the offset for the next page and attach as checkpoint info 

115 page_next_offset = str(start_at + len(issues)) 

116 

117 # Update total count if not set 

118 if total_issues == 0: 

119 total_issues = response.get("total", 0) 

120 logger.info(f"🎫 Found {total_issues} JIRA issues to process") 

121 

122 # Log progress every 100 issues instead of every 50 

123 progress_log_interval = 100 

124 

125 for i, issue in enumerate(issues): 

126 try: 

127 parsed_issue = self._parse_issue(issue, self.config.extra_fields) 

128 # Attach checkpoint info for this page 

129 parsed_issue.ingestion_checkpoint = { 

130 "cursor_kind": "jql_window", 

131 "cursor_value": page_next_offset, 

132 "batch_index": 0, 

133 } 

134 yield parsed_issue 

135 processed_count += 1 

136 

137 if (start_at + i + 1) % progress_log_interval == 0: 

138 progress_percent = ( 

139 round((start_at + i + 1) / total_issues * 100, 1) 

140 if total_issues > 0 

141 else 0 

142 ) 

143 logger.info( 

144 f"🎫 Progress: {start_at + i + 1}/{total_issues} issues ({progress_percent}%)" 

145 ) 

146 

147 except Exception as e: 

148 logger.error( 

149 "Failed to parse JIRA issue", 

150 issue_id=issue.get("id"), 

151 issue_key=issue.get("key"), 

152 error=str(e), 

153 error_type=type(e).__name__, 

154 ) 

155 # Continue processing other issues instead of failing completely 

156 continue 

157 

158 # Check if we've processed all issues and store checkpoint 

159 start_at += len(issues) 

160 

161 if start_at >= total_issues: 

162 logger.info( 

163 f"✅ Completed JIRA issue retrieval: " 

164 f"{start_at} issues attempted, " 

165 f"{processed_count} successfully processed" 

166 ) 

167 break