git-pull-request

#!/usr/bin/env python
#
# git-pull-request
#
# Uses the credentials stored by `git-github-login` to open a pull request for
# the current branch.
#
from backports import configparser as cp
import os
import os.path as path
import git
import util
import re
import requests
from requests.auth import HTTPBasicAuth


def get_reviewer_aliases(repo):
  """
  Query git config to get reviewer aliases.

  Useful if, e.g., you frequently request reviews from people with long or
  hard-to-remember git handles -- alias them to something easier!

  Aliases should be stored as key/value pairs in your git config, e.g.:
    [gitworkflow "alias"]
      alice = alice_has_a_long_git_handle
      bob = hard2remember31415
  """
  reader = repo.config_reader()

  # HACK(maiamcc): make sure the data is present for call to reader.items
  # (see https://github.com/gitpython-developers/GitPython/issues/888)
  _ = reader.sections()
  try:
    kv_pairs = reader.items("gitworkflow \"alias\"")
  except cp.NoSectionError:
    return {}
  except Exception as e:
    # Catch NoSectionError in a way that's compatible
    # across python versions.
    if 'NoSectionError' in str(type(e)):
      return {}
    raise e

  return {str(pair[0]): str(pair[1]) for pair in kv_pairs}


def alias_reviewers(reviewers, aliases):
  """
  Replace any reviewer aliases with the appropriate Github handle.
  """
  for i, r in enumerate(reviewers):
    if r in aliases:
      reviewers[i] = aliases[r]

  return reviewers


try:
  repo = git.Repo(os.getcwd(), search_parent_directories=True)
except git.exc.InvalidGitRepositoryError:
  util.fatal('git pull-request must be run from within a valid git repository.')

util.fatal_if_dirty(repo)

if not repo.remotes.origin.exists():
  util.fatal('git pull-request requires a configured origin')

creds = util.get_github_creds()

local_branch = repo.active_branch
local_sha = repo.head.object.hexsha
latest_commit = repo.head.commit.message.split('\n', 1)[0]

remote_origin = repo.remotes.origin
remote_branch = repo.active_branch # TODO: This might be not be true.
remote_ref = '%s/%s' % (remote_origin, remote_branch)

repo_root = repo.git.rev_parse("--show-toplevel")
repo_url = list(repo.remotes.origin.urls)[0]

# Important things to match:
# git@github.com:user/repo
# git@github.com:user/repo.with.dot
# git@github.com:user/repo.git
# git@github.com:user/repo.with.dot.git
m = re.search(':(.+?)(?:\.git)?$', repo_url)
if m:
  repo_path = m.group(1)
else:
  util.fatal('Failed to determine repo path: %s' % repo_url)

util.info('Checking for changes against %s' % remote_ref)

# Check remote is up to date
repo.remotes.origin.fetch()
repo.git.remote('prune', 'origin')
try:
  diff = repo.git.diff(local_branch, remote_ref)
except git.exc.GitCommandError:
  push_it = util.prompt_y_n('Remote branch %s not available; attempt to '
    'push it up?' % remote_ref, default=True)
  if push_it:
    split_remote_ref = remote_ref.split('/', 1)
    if len(split_remote_ref) != 2:
      util.fatal('Remote ref name "%s" in unexpected format (expected '
        '"[origin]/[branchname]"' % remote_ref)
      exit(1)
    repo.git.push([split_remote_ref[0], split_remote_ref[1]])
    try:
      # Try again to diff against remote branch
      diff = repo.git.diff(local_branch, remote_ref)
    except git.exc.GitCommandError:
      util.info('Remote branch %s is STILL not available, nothing to do.' % remote_ref)
      exit(1)
  else:
    util.warn('OK, aborting.')
    exit(1)

if diff != '':
  util.error('Local and remote branches are not in sync.')
  show_diff = util.prompt_y_n('See diff?')
  if show_diff:
    util.warn(diff)
  exit(1)

# Get changelog vs. main
main = util.main_branch_name(repo)
changes = repo.git.log('--pretty=%H (%ci)%n%s%n%b', '--abbrev-commit',
                       '--first-parent', '--no-merges', '--date=short',
                       '%s..%s' % (main, local_branch))

if changes == '':
  util.fatal('There are no changes to submit.')

# Prompt for PR title (default to last commit)
title = util.prompt('Enter a short title for the pull request', latest_commit)

# FUTURE: ask Y/N questions in .gitchecklist

reviewers = util.prompt('Enter reviewer', '')
reviewer_list = re.split('[,\s]+', reviewers.strip())
reviewer_list = alias_reviewers(reviewer_list, get_reviewer_aliases(repo))

# Load PR template, substitute variables.

default_template = path.join(util.get_script_path(), '.github', 'pull_request_template.md')
local_template_root = path.join(repo_root, 'pull_request_template.md')
local_template_hidden = path.join(repo_root, '.github', 'pull_request_template.md')
local_template_docs = path.join(repo_root, 'docs', 'pull_request_template.md')

if path.isfile(local_template_root):
  template_file = local_template_root
elif path.isfile(local_template_hidden):
  template_file = local_template_hidden
elif path.isfile(local_template_docs):
  template_file = local_template_docs
else:
  template_file = default_template

with open(template_file, 'r') as file:
  description = file.read() % {
    "reviewer": ', '.join([('@%s' % r) for r in reviewer_list]) or 'tbd',
    "branch": local_branch,
    "changes": changes.strip()
  }

# Add a help message to thet templated description.
description = """// Please enter a description for the pull request. Lines starting
// with '//' will be ignored, and an empty message aborts the request.

%s""" % description

# Allow the user to edit the description.
description = util.edit(repo, description)

# Strip comment lines.
description = '\n'.join([x for x in description.splitlines() if not x.startswith('//')]).strip()

if not description:
  util.fatal('Empty message, aborting.')

# Send the PR request to GitHub.
api_url = 'https://api.github.com/repos/%s/pulls' % repo_path
payload = {
  "title": title,
  "body": description,
  "head": remote_branch.name,
  "base": main
}

try:
  r = requests.post(api_url,
                    json=payload,
                    auth=HTTPBasicAuth(creds['username'], creds['token']))
except:
  err = sys.exc_info()[0]
  util.fatal('Failed to login. Request Failed. %s' % err)

if not r.ok:
  try:
    msg = ', '.join([m['message'] for m in r.json()["errors"]])
  except:
    msg = 'Request failed. Status %d' % r.status_code
  util.fatal('Failed to send PR. %s.' % msg)

data = r.json()
issue_id = data['id']
issue_number = data['number']
issue_url = data['html_url']

if reviewer_list:
  api_url = 'https://api.github.com/repos/%s/pulls/%s/requested_reviewers' % (repo_path, issue_number)
  payload = {"reviewers": reviewer_list}
  try:
    r = requests.post(api_url,
                      json=payload,
                      auth=HTTPBasicAuth(creds['username'], creds['token']))
  except:
    err = sys.exc_info()[0]
    util.error('Failed to request reviewers. Ignoring error: %s' % err)

util.success('Whoop! PR was sent! %s' % issue_url)

try:
  if os.system('which pbcopy') == 0: # osx
    os.system("echo '%s' | pbcopy" % issue_url)
    util.info('The issue URL was copied to your clipboard.')
  elif os.system('which xclip') == 0: # linux
    os.system("echo '%s' | xclip -selection clipboard" % issue_url)
    util.info('The issue URL was copied to your clipboard.')
except:
  pass