Coverage for src/qdrant_loader/connectors/jira/cloud_connector.py: 93%

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 JiraCloudConnector(BaseJiraConnector): 

19 """Jira cloud connector for fetching and processing issues.""" 

20 

21 CLOUD_JIRA_VERSION = "3" 

22 SEARCH_ENDPOINT = "search/jql" 

23 

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

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

26 

27 Args: 

28 endpoint: API endpoint path 

29 

30 Returns: 

31 str: Full API URL 

32 """ 

33 

34 return f"{self.base_url}/rest/api/{self.CLOUD_JIRA_VERSION}/{endpoint}" 

35 

36 async def get_issues( 

37 self, updated_after: datetime | None = None 

38 ) -> AsyncGenerator[JiraIssue, None]: 

39 """ 

40 Get all issues from Jira. 

41 

42 Args: 

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

44 

45 Yields: 

46 JiraIssue objects 

47 """ 

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

49 next_page_token: str | None = self._checkpoint_cursor 

50 processed_count = 0 

51 page_size = self.config.page_size 

52 attempted_count = 0 

53 # Log progress every 100 issues instead of every 50 

54 progress_log_interval = 100 

55 

56 if next_page_token: 

57 logger.info( 

58 "🎫 Resuming JIRA issue retrieval from checkpoint", 

59 project_key=self.config.project_key, 

60 page_size=page_size, 

61 next_page_token=next_page_token, 

62 ) 

63 else: 

64 logger.info( 

65 "🎫 Starting JIRA issue retrieval", 

66 project_key=self.config.project_key, 

67 page_size=page_size, 

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

69 ) 

70 

71 while True: 

72 jql = self._build_jql_filter(updated_after) 

73 

74 params = { 

75 "jql": jql, 

76 "maxResults": page_size, 

77 "expand": "changelog", 

78 "fields": "*all", 

79 } 

80 

81 if next_page_token: 

82 params["nextPageToken"] = next_page_token 

83 

84 logger.debug( 

85 "Fetching JIRA issues page", 

86 next_page_token=next_page_token, 

87 page_size=page_size, 

88 jql=jql, 

89 ) 

90 

91 try: 

92 response = await self._make_request( 

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

94 ) 

95 except Exception as e: 

96 logger.error( 

97 "Failed to fetch JIRA issues page", 

98 next_page_token=next_page_token, 

99 page_size=page_size, 

100 error=str(e), 

101 error_type=type(e).__name__, 

102 ) 

103 raise 

104 

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

106 logger.debug( 

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

108 next_page_token=next_page_token, 

109 total_processed=processed_count, 

110 ) 

111 break 

112 

113 issues = response["issues"] 

114 # Token that will be used to fetch the next page; save with documents 

115 page_next_token = response.get("nextPageToken") 

116 

117 for issue in issues: 

118 try: 

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

120 # Attach checkpoint info to the issue so downstream mapping 

121 # can include it in Document metadata for checkpoint saving. 

122 # Only attach checkpoint if there's a next page to resume from 

123 if page_next_token: 

124 parsed_issue.ingestion_checkpoint = { 

125 "cursor_kind": "page_token", 

126 "cursor_value": page_next_token, 

127 "batch_index": 0, 

128 } 

129 yield parsed_issue 

130 processed_count += 1 

131 

132 if (processed_count) % progress_log_interval == 0: 

133 logger.info( 

134 f"🎫 Processed {processed_count} JIRA issues so far" 

135 ) 

136 

137 except Exception as e: 

138 logger.error( 

139 "Failed to parse JIRA issue", 

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

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

142 error=str(e), 

143 error_type=type(e).__name__, 

144 ) 

145 # Continue processing other issues instead of failing completely 

146 continue 

147 

148 attempted_count += len(issues) 

149 # Check next page token and save it for checkpoint 

150 next_page_token = response.get("nextPageToken") 

151 is_last = response.get("isLast") 

152 

153 if is_last or not next_page_token: 

154 logger.info( 

155 f"✅ Completed JIRA issue retrieval: " 

156 f"{attempted_count} issues attempted, " 

157 f"{processed_count} successfully processed" 

158 ) 

159 break