297 lines
8.3 KiB
Python
297 lines
8.3 KiB
Python
import json
|
|
import logging
|
|
from enum import Enum
|
|
|
|
import requests
|
|
from requests.auth import HTTPBasicAuth
|
|
|
|
from core.Expando import Expando
|
|
|
|
JIRA_ROOT = "https://altares.atlassian.net/rest/api/3"
|
|
DEFAULT_HEADERS = {"Accept": "application/json"}
|
|
DEFAULT_SEARCH_FIELDS = "summary,status,assignee"
|
|
logger = logging.getLogger("Jira")
|
|
|
|
|
|
class NotFound(Exception):
|
|
pass
|
|
|
|
|
|
class JiraRequestTypes(Enum):
|
|
Search = "search"
|
|
Issue = "issue"
|
|
Comments = "comments"
|
|
Versions = "versions"
|
|
|
|
|
|
class Jira:
|
|
"""Manage default operation to JIRA"""
|
|
|
|
def __init__(self, user_name: str, api_token: str, fields=DEFAULT_SEARCH_FIELDS):
|
|
"""
|
|
Prepare a connection to JIRA
|
|
The initialisation do not to anything,
|
|
It only stores the user_name and the api_token
|
|
Note that user_name and api_token is the recommended way to connect,
|
|
therefore, the only supported here
|
|
:param user_name:
|
|
:param api_token:
|
|
"""
|
|
self.user_name = user_name
|
|
self.api_token = api_token
|
|
self.auth = HTTPBasicAuth(self.user_name, self.api_token)
|
|
self.fields = fields
|
|
|
|
def test(self):
|
|
logger.debug(f"test with no parameters")
|
|
|
|
url = f"{JIRA_ROOT}/myself"
|
|
logger.debug(f" url: {url}")
|
|
|
|
response = requests.request(
|
|
"GET",
|
|
url,
|
|
headers=DEFAULT_HEADERS,
|
|
auth=self.auth
|
|
)
|
|
logger.debug(f" response: {response}")
|
|
logger.debug(f" response.text: {response.text}")
|
|
|
|
return response
|
|
|
|
def issue(self, issue_id: str) -> list[Expando]:
|
|
"""
|
|
Retrieve an issue
|
|
:param issue_id:
|
|
:return:
|
|
"""
|
|
logger.debug(f"comments with {issue_id=}")
|
|
|
|
url = f"{JIRA_ROOT}/issue/{issue_id}"
|
|
logger.debug(f" url: {url}")
|
|
|
|
response = requests.request(
|
|
"GET",
|
|
url,
|
|
headers=DEFAULT_HEADERS,
|
|
auth=self.auth
|
|
)
|
|
logger.debug(f" response: {response}")
|
|
logger.debug(f" response.text: {response.text}")
|
|
|
|
return [Expando(json.loads(response.text))]
|
|
|
|
def fields(self) -> list[Expando]:
|
|
"""
|
|
Retrieve the list of all fields for an issue
|
|
:return:
|
|
"""
|
|
url = f"{JIRA_ROOT}/field"
|
|
|
|
response = requests.request(
|
|
"GET",
|
|
url,
|
|
headers=DEFAULT_HEADERS,
|
|
auth=self.auth
|
|
)
|
|
|
|
as_dict = json.loads(response.text)
|
|
return [Expando(field) for field in as_dict]
|
|
|
|
def search(self, jql: str, fields=None) -> list[Expando]:
|
|
"""
|
|
Executes a JQL and returns the list of issues
|
|
:param jql:
|
|
:param fields: list of fields to retrieve
|
|
:return:
|
|
"""
|
|
logger.debug(f"search with {jql=}, {fields=}")
|
|
|
|
if not jql:
|
|
raise ValueError("Jql cannot be empty.")
|
|
|
|
if not fields:
|
|
fields = self.fields
|
|
|
|
url = f"{JIRA_ROOT}/search"
|
|
logger.debug(f" url: {url}")
|
|
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
payload = {
|
|
"fields": [f.strip() for f in fields.split(",")],
|
|
"fieldsByKeys": False,
|
|
"jql": jql,
|
|
"maxResults": 500, # Does not seem to be used. It's always 100 !
|
|
"startAt": 0
|
|
}
|
|
logger.debug(f" payload: {payload}")
|
|
|
|
result = []
|
|
while True:
|
|
logger.debug(f" Request startAt '{payload['startAt']}'")
|
|
response = requests.request("POST",
|
|
url,
|
|
data=json.dumps(payload),
|
|
headers=headers,
|
|
auth=self.auth)
|
|
logger.debug(f" response: {response}")
|
|
logger.debug(f" response.text: {response.text}")
|
|
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(self._format_error(response))
|
|
|
|
as_dict = json.loads(response.text)
|
|
result += as_dict["issues"]
|
|
|
|
if as_dict["startAt"] + as_dict["maxResults"] >= as_dict["total"]:
|
|
logger.debug(f" response: {response}")
|
|
# We retrieve more than the total nuber of items
|
|
break
|
|
|
|
payload["startAt"] += as_dict["maxResults"]
|
|
|
|
return [Expando(issue) for issue in result]
|
|
|
|
def comments(self, issue_id: str) -> list[Expando]:
|
|
"""
|
|
Retrieve the list of comments for an issue
|
|
:param issue_id:
|
|
:return:
|
|
"""
|
|
logger.debug(f"comments with {issue_id=}")
|
|
|
|
url = f"{JIRA_ROOT}/issue/{issue_id}/comment"
|
|
logger.debug(f" url: {url}")
|
|
|
|
response = requests.request("GET",
|
|
url,
|
|
headers=DEFAULT_HEADERS,
|
|
auth=self.auth)
|
|
logger.debug(f" response: {response}")
|
|
logger.debug(f" response.text: {response.text}")
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(self._format_error(response))
|
|
|
|
as_dict = json.loads(response.text)
|
|
result = as_dict["comments"]
|
|
return [Expando(issue) for issue in result]
|
|
|
|
def versions(self, project_key):
|
|
"""
|
|
Given a project name and a version name
|
|
returns fixVersion number in JIRA
|
|
:param project_key:
|
|
:return:
|
|
"""
|
|
logger.debug(f"versions with {project_key=}")
|
|
|
|
url = f"{JIRA_ROOT}/project/{project_key}/versions"
|
|
logger.debug(f" url: {url}")
|
|
|
|
response = requests.request(
|
|
"GET",
|
|
url,
|
|
headers=DEFAULT_HEADERS,
|
|
auth=self.auth
|
|
)
|
|
|
|
logger.debug(f" response: {response}")
|
|
logger.debug(f" response.text: {response.text}")
|
|
|
|
if response.status_code != 200:
|
|
raise NotFound()
|
|
|
|
as_list = json.loads(response.text)
|
|
return [Expando(version) for version in as_list]
|
|
|
|
def extract(self, jql, mappings, updates=None) -> list[dict]:
|
|
"""
|
|
Executes a jql and returns list of dict
|
|
The <code>issue</code> object, returned by the <ref>jql</ref> methods
|
|
contains all the fields for Jira. They are not all necessary
|
|
This method selects the required fields
|
|
:param jql:
|
|
:param mappings:
|
|
:param updates: List of updates (lambda on issue) to perform
|
|
:return:
|
|
"""
|
|
logger.debug(f"Processing extract using mapping {mappings}")
|
|
|
|
def _get_field(mapping):
|
|
"""Returns the meaningful jira field, for the mapping description path"""
|
|
fields = mapping.split(".")
|
|
return fields[1] if len(fields) > 1 and fields[0] == "fields" else fields[0]
|
|
|
|
# retrieve the list of requested fields from what was asked in the mapping
|
|
jira_fields = [_get_field(mapping) for mapping in mappings]
|
|
as_string = ", ".join(jira_fields)
|
|
issues = self.issues(jql, as_string)
|
|
|
|
for issue in issues:
|
|
# apply updates if needed
|
|
if updates:
|
|
for update in updates:
|
|
update(issue)
|
|
|
|
row = {cvs_col: issue.get(jira_path) for jira_path, cvs_col in mappings.items() if cvs_col is not None}
|
|
yield row
|
|
|
|
def get_version(self, project_key, version_name):
|
|
"""
|
|
Given a project name and a version name
|
|
returns fixVersion number in JIRA
|
|
:param project_key:
|
|
:param version_name:
|
|
:return:
|
|
"""
|
|
|
|
for version in self.versions(project_key):
|
|
if version.name == version_name:
|
|
return version
|
|
|
|
raise NotFound()
|
|
|
|
def get_all_fields(self):
|
|
"""
|
|
Helper function that returns the list of all field that can be requested in an issue
|
|
:return:
|
|
"""
|
|
url = f"{JIRA_ROOT}/field"
|
|
response = requests.request(
|
|
"GET",
|
|
url,
|
|
headers=DEFAULT_HEADERS,
|
|
auth=self.auth
|
|
)
|
|
|
|
as_dict = json.loads(response.text)
|
|
return [Expando(issue) for issue in as_dict]
|
|
|
|
@staticmethod
|
|
def update_customer_refs(issue: Expando, bug_only=True, link_name=None):
|
|
issue["ticket_customer_refs"] = []
|
|
if bug_only and issue.fields.issuetype.name != "Bug":
|
|
return
|
|
|
|
for issue_link in issue.fields.issuelinks: # [i_link for i_link in issue.fields.issuelinks if i_link["type"]["name"] == "Relates"]:
|
|
if link_name and issue_link["type"]["name"] not in link_name:
|
|
continue
|
|
|
|
direction = "inwardIssue" if "inwardIssue" in issue_link else "outwardIssue"
|
|
related_issue_key = issue_link[direction]["key"]
|
|
if related_issue_key.startswith("ITSUP"):
|
|
issue.ticket_customer_refs.append(related_issue_key)
|
|
continue
|
|
|
|
@staticmethod
|
|
def _format_error(response):
|
|
if "errorMessages" in response.text:
|
|
error_messages = json.loads(response.text)["errorMessages"]
|
|
else:
|
|
error_messages = response.text
|
|
return f"Error {response.status_code} : {response.reason} : {error_messages}"
|