Using Python to automate the creation & merging of pull requests in Bitbucket

This automation will use Python along with the Bitbucket (formerly known as Stash) API.

A summary of the Bitbucket services used

For a full list see https://developer.atlassian.com/bitbucket/api/2/reference/resource/

Endpoint	Method	Purpose
/rest/api/2.0/projects/repos/pull-requests/merge?version=0	POST	Merge PR
/rest/api/2.0/projects/repos/pull-requests	POST	Create PR
/rest/api/2.0/projects/repos//pull-requests	GER	Fetch PRs

Additional resources

Since we are sending a JSON payload to create a PR. A sample JSON file is shown below.

Payload: Create PR

{
    "title": "Automated PR",
    "description": "PR generated from Jenkins.",
    "state": "OPEN",
    "open": true,
    "closed": false,
    "fromRef": {
        "id": "refs/heads/feature/<source branch name>",
        "repository": {
            "slug": "<repo_id>",
            "name": null,
            "project": {
                "key": "<project_key>"
            }
        }
    },
    "toRef": {
        "id": "refs/heads/feature/<target branch name>",
        "repository": {
            "slug": "<repo_id>",
            "name": null,
            "project": {
                "key": "<project_key>"
            }
        }
    }
}

Settings file

This file will contain some several static information regarding branches, and payload data file names. This simplifies the process of storing key information necessary for the Python script to read into memory and process the payload data along with branch information.

[BRANCHES]
pr_branch_from="refs/heads/feature/<branch_name>"
pr_branch_to="refs/heads/feature/<branch_name>"

[PAYLOADS]
pr_payload="pr.json"

Python: Reading in a settings file

mysettings is just a key value pair dictionary
mysettings[key]=[value]
e.g. mysettings[pr_branch_from]="refs/heads/feature/<branch_name>”
logger is a Python module to improve the logging output to the console

if(os.path.exists(settings)):
    mysettings={}
    logger.info("Settings file located.")
    with open(settings) as settingsFile:
        for line in settingsFile:
        name,var = line.partition("=")[::2]
        mysettings[name.strip()] = var.strip().replace('\"','')
            
    settingsFile.close()

    if(len(mysettings) == 0):
        logger.error("Failed to read settings file into memory, please verify the file.")
        logger.error("Unable to continue, terminating script.")
        sys.exit(1)

Python: Making web service requests

Creating a Pull request

headers = {'Content-Type': 'application/json'}
create_pr_api = "<bitbucket_url>/rest/api/2.0/projects/<project_key>/repos/<repo_id>/pull-requests"

try:

    response = requests.post(create_pr_api,
                                headers = headers,
                                data = payload_data,
                                auth = HTTPBasicAuth(bitbucket_username,bitbucket_password))

    if(response.status_code == 201):
            print("Pull request created.")
    else:
            print("Pull request failed with {}".format(response.text))
except requests.exceptions.RequestException as e:
    
    print(e)
    sys.exit(1)

Here we define the HTTP header, with the only accepted content-type being ‘application/json’.

We use the Python requests module to send a request to the Create PR webservice. Sending the header, payload data (example shown earlier), as well as the authentication - in this case using the simplistic HTTPBasicAuth method from the requests module to send the username and password of a given Bitbucket user.

Checking the status code from the response, whereby anything other than 201 means success.

Merging a Pull request

First we need to find the correct Pull request ID to merge.

We do a simple scan across all open PRs. As only 1 PR can be open from a specific branch we just do a pattern match for the following:

title
fromRef (source branch)
toRef (target branch)

If the above is found we extract the Pull Request ID from the JSON response returned by the webservice request to get all open PRs.

headers = {'Content-Type': 'application/json'}
pull_requests_api = "<bitbucket_url>/rest/api/2.0/projects/<project_key>/repos/<repo_id>/pull-requests"
try:
    response = requests.get(
        pull_requests_api,
        headers=headers,
        auth=HTTPBasicAuth(bitbucket_username,bitbucket_password))

    if(response.status_code == 200):
        logger.debug("Response code = 200")
        
        if(response.json()):
            
            for val in response.json()['values']:
                if(val['title'] == payload_data['title'] and pr_branch_from == val['fromRef']['id'] 
                    and pr_branch_to == val['toRef']['id']):
                    logger.debug("Id = " + str(val['id']))
                    logger.debug("Title = " + val['title'])
                    logger.debug("fromRef = " + val['fromRef']['id'])
                    logger.debug("toRef="+ val['toRef']['id'])

                    return str(val['id'])
except requests.exceptions.RequestException as e:
    logger.error(e)
    sys.exit(1)

Merging a PR is simple enough and this is demonstrated by the code below. Required is the pull request id taken from the code above.

headers = {'Content-Type': 'application/json'}
pull_requests_merge_api = "<bitbucket_url>/rest/api/2.0/projects/<project_key>/repos/<repo_id>/pull-requests/{}/merge?version=0"

try:
    response = requests.post(
        pull_requests_merge_api.format(pr_id),
        headers=headers,
        auth=HTTPBasicAuth(bitbucket_username,bitbucket_password))
    
    if(response.status_code == 200):
        logger.debug("Response code = 200")
        logger.info("Pull Request Merged.")
        sys.exit(0)
    else:
        processResponseError(response,"Merge Pull Request")
except requests.exceptions.RequestException as e:
        logger.error(e)
        sys.exit(1)

Evaluating response status error codes

There are cases where the request fails to execute as expected, so it is important to evaluate the status code from the response coming back from the web service.

Looking through the status code of the response helps to filter out the different errors.

Typically the JSON array returned within the response, has keys such as ‘errors’ and ‘message’ and it is easy to pull this out via Python.

def processResponseError(response,msg):
    if(response.status_code == 400):
        logger.error("Response code = 400, Bad Request.")
        logger.error("Message = " + response.json()['errors'][0]['message'])
        logger.error("Exception Name = " + response.json()['errors'][0]['exceptionName'])
    elif(response.status_code == 401):
        logger.error("Response code = 401, Unauthorised.")
        logger.error("Message = " + response.json()['errors'][0]['message'])
        logger.error("Exception Name = " + response.json()['errors'][0]['exceptionName'])
    elif(response.status_code == 403):
        logger.error("Response code = 403, Forbidden.")
        logger.error("Message = " + response.json()['errors'][0]['message'])
        logger.error("Exception Name = " + response.json()['errors'][0]['exceptionName'])
    elif(response.status_code == 404):
        logger.error("Response code = 404, Not found.")
        logger.error("Message = " + response.json()['errors'][0]['message'])
        logger.error("Exception Name = " + response.json()['errors'][0]['exceptionName'])
    elif(response.status_code == 405):
        logger.error("Response code = 405, HTTP method not supported.")
        logger.error("Message = " + response.json()['errors'][0]['message'])
        logger.error("Exception Name = " + response.json()['errors'][0]['exceptionName'])
    elif(response.status_code == 409):
        logger.error("Response code = 409, Conflict.")
        logger.error("Message = " + response.json()['errors'][0]['message'])
        logger.error("Exception Name = " + response.json()['errors'][0]['exceptionName'])
    elif(response.status_code == 415):
        logger.error("Response code = 415, Unsupported media type.")
        logger.error("Message = " + response.json()['errors'][0]['message'])
        logger.error("Exception Name = " + response.json()['errors'][0]['exceptionName'])
    
    logger.error("Operation failed = "+ msg)

    # do not fail if the 2 branches are equal
    if(response.json()['errors'][0]['exceptionName'] == "com.atlassian.bitbucket.pull.EmptyPullRequestException" or 
    "is already up-to-date" in response.json()['errors'][0]['message'] ):
        logger.warning("Both branches are up to date.")
    else:
        logger.error("Unable to continue, terminating script.")
        sys.exit(1)

Running the script

The script has a couple of pre-defined operations. These are arguments which are used in conjunction with the flag – operation

create-pr1 - creates a PR
merge-pr1 - merges a PR


    
    usage: merge.py [-h] [--username USERNAME] [--password PASSWORD] [--operation OPERATION] [--settings SETTINGS]
    
    Utils for executing Bitbucket operations via REST API.
    
    optional arguments:
      -h, --help            show this help message and exit
      --username USERNAME   Specify the username to connect to Bitbucket.
      --password PASSWORD   Specify the password to connect to Bitbucket.
      --operation OPERATION
                            Specify the operation to execute.
      --settings SETTINGS   Specify the path to the settings.properties file.

The script plus additional resources are available on Gitlab - https://gitlab.com/spengler84/python-utils/tree/master/bitbucket

Last updated on 23 Jan 2020
Published on 23 Jan 2020