-
-
Save qxf2/53958d4dd797988c1ad4f21875dd66cb to your computer and use it in GitHub Desktop.
Sample code to support a few Qxf2 blog posts about analyzing JIRA data using Python. In this gist, we show you the ConnectJira module which contains wrappers for many Jira API calls.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
Class to connect to Jira server and perform some commonly actions (e.g.: execute a JQL) | |
DISCLAIMER: This is sample code written by Qxf2 Services to support a blog post about analyzing JIRA data using Python | |
This code does not represent Qxf2 Services's coding standards. | |
""" | |
from jira import JIRA | |
import arrow | |
import datetime | |
import pandas as pd | |
import matplotlib.pyplot as plt | |
from jira import JIRAError | |
from requests.exceptions import ConnectionError | |
class ConnectJira(): | |
""" | |
Python Jira Library to automate jira dashboard | |
""" | |
def __init__(self, server, username, password, project_name, agile_rest_path=None): | |
"Initializer" | |
self.error = None | |
try: | |
if agile_rest_path is None: | |
jira_options = {'server': server,'agile_rest_path':'greenhopper'} | |
else: | |
jira_options = {'server': server,'agile_rest_path':agile_rest_path} | |
self.jira_obj = JIRA(jira_options, basic_auth=(username, password),validate=False) | |
self.project = project_name | |
except JIRAError as error: | |
self.error = error.status_code | |
except ConnectionError as error: | |
self.error = error | |
# this method stays in ConnectJira | |
def execute_query(self, query=None): | |
"Execute a given query" | |
ticket_list, error = None, None | |
try: | |
ticket_list = self.jira_obj.search_issues( | |
query, maxResults=False) | |
while ticket_list.total > len(ticket_list): | |
ticket_list_extended = self.jira_obj.search_issues( | |
query, startAt=len(ticket_list), maxResults=ticket_list.total - len(ticket_list)) | |
ticket_list += ticket_list_extended | |
except JIRAError as error: | |
error = "error_msg:%s\n error_code:%s" %(error.text, error.status_code) | |
finally: | |
return {'ticket_list':ticket_list,'error':error} | |
# this method stays in ConnectJira | |
def get_ticket(self, ticket, expand_args=None): | |
"Fetch a given ticket" | |
# For now, we still use the JIRA API. | |
# But we just need to change this one method when we switch to using a DB | |
return self.jira_obj.issue(ticket, expand=expand_args) | |
# this method stays in ConnectJira | |
def get_ticket_in_json(self, ticket): | |
"Fetch a given ticket in json format" | |
ticket_expanded, error = None, None | |
jql = "'key'='%s'" % (ticket) | |
try: | |
ticket_expanded = self.jira_obj.search_issues(jql, expand='changelog',fields='comment,created,status,reporter', json_result=True) | |
except JIRAError as error: | |
error = "error_msg:%s\n error_code:%s" %(error.text, error.status_code) | |
return {'ticket':ticket_expanded, 'error':error} | |
# this method stays in ConnectJira | |
def get_statuses_from_jira(self): | |
" get statuses from jira work flow" | |
statuses,error = [],None | |
try: | |
all_status = self.jira_obj.statuses() | |
for status in all_status: | |
statuses.append(status.name.lower()) | |
except Exception as error: | |
error = error | |
return {'statuses':statuses, 'error':error} | |
# this method stays in ConnectJira | |
def get_jira_project_boards(self,jira_project): | |
"Return boards for the given project" | |
jira_project_boards,error = [],None | |
try: | |
jira_boards = self.jira_obj.boards() | |
#Get jira_boards for the given project | |
for board in jira_boards: | |
if jira_project in (board.location.name or board.location.key) : | |
jira_project_boards.append({'id':board.id,'name':board.name}) | |
except Exception as error: | |
error = error | |
return {'jira_project_boards':jira_project_boards,'error':error} | |
# this method stays in ConnectJira | |
def get_jira_project_sprints(self,jira_project_boards): | |
"Return sprints for the given boards related to project" | |
jira_project_sprints,error = [] ,None | |
try: | |
board_ids = [board['id'] for board in jira_project_boards] | |
for id in board_ids: | |
jira_project_sprints.append(self.jira_obj.sprints(id)) | |
except Exception as error: | |
error = error | |
return {'jira_project_sprints':jira_project_sprints,'error':error} | |
# this method stays in ConnectJira | |
def get_sprint_ticket_list(self,sprint_id): | |
"Returns ticket for the given sprint id" | |
sprint_ticket_list,error = None,None | |
jql = "sprint = %s"%sprint_id | |
result = self.execute_query(jql) | |
sprint_ticket_list = result['ticket_list'] | |
error = result['error'] | |
return {'sprint_ticket_list':sprint_ticket_list,'error':error} | |
# this method stays in ConnectJira | |
def get_sprint_details(self,sprint_id): | |
"Returns sprint details for the given sprint id" | |
sprint_details,error = None,None | |
try: | |
sprint_details = self.jira_obj.sprint(sprint_id) | |
except Exception as error: | |
error = error | |
return {'sprint_details':sprint_details, 'error':error} | |
# all of the following methods will move to utils | |
def order_list_of_dicts(list_of_dicts,order_by_key,order): | |
if order == 'desc': | |
ordered_list_of_dicts = sorted(list_of_dicts,key = lambda k:k[order_by_key],reverse=True) | |
else: | |
ordered_list_of_dicts = sorted(list_of_dicts,key = lambda k:k[order_by_key]) | |
return ordered_list_of_dicts | |
def get_data_frame(columns): | |
"Create a data frame with only the column headers" | |
data_frame = pd.DataFrame(columns=columns) | |
return data_frame | |
def plot_graph(x_axis, y_axis): | |
"Plot Graph using the X and Y axis passed" | |
plt.plot(x_axis, y_axis) | |
plt.show() | |
def write(msg): | |
"Generic write method" | |
# We'll move this into the logging class when we develop that | |
print msg | |
def calculate_time_delta(start, end): | |
"Generic method to calculate time delta" | |
# Assumes time format to be like JIRA action items | |
# This should ideally not be part of the ConnectJIRA class | |
delta = None | |
if start and end is not None: | |
delta = end - start | |
return delta | |
if __name__ == '__main__': | |
import credentials as secret | |
jira_obj_test = ConnectJira(secret.JIRA_URL, secret.USERNAME, | |
secret.PASSWORD, secret.PROJECT) | |
if jira_obj_test.error is None: | |
date_1 = '01/01/2018' | |
date_2 = '01/10/2018' | |
date_1_obj = datetime.datetime.strptime(date_1, '%m/%d/%Y') | |
date_2_obj = datetime.datetime.strptime(date_2, '%m/%d/%Y') | |
start = date_1_obj.strftime('%Y/%m/%d') | |
end = date_2_obj.strftime('%Y/%m/%d') | |
print jira_obj_test.get_ticket_in_json('EN-test') | |
else: | |
if jira_obj_test.error == 401: | |
print "Unauthorized error: 401, Jira Login Failed - Please check Username and Password" | |
else: | |
print "Please check if Jira server is correct, error detailss: %s"%jira_obj_test.error | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Damn python 2...