Kick off a Jenkins Job programmatically.

StartJenkinsJobs.py

#!/usr/bin/env python
#from https://gist.github.com/tstone2077/6106950
#__doc__ =
"""
   OriginalAuthor: Thurston Stone
   Description: Kick off a Jenkins Job programmatically.
   Usage: run 'python StartJenkinsJob.py --help' for details.

   Known Issues:
    -c|--cause does not seem to work yet
    This script anticipates the next build number.  If multiple builds are
            triggered simultaneously, it is possible for them to check the
            wrong build for its completion status.  There is no way that I
            have found to link a build queue request to the build it kicks
            off.
"""
__author__ = "Thurston Stone"
__versioninfo__ = (0, 1, 0)
__version__ = '.'.join(map(str, __versioninfo__))

cmdDesc = """
    Kick off a Jenkins Job programatically.
    """

import sys
import os
import argparse
import logging
#-------------------
#Compatable with python 2.6 to 3
import ast
import base64
import socket
import time
try:
    from urllib2 import urlopen, HTTPError, Request
    from urllib import urlencode
    from urllib import quote
except ImportError:
    from urllib.request import urlopen, Request
    from urllib.error import HTTPError
    from urllib.parse import urlencode
    from urllib.parse import quote

INVALID_USAGE_RETURN_CODE = 2
UNCAUGHT_EXCEPTION = 3

SCRIPT_FILE = os.path.basename(os.path.abspath(__file__))
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))


#set the defaults
BUILD_TIMEOUT = 60  # minutes
POLL_INTERVAL = 5   # seconds


class UsageError(Exception):
    """ Exception class when the file is used incorrectly on the command line
    """
    def __init__(self, parser, error):
        self.parser = parser
        self.error = error

    def __str__(self):
        return "%s\nERROR: %s" % (self.parser.format_usage(), self.error)


def validate_args(argv):
    """
    Function: validate_args():
    will validate the command line arguments and options passed.
    it returns opts,args
    """

    parser = argparse.ArgumentParser(description=cmdDesc)
    #general options

    parser.add_argument("-r",
                        "--root-url",
                        required=True,
                        help="Root url of the jenkins server (e.g. "
                             "http://jenkins.domain.com:1234)")

    parser.add_argument("-j",
                        "--job-name",
                        required=True,
                        help="Name of the jenkins job to run (e.g. "
                             '"Test Job")')

    parser.add_argument("-a",
                        "--authentication",
                        required=True,
                        help='User name and Token string used for'
                             'authentication in the format username=token.  '
                             'The token can be found under "Jenkins->'
                             '<username>->Configure->API Token->Show API '
                             'Token..."')

    parser.add_argument("-p",
                        "--parameters",
                        default=None,
                        help='Optional parameters to be sent to the job in '
                             'the format "param1=value1,param2=value2"')

    parser.add_argument("-c",
                        "--cause",
                        default=None,
                        help='The cause of the build.')

    parser.add_argument("-w",
                        "--wait",
                        action="store_true",
                        default=False,
                        help='Wait for the build to complete and return the '
                             'build status.')

    parser.add_argument("-t",
                        "--timeout",
                        default=None,
                        type=int,
                        help='Legth of time to consider the build timed out '
                             'and abandon wait. (assumes -w|--wait)')

    parser.add_argument("-i",
                        "--poll-interval",
                        default=None,
                        type=int,
                        help='Interval in seconds to poll the jenkins server '
                             'to check if the build is done (assumes '
                             '-w|--wait)')

    parser.add_argument("-v",
                        "--verbose",
                        default="WARNING",
                        const="INFO",
                        nargs="?",
                        help="Level of verbose output to Display to stdout"
                             "(DEBUG, INFO, WARNING, ERROR, CRITICAL, FATAL)")

    args = parser.parse_args(argv[1:])

    if args.poll_interval is not None or args.timeout is not None:
        args.wait = True
    if args.poll_interval is None:
        args.poll_interval = POLL_INTERVAL

    error = None
    #place error handling here
    if error is not None:
        raise UsageError(parser, error)
    return args


def wait_for_job_result(build_data, poll_interval, timeout=None):
    """
    returns the data structure returned by jenkins for the specific build
    (<job_url>/<buildnum>/api/python).
    build_data must contain:
        "url" which contains the url to the job that we are waiting for
    If timeout is used, it must be an int representing minutes.
    """
    #wait until the build is done
    startTime = time.time()
    elapsed_time = 0
    result = None
    build_data['building'] = True
    if timeout is not None:
        timeout *= 60  # turn the int into minutes
    not_timedout = True
    while build_data['building'] and not_timedout:
        now = time.time()
        elapsed_time = now-startTime
        try:
            logging.debug("Checking build status...")
            conn = urlopen(build_data["url"]+"/api/python")
            build_data = ast.literal_eval(conn.read().decode('UTF-8'))
        except HTTPError:
            pass
        if timeout is not None:
            not_timedout = elapsed_time < timeout
        if build_data['building'] and not_timedout:
            #wait 10 seconds for the next query
            time.sleep(poll_interval)
    if build_data["result"] is None:
        build_data["result"] = "TIMEDOUT"
    return build_data


def start_jenkins_job(root_url,
                      job_name,
                      user_auth,  # tuple or list containing (username,token)
                      parameters=None,
                      cause=None,
                      wait=False,
                      build_timeout=BUILD_TIMEOUT,
                      poll_interval=POLL_INTERVAL):
    user_name, user_token = user_auth
    #fill in as much data as we have in the format of the build_data outputted
    #by the jenkins python apis
    build_data = {"building": False}
    job_url = "%s/job/%s" % (root_url, quote(job_name))
    #determine if job parameters are needed
    dataStr = urlopen(job_url+"/api/python").read().decode('UTF-8')
    job_data = ast.literal_eval(dataStr)
    params_required = False
    for action in job_data["actions"]:
        if "parameterDefinitions" in action:
            params_required = True
    if params_required and parameters is None:
        raise AttributeError("This job requires parameters.")
    elif not params_required and parameters is not None:
        raise AttributeError("This job cannot use parameters.")

    if parameters is None:
        build_url = job_url + "/build?"
    else:
        build_url = job_url + "/buildWithParameters?"
        parameters = urlencode(parameters).encode('ascii')
    build_url += "&token=" + user_token
    if cause is not None:
        logging.debug("Adding cause to url: %s" % cause)
        build_url += "&%s" % urlencode({"cause": cause})
    #determine what the next build is.  This is needed because if there is a
    #quiet time... after the build is triggered, we wait for 10 seconds to
    #start it.
    last_build_url = job_url + "/lastBuild/api/python"
    logging.debug("last_build_url=%s" % last_build_url)

    dataStr = urlopen(last_build_url).read().decode('UTF-8')
    build_data["number"] = ast.literal_eval(dataStr)["number"] + 1
    build_data["url"] = "%s/%s/" % (job_url, build_data['number'])
    logging.debug("next build url=%s" % build_data['url'])

    #start the build
    request = Request(build_url)
    auth_str = ('%s:%s' % (user_name, user_token)).encode('ascii')
    base64string = base64.encodestring(auth_str).decode().replace('\n', '')
    request.add_header("Authorization", "Basic %s" % base64string)
    logging.debug("opening url to start build : %s" % build_url)
    urlopen(request, parameters)
    build_data["result"] = "REQUEST_SENT"
    logging.info("started build #%d : %s" % (build_data["number"],
                                             build_data["url"]))

    if wait:
        wait_debug_str = "Waiting for the build to finish"
        if build_timeout is not None:
            wait_debug_str += " (up to %d minutes)" % build_timeout
        wait_debug_str += "..."
        logging.info(wait_debug_str)
        build_data = wait_for_job_result(build_data, poll_interval,
                                         build_timeout)
    logging.debug(build_data)
    return build_data


def main(argv):
    """
    The main function.  This function will run if the command line is called as
    opposed to this file being imported.
    """
    args = validate_args(argv)
    level = getattr(logging, args.verbose.upper())
    logging.basicConfig(level=level,
                        format='%(module)s(%(lineno)d)|[%(asctime)s]|' +
                               '[%(levelname)s]| %(message)s')
    logging.debug("Python %s" % sys.version)
    parameters = None
    if args.parameters is not None and len(args.parameters) != 0:
        parameters = {}
        for items in args.parameters.split(","):
            key, val = items.split("=")
            parameters[key] = val
    logging.debug("Parameters = %s" % parameters)
    build_data = start_jenkins_job(root_url=args.root_url,
                                   job_name=args.job_name,
                                   user_auth=args.authentication.split("="),
                                   parameters=parameters,
                                   cause=args.cause,
                                   wait=args.wait,
                                   build_timeout=args.timeout,
                                   poll_interval=args.poll_interval
                                   )
    print('Click here to see the status of the build: <a href="%s/console">'
          '%s/console</a>' % (build_data["url"],build_data["url"]))
    print(build_data["result"])
    if build_data["result"] != "SUCCESS" and \
       build_data["result"] != "REQUEST_SENT":
        return 1


# if this program is run on the command line, run the main function
if __name__ == '__main__':
    showTraceback = False
    if '--show-traceback' in sys.argv:
        showTraceback = True
        del sys.argv[sys.argv.index('--show-traceback')]
    try:
        retval = main(sys.argv)
        sys.exit(retval)
    except UsageError as e:
        if showTraceback:
            raise
        print(e)
        returnCode = INVALID_USAGE_RETURN_CODE
    except Exception as e:
        if showTraceback:
            raise
        print("Uncaught Exception: %s: %s" % (type(e), e))
        returnCode = UNCAUGHT_EXCEPTION
    sys.exit(returnCode)