Skip to content

Instantly share code, notes, and snippets.

Created April 2, 2020 10:48
Show Gist options
  • Save Akasurde/fc5dc49e5a475e7e6a1311e0b4a560b1 to your computer and use it in GitHub Desktop.
Save Akasurde/fc5dc49e5a475e7e6a1311e0b4a560b1 to your computer and use it in GitHub Desktop.
Python script to help with migrating issues and PRs from ansible/ansible to a target collection. Original Script -
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright: (c) 2020, Jordan Borean (@jborean93) <>
# MIT License (see LICENSE or
Script that can be used to copy issues and PRs from the Ansible GitHub repo to it's target collection repo. Current
limitations are;
* Original issue/PR authors are lost in migration, they will appear as your user with a comment in the post stating the
original author
* PR review comments will be missing, only comments on the PR itself (not the code) will be migrated
* Has a very basic patch path rewriter, probably will fall short on some examples
To migrate an issue run
./ -i 1234 -c --token gh_pat
To migrate a PR run
./ -p 1234 -c -r --token gh_pat
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import argparse
import contextlib
import json
import os
import re
import subprocess
import tempfile
from datetime import datetime
import argcomplete
except ImportError:
argcomplete = None
from urllib.request import Request, urlopen
except ImportError: # Python 2
from urllib2 import Request, urlopen
def open_url(url, method=None, data=None, headers=None):
req = Request(url, method=method, data=data, headers=headers)
resp = urlopen(req)
yield resp
def _gh_call(args, url, method=None, data=None, json_response=True):
headers = {'Authorization': 'token %s' % args.gh_token}
with open_url(url, method=method, data=data, headers=headers) as resp:
resp_data ='utf-8')
if json_response:
return json.loads(resp_data)
return resp_data
def parse_args():
parser = argparse.ArgumentParser(description='Migrate a PR or Issue from ansible/ansible to the specified '
'collection repo.')
id_group = parser.add_mutually_exclusive_group(required=True)
id_group.add_argument('-i', '--issue',
help='The ansible/ansible Issue to migrate.')
id_group.add_argument('-p', '--pull-request',
help='The ansible/ansible PR to migrate.')
parser.add_argument('-c', '--target-collection',
help='The target collection name under ansible-collections to migrate the PR/Issue to.')
parser.add_argument('-r', '--target-repo',
help='The target repository URI to push the new branch to.')
parser.add_argument('-b', '--base-branch',
help='When migrating pull requests, this is the target branch of the collection repo that the '
'changes will be targeted to.')
# To generate a new token for this argument go to Github:
# User Settings -> Developer settings -> Personal access tokens -> Generate new token
# Call it whatever you want but it needs the following scopes
# public_repo
# read:org
def get_token():
token = None
github_api_file = os.path.join(os.path.expanduser("~"), '.github_api')
with open(github_api_file) as file_:
token ="\n")
return token
parser.add_argument('-t', '--token',
default=get_token() or os.environ.get('GITHUB_TOKEN'),
help='The GitHub password or personal access token to authenticate with, defaults to '
'GITHUB_PASS env var.')
if argcomplete:
args = parser.parse_args()
if args.gh_pr and not args.target_repo:
raise ValueError("--target-repo must be set when migrating a pull request.")
if not args.gh_token:
raise ValueError("--token must be set through the CLI or with GITHUB_TOKEN env var.")
return args
def gh_patch_close_issue(args, issue_info):
data = json.dumps({'state': 'closed'}).encode('utf-8')
return _gh_call(args, issue_info['url'] , method='PATCH', data=data)
def gh_get_repos_pulls(args, owner, repo, pr_id):
return _gh_call(args, '%srepos/%s/%s/pulls/%s' % (GITHUB_API_ROOT, owner, repo, pr_id))
def gh_get_repos_issues(args, owner, repo, issue_id):
return _gh_call(args, '%srepos/%s/%s/issues/%s' % (GITHUB_API_ROOT, owner, repo, issue_id))
def gh_post_repos_pulls(args, owner, repo, title, head, base, body):
data = json.dumps({
'title': title,
'head': head,
'base': base,
'body': body,
'maintainer_can_modify': True,
return _gh_call(args, '%srepos/%s/%s/pulls' % (GITHUB_API_ROOT, owner, repo), method='POST', data=data)
def gh_post_repos_issues(args, owner, repo, title, body):
data = json.dumps({
'title': title,
'body': body,
return _gh_call(args, '%srepos/%s/%s/issues' % (GITHUB_API_ROOT, owner, repo), method='POST', data=data)
def gh_post_repos_issues_comments(args, issue_info, body):
data = json.dumps({'body': body}).encode('utf-8')
return _gh_call(args, issue_info['comments_url'], method='POST', data=data)
def get_github_comments(args, source_info):
comments = []
if source_info['comments'] > 0:
comments = [c for c in _gh_call(args, source_info['comments_url'])
if c.get('user', {}).get('login') != 'ansibot']
return comments
def get_ansible_issue(args, issue_id):
source_info = gh_get_repos_issues(args, 'ansible', 'ansible', issue_id)
return {
'post': source_info,
'comments': get_github_comments(args, source_info)
def get_ansible_pr(args, pr_id):
source_info = gh_get_repos_pulls(args, 'ansible', 'ansible', pr_id)
return {
'post': source_info,
'comments': get_github_comments(args, source_info)
def format_migrated_body(post):
source_date = datetime.strptime(post['created_at'], '%Y-%m-%dT%H:%M:%Sz')
new_post = '''_From @{0} on {1}_
{2}'''.format(post['user']['login'], source_date.strftime('%b %d, %Y %H:%M'), post['body'])
if 'number' in post:
new_post += '\n\n_Copied from original issue: ansible/ansible#{0}_'.format(post['number'])
return new_post
def main():
args = parse_args()
if args.gh_issue:
print("Getting existing GitHub issue %s info" % args.gh_issue)
source_info = get_ansible_issue(args, args.gh_issue)
new_issue_body = format_migrated_body(source_info['post'])
title = source_info['post']['title']
print("Creating new GitHub issue at ansible-collections/%s - %s" % (args.target_collection, title))
info = gh_post_repos_issues(args, 'ansible-collections', args.target_collection, title, new_issue_body)
print("Created new GitHub issue at '%s'" % info['html_url'])
# Comment 'migrate_to' on source issue
gh_post_repos_issues_comments(args, source_info['post'], 'Issue migrated to %s' % info['html_url'])
# Close the source issue
print("Closing %s" % source_info['post']['html_url'])
gh_patch_close_issue(args, source_info['post'])
print("Closed %s" % source_info['post']['html_url'])
print("Getting existing GitHub PR %s info" % args.gh_pr)
source_info = get_ansible_pr(args, args.gh_pr)
new_branch = 'ansible_migration_%s' % source_info['post']['number']
title = source_info['post']['title']
user_match = re.match(r'.*github\.com[/|:](\w+)/%s(\.git)?$' % re.escape(args.target_collection),
if not user_match:
raise ValueError("Failed to get username from target repo URL")
head = '%s:%s' % (, new_branch)
base = args.base_branch
# Get patch for original PR and do a poor man's path replacement.
print("Getting PR patch and rewrite paths")
pr_patch = _gh_call(args, source_info['post']['patch_url'], json_response=False)
pr_patch = (
# Use regex to handle nested module directories to a flat dir.
re.sub(r'lib/ansible/modules/([\w]*/?)(\w+)\.(\w)', r'plugins/modules/\2.\3', pr_patch)
# This is a poor man's replacement, could be better but should be ok for general cases.
.replace('lib/ansible/plugins/', 'plugins/')
.replace('test/integration/', 'tests/integration/')
.replace('test/units/', 'tests/units/')
# Checkout the target branch and try and apply the patch and push to the repo.
with tempfile.TemporaryDirectory() as temp_dir:
repo_dir = os.path.join(temp_dir, args.target_collection)
print("Creating new clone of git repo '%s'" % args.target_repo)['git', 'clone', args.target_repo, repo_dir])
print("Creating new git branch '%s'" % new_branch)['git', 'checkout', '-b', new_branch], cwd=repo_dir)
patch_path = os.path.join(temp_dir, 'my.patch')
with open(patch_path, mode='wb') as patch_fd:
print("Applying git patch")['git', 'am', patch_path], cwd=repo_dir)
print("Pushing branch to remote")['git', 'push', '-u', 'origin', new_branch, '--force'], cwd=repo_dir)
new_pr_body = format_migrated_body(source_info['post'])
print("Creating new GitHub PR at ansible-collections/%s - %s" % (args.target_collection, title))
info = gh_post_repos_pulls(args, 'ansible-collections', args.target_collection, title, head, base, new_pr_body)
print("Created new GitHub PR at '%s'" % info['html_url'])
for idx, comment in enumerate(source_info['comments'], 1):
new_comment = format_migrated_body(comment)
print("Copying comment %d of %d" % (idx, len(source_info['comments'])))
gh_post_repos_issues_comments(args, info, new_comment)
if __name__ == '__main__':
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment