Example of viewing and editing issues through JIRA REST API

jiraczar.py

#!/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'])
                print
            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