Created
February 20, 2015 18:53
-
-
Save mike10004/7857ed1f92e477822c4b to your computer and use it in GitHub Desktop.
Example of viewing and editing issues through JIRA REST API
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
#!/usr/bin/env python | |
# | |
# (c) 2015 Mike Chaberski | |
# | |
# MIT License | |
# | |
# TODO: | |
# * use keyring for username/password | |
# * support OAuth authentication | |
# * support something besides issue assignment | |
import sys | |
import os | |
import os.path | |
import logging | |
import json | |
import requests | |
from requests.auth import HTTPBasicAuth | |
_DEFAULT_CONFIG_FILENAME = 'jiraczar-config.json' | |
_ERR_UNEXPECTED_RESPONSE = 1 | |
_HTTP_METHODS = ('GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS') | |
_LOG_LEVELS = ('DEBUG', 'INFO', 'WARN', 'ERROR') | |
_log = logging.getLogger('jiraczar') | |
def get_default_config_pathname(): | |
cfg_pathname = os.path.join(os.getcwd(), _DEFAULT_CONFIG_FILENAME) | |
return cfg_pathname | |
class ConfigLoader: | |
defaults = { | |
'verify': True, | |
'rest_base_url': 'https://httpbin.org/my-jira/rest/api/2', | |
'json_indent': 2, | |
'json_separators': (',', ': ') | |
} | |
def load(self, source=None): | |
if source is None: | |
cfg_pathname = get_default_config_pathname() | |
return self.load(cfg_pathname) | |
elif isinstance(source, str): | |
_log.debug("reading config from %s", source) | |
with open(source, 'r') as ifile: | |
return self.load(ifile) | |
else: # assume source is open file object | |
config = json.loads(source.read()) | |
for k in self.defaults: | |
if k not in config: config[k] = self.defaults[k] | |
return config | |
class Requester: | |
def __init__(self, config, session): | |
self.config = config | |
self.session = session | |
def build_url(self, endpoint_path): | |
base = self.config['rest_base_url'] | |
url = base + endpoint_path | |
return url | |
def send(self, endpoint_path, params, json_data, method='GET'): | |
url = self.build_url(endpoint_path) | |
r = requests.Request(method, url, json=json_data, params=params) | |
pr = self.session.prepare_request(r) | |
return self.session.send(pr, verify=self.config['verify']) | |
class EndpointActor: | |
def __init__(self, path_template, param_names): | |
self.path_template = path_template | |
self.param_names = tuple(param_names) | |
def get_path(self, variables={}): | |
expanded = self.path_template.format(**variables) | |
_log.debug("expanded '%s' into '%s' using vars %s", self.path_template, expanded, str(variables)) | |
return expanded | |
def set_params_from_args(self, args, allow_multiple=False): | |
if allow_multiple: raise NotImplementedError('multiple parameters of same name not supported') | |
defined = [] | |
for arg in args: | |
pdef = arg.split('=', 1) | |
k = pdef[0] | |
v = pdef[1] if len(pdef) > 1 else '' | |
if k in self.param_names: | |
self.__dict__[k] = v | |
defined.append((k, v)) | |
else: | |
_log.info("unexpected parameter for search query: %s", k) | |
_log.debug("set params %s", str([d[0] for d in defined])) | |
def get_params(self): | |
p = {} | |
for k in self.param_names: p[k] = self.__dict__[k] | |
return p | |
class IssueSearcher(EndpointActor): | |
def __init__(self, startAt=0, maxResults=20, validateQuery=True, fields=None, expand=None): | |
EndpointActor.__init__(self, '/search', ('jql', 'startAt', 'maxResults', 'validateQuery', 'fields', 'expand')) | |
self.startAt = startAt | |
self.maxResults = maxResults | |
self.validateQuery = validateQuery | |
self.fields = fields | |
self.expand = expand | |
class IssueAssigner(EndpointActor): | |
def __init__(self): | |
EndpointActor.__init__(self, '/issue/{issueIdOrKey}/assignee', tuple()) | |
ENDPOINTS = { | |
'search': IssueSearcher, | |
'assign': IssueAssigner | |
} | |
def _configure_logging(args): | |
level = eval('logging.' + args.log_level) | |
logging.basicConfig(level=level) | |
def is_http_success(status): | |
return status / 100 == 2 | |
def main(args): | |
config = ConfigLoader().load(args.config_file) | |
request_data = None | |
if args.data_from is not None: | |
if args.data_from == '-': | |
request_data = json.load(sys.stdin) | |
else: | |
with open(args.data_from, 'r') as ifile: | |
request_data = json.load(ifile) | |
session = requests.Session() | |
session.auth = HTTPBasicAuth(config['user'], config['password']) | |
_log.debug("") | |
actor = ENDPOINTS[args.endpoint]() | |
actor.set_params_from_args(args.params) | |
variables = {} | |
for variable_def in args.variables: | |
k, v = variable_def.split('=', 1) | |
variables[k] = v | |
endpoint_path = actor.get_path(variables) | |
params = actor.get_params() | |
requester = Requester(config, session) | |
_log.debug("sending request to endpoint %s with params %s", endpoint_path, str(params)) | |
response = requester.send(endpoint_path, params, request_data, method=args.method) | |
_log.debug("response status %s", response.status_code) | |
if is_http_success(response.status_code): | |
exit_code = 0 | |
if response.text is not None and len(response.text) > 0: | |
try: | |
response_obj = response.json() | |
json.dump(response_obj, sys.stdout, indent=config['json_indent'], separators=config['json_separators']) | |
except ValueError: | |
_log.info("failed to parse json from response of length %s", len(response.text)) | |
print response.text | |
else: | |
print response.text | |
else: | |
response_data = response.text | |
print response_data | |
exit_code = _ERR_UNEXPECTED_RESPONSE | |
return exit_code | |
if __name__ == '__main__': | |
from argparse import ArgumentParser | |
p = ArgumentParser(description="make jira rest calls") | |
p.add_argument("--config-file", default=None, help="configuration file pathname", metavar="FILE") | |
p.add_argument("endpoint", choices=ENDPOINTS.keys(), help="API endpoint") | |
p.add_argument("params", nargs='*', metavar="NAME=VALUE", | |
help="request parameters in the format key=value (not URL encoded)") | |
p.add_argument("-l", "-L", "--log-level", choices=_LOG_LEVELS, | |
default='INFO', metavar="LEVEL", | |
help="log level; one of " + ", ".join(_LOG_LEVELS)) | |
p.add_argument("--method", choices=_HTTP_METHODS, metavar="METHOD", | |
help="HTTP method", default="GET") | |
p.add_argument("--variables", nargs="+", metavar="KEY=VALUE", default=[], | |
help="set endpoint path variable; X=Y expands '/foo/{X}/bar' into '/foo/Y/bar'") | |
p.add_argument("--data-from", help="file containing json-formatted data for POST or PUT") | |
args = p.parse_args() | |
_configure_logging(args) | |
exit(main(args)) |
What I actually ran in the shell was something like this:
$ ./jiraczarview.py search jql='project = VIDET AND assignee=wrongname' maxResults=500 fields=assignee | python -c '
import sys,json
for issue in json.load(sys.stdin)["issues"]:
print issue["key"]
' | xargs -I{} -n1 ./jiraczar.py assign --method PUT --variables issueIdOrKey={} --data-from assignee-correctname.json
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Program written to perform multiple edits of issues in JIRA. Probably should have searched for a Python JIRA client library first.