Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
Looks at all pull requests across all repos in a Bitbucket (http://bitbucket.org) team and is able to decline if there are not enough reviewers and/or merge if all/required participants have approved
#!/usr/bin/env python
# -*- coding: utf-8
import urllib2
import base64
import json
import re
import sys
import os
import argparse
bb_base_url1="https://bitbucket.org/api/1.0"
bb_base_url2="https://bitbucket.org/api/2.0"
parser = argparse.ArgumentParser(description='Summarize and, optionally, auto-decline/merge pull requests')
parser.add_argument('--username', dest='username', help='your bitbucket username (required if BB_USERNAME environment variable is not set)', required=not os.environ.get('BB_USERNAME'), default=os.environ.get('BB_USERNAME'))
parser.add_argument('--password', dest='password', help='your bitbucket password (required if BB_PASSWORD environment variable is not set)', required=not os.environ.get('BB_PASSWORD'), default=os.environ.get('BB_PASSWORD'))
parser.add_argument('--owner', dest='owner', help='the owner of the bitbucket team', required=True)
parser.add_argument('--auto-merge', dest='auto_merge', help='auto-merge pull requests that meet approval requirements', action='store_true', default=False)
parser.add_argument('--auto-decline', dest='auto_decline', help='auto-decline pull requests that do not meet minimum requirements', action='store_true', default=False)
parser.add_argument('--debug', dest='debug', help='turn on debug logging', action='store_true', default=False)
parser.add_argument('--min-reviewers', dest='minimum_reviewers', metavar='NUM', type=int, help='the minimum number of reviewers (not including participants)', default=2)
parser.add_argument('--required-participants', metavar='USERNAME1,USERNAME2,...', dest='required_participants', help='required participants (defaults to \'username\'), separate multiple usernames with commas')
parser.add_argument('--ignore-reviewers', metavar='USERNAME1,USERNAME2,...', dest='ignore_reviewers', help='reviewers ignored when counting the minimum (defaults to \'username\'), separate multiple with with commas')
args = parser.parse_args()
required_participants = args.required_participants.split(',') if args.required_participants else [ args.username ]
minimum_reviewer_ignore = args.ignore_reviewers.split(',') if args.ignore_reviewers else [ args.username ]
def bitbucket_get(url):
request = urllib2.Request(url)
request.add_header("Authorization", "Basic %s" % base64.encodestring("%s:%s" % ( args.username, args.password )).replace('\n', ''));
result = urllib2.urlopen(request)
response = json.load(result.fp)
result.close()
return response
def bitbucket_post(url, payload):
request = urllib2.Request(url, json.dumps(payload))
request.add_header("Authorization", "Basic %s" % base64.encodestring("%s:%s" % ( args.username, args.password )).replace('\n', ''));
request.add_header("Content-Type", "application/json")
response = { }
result = urllib2.urlopen(request)
response = json.load(result.fp)
result.close()
return response
def main():
repos_page = "%s/repositories/%s" % (bb_base_url2, args.owner)
while repos_page:
# List the repos
repos = bitbucket_get(repos_page)
for repo in repos['values']:
prs_page = repo['links']['pullrequests']['href']
while prs_page:
# List the pull requests
prs = bitbucket_get(prs_page)
if (len(prs['values']) > 0):
print repo['name']
for prs_value in prs['values']:
if prs_value['state'] != 'OPEN':
continue
# Get the pull request
pr = bitbucket_get(prs_value['links']['self']['href'])
print " #%s %s (%s -> %s) by %s" % (prs_value['id'], pr['title'], pr['source']['branch']['name'], pr['destination']['branch']['name'], pr['author']['display_name'])
all_reviewers_accepted = len(pr['participants']) > 0
required_participant_accepted = 0
reviewer_count = 0
# Look at each reviewer/participant
for reviewer in pr['participants']:
if reviewer['role'] == 'REVIEWER':
print u' %s %s' % ((u'' if reviewer['approved'] else u' '), reviewer['user']['display_name'])
all_reviewers_accepted = False if not reviewer['approved'] else all_reviewers_accepted
if reviewer['user']['username'] not in minimum_reviewer_ignore:
reviewer_count += 1
elif reviewer['role'] == 'PARTICIPANT':
print u' %s (%s)' % ((u'' if reviewer['approved'] else u' '), reviewer['user']['display_name'])
else:
print u' %s %s (%s)' % ((u'' if reviewer['approved'] else u' '), reviewer['user']['display_name'], reviewer['role'])
if reviewer['user']['username'] in required_participants and reviewer['approved']:
required_participant_accepted += 1
if args.debug:
print "reviewer_count " + str(reviewer_count)
print "minimum_reviewer_count " + str(args.minimum_reviewers)
print "all_reviewers_accepted " + str(all_reviewers_accepted)
print "required_participant_accepted " + str(required_participant_accepted)
print "len(required_participants) " + str(len(required_participants))
if reviewer_count < args.minimum_reviewers:
# Decline the PR
if not args.auto_decline:
print " PR #%s DOES NOT HAVE AT LEAST %d REVIEWERS - %s" % (prs_value['id'], args.minimum_reviewers, prs_value['links']['decline']['href'])
else:
message = "** AUTO-DECLINE ** - Pull requests must have at least %d reviewers" % (args.minimum_reviewers)
v1_comment_url = "%s/repositories/%s/pullrequests/%s/comments" % (bb_base_url1, repo['full_name'], prs_value['id'])
response = bitbucket_post(v1_comment_url, { "content": message })
response = bitbucket_post(prs_value['links']['decline']['href'], { "message": message })
print " " + message
elif all_reviewers_accepted and required_participant_accepted >= len(required_participants):
# Merge the PR
if not args.auto_merge:
print " PR #%s IS READY TO MERGE - %s" % (prs_value['id'], prs_value['links']['approve']['href'])
else:
message = "** AUTO-MERGE ** - All reviewers approved, including required participants"
v1_comment_url = "%s/repositories/%s/pullrequests/%s/comments" % (bb_base_url1, repo['full_name'], prs_value['id'])
response = bitbucket_post(v1_comment_url, { "content": message })
response = bitbucket_post(prs_value['links']['merge']['href'], None)
print " " + message
prs_page = prs['next'] if 'next' in prs else None
repos_page = repos['next'] if 'next' in repos else None
try:
if __name__ == "__main__":
main()
except KeyboardInterrupt:
sys.exit(-1)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment