Generate a list of open PRs on mdn/browser-compat-data, from smallest to largest

bcd_prs.py

#!/usr/bin/env python
'''Create list of PRs, smallest to largest.'''
from __future__ import print_function, unicode_literals

import pprint
import re
import sys
from argparse import ArgumentParser
from datetime import timedelta

import requests
import requests_cache


class GithubRequester(object):
    '''Make optionally authenticated requests to GitHub.'''

    # Parse the Link: header from a GitHub API response
    re_link_spec = re.compile(r"""(?x)   # Be verbose
        <(?P<url>[^>]*)>;       # An URL wrapped in <>
        \srel="(?P<rel>[^"]*)"  # rel="type"
    """)

    def __init__(self, org, repo, client_id=None, client_secret=None,
                 verbose=False):
        '''
        Initialize GithubRequester.

        Keyword Arguments:
        - org - The GitHub user or organization
        - repo - The GitHub repository
        - client_id - An optional OAuth client ID
        - client_secret - An optional OAuth client secret
        - verbose - Report requests, etc.
        '''
        self.org = org
        self.repo = repo
        self.client_id = client_id
        self.client_secret = client_secret
        self.verbose = verbose

    def get_params(self, **kwargs):
        '''Get parameters list for API calls'''
        params = {}
        if self.client_id and self.client_secret:
            params['client_id'] = self.client_id
            params['client_secret'] = self.client_secret
        for key, value in kwargs.items():
            if value is not None:
                params['key'] = value
        return params

    def request_detail(self, url, params=None):
        '''Request a detailed resource.'''
        resp = requests.get(url, params=params)
        if self.verbose:
            print(resp.url, file=sys.stderr)

        # Abort if the return is an error
        out = resp.json()
        if 'message' in out:
            pprint.pprint(out, stream=sys.stderr)
            raise Exception(resp.text)

        return out

    def request_all(self, url, params):
        '''Request and combine a paginated response.'''
        next_url = None
        first = True
        is_last = False
        resp = requests.get(url, params=params)
        results = []
        while not is_last:
            # Request the next_url if this is not the first request
            if first:
                first = False
            else:
                resp = requests.get(next_url, params=params)
            if self.verbose:
                print(resp.url, file=sys.stderr)

            # Abort if the return is an error
            out = resp.json()
            if 'message' in out:
                pprint.pprint(out, stream=sys.stderr)
                raise Exception(resp.text)

            # Collect the results
            results.extend(out)

            # Process the links to get the next URL
            link_header = resp.headers.get('Link', '')
            links = {rel: url for url, rel
                     in self.re_link_spec.findall(link_header)}
            next_url = links.get('next')
            is_last = next_url is None

        return results

    def get_prs(self, state=None, head=None, base=None, sort=None,
                direction=None):
        '''
        Request open PRs.

        https://developer.github.com/v3/pulls/#list-pull-requests

        Keyword Arguments:
        - state - open (default), closed, all
        - head - Filter by user:branch-name (default unfiltered)
        - base - Filter by base branch name like 'master' (default unfiltered)
        - sort - created (default), updated, popularity, long-running
        - direction - asc, desc. Default is desc when sort=created, or asc
        '''
        url = 'https://api.github.com/repos/%s/%s/pulls' % (self.org,
                                                            self.repo)
        params = self.get_params(state=state, head=head, base=base, sort=sort,
                                 direction=direction)
        results = self.request_all(url, params)
        return results

    def get_pr(self, number):
        '''
        Request PR details.

        https://developer.github.com/v3/pulls/#get-a-single-pull-request

        Keyword Arguments:
        - number: The PR number
        '''
        url = 'https://api.github.com/repos/%s/%s/pulls/%s' % (
            self.org, self.repo, number)
        params = self.get_params()
        details = self.request_detail(url, params)
        return details


def get_pr_report(org, repo, client_id=None, client_secret=None,
                  verbose=False):
    '''Generate report of PRs.'''
    requester = GithubRequester(org, repo, client_id, client_secret, verbose)
    prs = sorted_prs(requester)
    report_lines = []
    for pr in prs:
        changes, number, title, author, html_url = pr
        line = 'PR %s, %s by %s, has %s changes: %s' % (
            number, title, author, changes, html_url)
        report_lines.append(line)
    return '\n'.join(report_lines)


def sorted_prs(requester):
    '''Get open PRs for a project, smallest first.'''
    prs_data = requester.get_prs()
    prs = []
    for pr_data in prs_data:
        number = pr_data['number']
        title = pr_data['title']
        html_url = pr_data['html_url']
        author = pr_data['user']['login']
        detail = requester.get_pr(number)
        additions = detail['additions']
        deletions = detail['deletions']
        sort_data = (
            additions + deletions,
            number,
            title,
            author,
            html_url)
        prs.append(sort_data)
    prs.sort()
    return prs


def get_args():
    parser = ArgumentParser(
        description='List PRs from smallest to largest.',
        epilog=("To create an OAuth app, see "
                " https://github.com/settings/developers"))
    parser.add_argument('--client-id', type=str,
                        help='GitHub OAuth2 client ID')
    parser.add_argument('--client-secret', type=str,
                        help='GitHub OAuth2 client secret')
    parser.add_argument('--cachefile', type=str,
                        help='cache file for GitHub requests',
                        default='github_cache')
    args = parser.parse_args()
    return args


if __name__ == '__main__':
    args = get_args()
    requests_cache.install_cache(args.cachefile,
                                 expire_after=timedelta(days=1))
    report = get_pr_report(
        'mdn', 'browser-compat-data', args.client_id,
        args.client_secret, True)
    print("Done!", file=sys.stderr)
    print(report.encode('utf8'))