Last active
May 19, 2022 08:48
-
-
Save martyngigg/53d37b3fc99dab65ea7fadb76ed07146 to your computer and use it in GitHub Desktop.
Close issues linked to merged pull requests by milestone
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""Examines all issues assigned to a given milestone and if they are associated with | |
a merged pull request that is linked to fix the issue then it closes that issue. | |
This module exists as GitHub only closes issues linked with 'Fixes #1234' if a | |
pull request was merged to the default branch. During a beta period pull requests | |
merged to the release branch leaving dangling issues around. | |
Requires the pygithub module. | |
""" | |
import argparse | |
import logging | |
import posixpath | |
import pprint | |
import re | |
import sys | |
from typing import Optional, Sequence, Mapping | |
import github | |
DEFAULT_REPO = 'mantidproject/mantid' | |
ISSUE_CLOSE_RE = re.compile( | |
r'\b(?:close|closes|closed|fix|fixes|fixed|resolve|resolves|resolved)\b\s+#(\d+)', | |
flags=re.IGNORECASE) | |
class _HashableGithubIssue: | |
"""Thin extension to store an issue as a dict key""" | |
def __init__(self, issue: github.Issue.Issue): | |
self._issue = issue | |
def __hash__(self): | |
return self._issue.number | |
# Link issues to pull requests | |
IssueToPRs = Mapping[_HashableGithubIssue, Sequence[github.PullRequest.PullRequest]] | |
class IssuesToCloseInfo: | |
"""Simple type to aggreagate summary information""" | |
_issues: IssueToPRs = None | |
def __init__(self): | |
self._issues = dict() | |
def add_issue(self, issue: github.Issue.Issue, pull_request: github.PullRequest.PullRequest): | |
prs = self._issues.setdefault(_HashableGithubIssue(issue), []) | |
prs.append(pull_request) | |
def close_all(self): | |
"""Edits the state of the issues to closed""" | |
for issue, prs in self._issues.items(): | |
issue._issue.edit(state='closed') | |
pr_refs = '#' + ',#'.join(map(lambda x: str(x.number), prs)) | |
issue._issue.create_comment(f'Fixed by {pr_refs}') | |
def prettyprint(self, writer): | |
lines = [f'{len(self._issues)} open issues are referenced to be closed'] | |
row_format = '{:^15}|{:^15}' | |
lines.append(row_format.format('Issue', 'Pulls')) | |
lines.append('-' * 30) | |
for issue, prs in self._issues.items(): | |
formatted_prs = ','.join(map(lambda x: str(x.number), prs)) | |
lines.append(row_format.format(issue._issue.number, formatted_prs)) | |
writer('\n'.join(lines)) | |
def enable_console_logging(): | |
""" | |
This function sets up a very simple logging configuration (log everything on standard output) that is useful for troubleshooting. | |
""" | |
logging.basicConfig(level=logging.INFO) | |
def main() -> int: | |
"""Main cli entry point""" | |
enable_console_logging() | |
args = parse_arguments() | |
github_api = github.Github(args.token) | |
repository = github_api.get_repo(args.repo) | |
milestone = None | |
if args.milestone is not None: | |
milestone = find_matching_milestone(repository, args.milestone) | |
if milestone is None: | |
logging.error( | |
f'Unable to find milestone with title "{args.milestone}" in repository "{args.repo}"') | |
return 1 | |
close_issues_with_merged_pull_request(repository, milestone, args.dry_run) | |
return 0 | |
def parse_arguments() -> argparse.Namespace: | |
""" | |
Parse options given to program | |
:return: The arguments as a argparse object | |
""" | |
parser = argparse.ArgumentParser(description="Close issues fixed by merged pull requests") | |
parser.add_argument("--repo", | |
dest="repo", | |
default=DEFAULT_REPO, | |
help="The full org/repo name of a GitHubrepository") | |
parser.add_argument("--token", required=True, help="GitHub authorization token") | |
parser.add_argument( | |
"--milestone", help="An optional milestone on the project. Checks all open issues for this milestone.") | |
parser.add_argument( | |
"--dry-run", help="If true, just print what would happen but don't do anything", action='store_true') | |
return parser.parse_args() | |
def find_matching_milestone(repository: github.Repository.Repository, | |
milestone_title: str) -> Optional[github.Milestone.Milestone]: | |
"""Given a milestone title, find the matching milestone | |
in the Github repo""" | |
for milestone in repository.get_milestones(state='open'): | |
if milestone.title == milestone_title: | |
return milestone | |
return None | |
def close_issues_with_merged_pull_request(repository: github.Repository.Repository, | |
milestone: Optional[github.Milestone.Milestone] = None, dry_run: bool = False): | |
""" | |
For each open issue, optionally constrained to a milestone,: | |
- check if a pull request is connected to it | |
- if the pull request fixes/closes the connected issue then close the issue | |
:param repository: A repository object | |
:param milestone: An optional Milestone object | |
:param dry_run: If True no action is take but the steps are printed | |
""" | |
if milestone: | |
open_issues = repository.get_issues(milestone=milestone, state='open') | |
logging.info(f'Milestone "{milestone.title}" has {open_issues.totalCount} open issues') | |
else: | |
open_issues = repository.get_issues(state='open') | |
logging.info(f'Repository has {open_issues.totalCount} open issues') | |
logging.info('Checking for merged linked pull requests') | |
# Github does not provide an easy between pull requests & issues | |
# We use the timeline on the issue to look for a referenced event and extract the pull_request from the raw_data | |
issues_to_close = find_issues_fixed_by_pull_requests(open_issues) | |
issues_to_close.prettyprint(logging.info) | |
if dry_run: | |
logging.info('Dry run requested. No close action taken') | |
else: | |
logging.info('Closing these issues') | |
issues_to_close.close_all() | |
def find_issues_fixed_by_pull_requests( | |
issues: Sequence[github.Issue.Issue]) -> Sequence[github.Issue.Issue]: | |
""" | |
Filter the list of issues and return any fixed/closed by a pull request | |
:param issues: A sequence of Issue objects | |
:return: True if the issue was close, false otherwise | |
""" | |
fixed_issues = IssuesToCloseInfo() | |
for issue in issues: | |
prs = get_linked_pull_requests(issue) | |
if len(prs) == 0: | |
continue | |
for pr in prs: | |
if is_fixed_by(issue, pr): | |
fixed_issues.add_issue(issue, pr) | |
return fixed_issues | |
def get_linked_pull_requests(issue: github.Issue.Issue) -> Sequence[github.PullRequest.PullRequest]: | |
""" | |
Returns linked pull requests if any exist. | |
""" | |
# There is no first-class connection between an issue & a pull request | |
# so we use the events timeline to search for a cross-referenced event | |
# that contains a link to the pull request | |
def get_pr_number(event): | |
if event.event != 'cross-referenced': | |
return | |
raw_data = event.raw_data['source']['issue'] | |
pull_request = raw_data.get('pull_request', None) | |
if pull_request is None: | |
return | |
return int(posixpath.basename(pull_request['url'])) | |
pr_numbers = filter(lambda x: x is not None, map(get_pr_number, issue.get_timeline())) | |
repository = issue.repository | |
return [repository.get_pull(pr_number) for pr_number in pr_numbers] | |
def is_fixed_by(issue: github.Issue.Issue, pull_request: github.PullRequest.PullRequest) -> bool: | |
""" | |
Returns True if the issue is fixed by the given pull request and that pull request is merged | |
:param issue: An Issue object | |
:param pull_request: A PullRequest object | |
""" | |
if not pull_request.merged: | |
return False | |
# We use the body of the pull request and check for any of the keywords defined by GitHub | |
# that close an issue | |
# https://docs.github.com/en/github/managing-your-work-on-github/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword | |
pr_closed_issues = ISSUE_CLOSE_RE.findall(pull_request.body) | |
return str(issue.number) in pr_closed_issues | |
if __name__ == '__main__': | |
sys.exit(main()) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
PyGithub |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment