Coverage for chalicelib/utils/jira_client.py: 12%
237 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1import logging
2import time
3from datetime import datetime
5import requests
6from fastapi import HTTPException, status
7from jira import JIRA
8from jira.exceptions import JIRAError
9from requests.auth import HTTPBasicAuth
11logger = logging.getLogger(__name__)
12fields = "id, summary, description, creator, reporter, created, assignee, status, updated, comment, issuetype, labels"
15class JiraManager:
16 retries = 0
18 def __init__(self, url, username, password, project_id=None):
19 self._config = {"JIRA_PROJECT_ID": project_id, "JIRA_URL": url, "JIRA_USERNAME": username,
20 "JIRA_PASSWORD": password}
21 try:
22 self._jira = JIRA(url, basic_auth=(username, password), logging=True, max_retries=0, timeout=3)
23 except Exception as e:
24 logger.warning("!!! JIRA AUTH ERROR")
25 logger.error(e)
26 raise e
28 def set_jira_project_id(self, project_id):
29 self._config["JIRA_PROJECT_ID"] = project_id
31 def get_projects(self):
32 try:
33 projects = self._jira.projects()
34 except JIRAError as e:
35 self.retries -= 1
36 if (e.status_code // 100) == 4 and self.retries > 0:
37 time.sleep(1)
38 return self.get_projects()
39 logger.error(f"=>JIRA Exception {e.text}")
40 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
41 projects_dict_list = []
42 for project in projects:
43 projects_dict_list.append(self.__parser_project_info(project))
45 return projects_dict_list
47 def get_project(self):
48 try:
49 project = self._jira.project(self._config['JIRA_PROJECT_ID'])
50 except JIRAError as e:
51 self.retries -= 1
52 if (e.status_code // 100) == 4 and self.retries > 0:
53 time.sleep(1)
54 return self.get_project()
55 logger.error(f"=>Exception {e.text}")
56 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
57 return self.__parser_project_info(project)
59 def get_issues(self, sql: str, offset: int = 0):
60 jql = "project = " + self._config['JIRA_PROJECT_ID'] \
61 + ((" AND " + sql) if sql is not None and len(sql) > 0 else "") \
62 + " ORDER BY createdDate DESC"
64 try:
65 issues = self._jira.search_issues(jql, maxResults=1000, startAt=offset, fields=fields)
66 except JIRAError as e:
67 self.retries -= 1
68 if (e.status_code // 100) == 4 and self.retries > 0:
69 time.sleep(1)
70 return self.get_issues(sql, offset)
71 logger.error(f"=>Exception {e.text}")
72 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
74 issue_dict_list = []
75 for issue in issues:
76 issue_dict_list.append(self.__parser_issue_info(issue, include_comments=False))
78 # return {"total": issues.total, "issues": issue_dict_list}
79 return issue_dict_list
81 def get_issue(self, issue_id: str):
82 try:
83 # issue = self._jira.issue(issue_id)
84 issue = self._jira.issue(issue_id, fields=fields)
85 except JIRAError as e:
86 self.retries -= 1
87 if (e.status_code // 100) == 4 and self.retries > 0:
88 time.sleep(1)
89 return self.get_issue(issue_id)
90 logger.error(f"=>Exception {e.text}")
91 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
92 return self.__parser_issue_info(issue)
94 def get_issue_v3(self, issue_id: str):
95 try:
96 url = f"{self._config['JIRA_URL']}/rest/api/3/issue/{issue_id}?fields={fields}"
97 auth = HTTPBasicAuth(self._config['JIRA_USERNAME'], self._config['JIRA_PASSWORD'])
98 issue = requests.get(
99 url,
100 headers={
101 "Accept": "application/json"
102 },
103 auth=auth
104 )
105 except Exception as e:
106 self.retries -= 1
107 if self.retries > 0:
108 time.sleep(1)
109 return self.get_issue_v3(issue_id)
110 logger.error(f"=>Exception {e}")
111 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: get issue error")
112 return self.__parser_issue_info(issue.json())
114 def create_issue(self, issue_dict):
115 issue_dict["project"] = {"id": self._config['JIRA_PROJECT_ID']}
116 try:
117 issue = self._jira.create_issue(fields=issue_dict)
118 return self.__parser_issue_info(issue)
119 except JIRAError as e:
120 self.retries -= 1
121 if (e.status_code // 100) == 4 and self.retries > 0:
122 time.sleep(1)
123 return self.create_issue(issue_dict)
124 logger.error(f"=>Exception {e.text}")
125 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
127 def close_issue(self, issue):
128 try:
129 # jira.transition_issue(issue, '5', assignee={'name': 'pm_user'}, resolution={'id': '3'})
130 self._jira.transition_issue(issue, 'Close')
131 except JIRAError as e:
132 self.retries -= 1
133 if (e.status_code // 100) == 4 and self.retries > 0:
134 time.sleep(1)
135 return self.close_issue(issue)
136 logger.error(f"=>Exception {e.text}")
137 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
139 def assign_issue(self, issue_id, account_id) -> bool:
140 try:
141 return self._jira.assign_issue(issue_id, account_id)
142 except JIRAError as e:
143 self.retries -= 1
144 if (e.status_code // 100) == 4 and self.retries > 0:
145 time.sleep(1)
146 return self.assign_issue(issue_id, account_id)
147 logger.error(f"=>Exception {e.text}")
148 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
150 def add_comment(self, issue_id: str, comment: str):
151 try:
152 comment = self._jira.add_comment(issue_id, comment)
153 except JIRAError as e:
154 self.retries -= 1
155 if (e.status_code // 100) == 4 and self.retries > 0:
156 time.sleep(1)
157 return self.add_comment(issue_id, comment)
158 logger.error(f"=>Exception {e.text}")
159 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
160 return self.__parser_comment_info(comment)
162 def add_comment_v3(self, issue_id: str, comment: str):
163 try:
164 url = f"{self._config['JIRA_URL']}/rest/api/3/issue/{issue_id}/comment"
165 auth = HTTPBasicAuth(self._config['JIRA_USERNAME'], self._config['JIRA_PASSWORD'])
166 comment_response = requests.post(
167 url,
168 headers={
169 "Accept": "application/json"
170 },
171 auth=auth,
172 json={
173 "body": {
174 "type": "doc",
175 "version": 1,
176 "content": [
177 {
178 "type": "paragraph",
179 "content": [
180 {
181 "text": comment,
182 "type": "text"
183 }
184 ]
185 }
186 ]
187 }
188 }
189 )
190 except Exception as e:
191 self.retries -= 1
192 if self.retries > 0:
193 time.sleep(1)
194 return self.add_comment_v3(issue_id, comment)
195 logger.error(f"=>Exception {e}")
196 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: comment error")
197 return self.__parser_comment_info(comment_response.json())
199 def get_comments(self, issueKey):
200 try:
201 comments = self._jira.comments(issueKey)
202 results = []
203 for c in comments:
204 results.append(self.__parser_comment_info(c.raw))
205 return results
206 except JIRAError as e:
207 self.retries -= 1
208 if (e.status_code // 100) == 4 and self.retries > 0:
209 time.sleep(1)
210 return self.get_comments(issueKey)
211 logger.error(f"=>Exception {e.text}")
212 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
214 def get_meta(self):
215 meta = {}
216 meta['issueTypes'] = self.get_issue_types()
217 meta['users'] = self.get_assignable_users()
218 return meta
220 def get_assignable_users(self):
221 try:
222 users = self._jira.search_assignable_users_for_issues(project=self._config['JIRA_PROJECT_ID'], query="*")
223 except JIRAError as e:
224 self.retries -= 1
225 if (e.status_code // 100) == 4 and self.retries > 0:
226 time.sleep(1)
227 return self.get_assignable_users()
228 logger.error(f"=>Exception {e.text}")
229 if e.status_code == 401:
230 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="JIRA: 401 Unauthorized")
231 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
232 users_dict = []
233 for user in users:
234 users_dict.append({
235 'name': user.displayName,
236 'email': user.emailAddress,
237 'id': user.accountId,
238 'avatarUrls': user.raw["avatarUrls"]
239 })
241 return users_dict
243 def get_issue_types(self):
244 try:
245 types = self._jira.project(self._config['JIRA_PROJECT_ID']).issueTypes
246 except JIRAError as e:
247 self.retries -= 1
248 if (e.status_code // 100) == 4 and self.retries > 0:
249 time.sleep(1)
250 return self.get_issue_types()
251 logger.error(f"=>Exception {e.text}")
252 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"JIRA: {e.text}")
253 types_dict = []
254 for type in types:
255 if not type.subtask and not type.name.lower() == "epic":
256 types_dict.append({
257 'id': type.id,
258 'name': type.name,
259 'iconUrl': type.iconUrl,
260 'description': type.description
261 })
262 return types_dict
264 def __parser_comment_info(self, comment):
265 if not isinstance(comment, dict):
266 comment = comment.raw
268 pattern = '%Y-%m-%dT%H:%M:%S.%f%z'
269 creation = datetime.strptime(comment['created'], pattern)
270 # update = datetime.strptime(comment['updated'], pattern)
272 return {
273 'id': comment['id'],
274 'author': comment['author']['accountId'],
275 'message': comment['body'],
276 # 'created': comment['created'],
277 'createdAt': int((creation - creation.utcoffset()).timestamp() * 1000),
278 # 'updated': comment['updated'],
279 # 'updatedAt': int((update - update.utcoffset()).timestamp() * 1000)
280 }
282 @staticmethod
283 def __get_closed_status(status):
284 return status.lower() == "done" or status.lower() == "close" or status.lower() == "closed" or status.lower() == "finish" or status.lower() == "finished"
286 def __parser_issue_info(self, issue, include_comments=True):
287 results_dict = {}
288 if not isinstance(issue, dict):
289 raw_info = issue.raw
290 else:
291 raw_info = issue
293 fields = raw_info['fields']
294 results_dict["id"] = raw_info["id"]
295 results_dict["key"] = raw_info["key"]
296 # results_dict["ticketNumber"] = raw_info["key"]
297 results_dict["title"] = fields["summary"]
298 results_dict["description"] = fields["description"]
299 results_dict["issueType"] = [fields["issuetype"]["id"]]
301 # results_dict["assignee"] = None
302 # results_dict["reporter"] = None
304 if isinstance(fields["assignee"], dict):
305 results_dict["assignees"] = [fields["assignee"]["accountId"]]
306 # if isinstance(fields["reporter"], dict):
307 # results_dict["reporter"] = fields["reporter"]["accountId"]
308 if isinstance(fields["creator"], dict):
309 results_dict["creator"] = fields["creator"]["accountId"]
311 if "comment" in fields:
312 if include_comments:
313 comments_dict = []
314 for comment in fields["comment"]["comments"]:
315 comments_dict.append(self.__parser_comment_info(comment))
317 results_dict['comments'] = comments_dict
318 results_dict['commentsCount'] = fields["comment"]["total"]
320 results_dict["status"] = fields["status"]['name']
321 results_dict["createdAt"] = fields["created"]
322 # results_dict["updated"] = fields["updated"]
323 results_dict["labels"] = fields["labels"]
324 results_dict["closed"] = self.__get_closed_status(fields["status"]['name'])
326 return results_dict
328 @staticmethod
329 def __parser_project_info(project):
330 results_dict = {}
331 raw_info = project.raw
332 results_dict["id"] = raw_info["id"]
333 results_dict["name"] = raw_info["name"]
334 results_dict["avatarUrls"] = raw_info["avatarUrls"]
335 results_dict["description"] = raw_info["description"] if "description" in raw_info else ""
337 return results_dict