GitHubRepo._filer_closed_prs() - Code Metrics - Inspection of "PR: Add batch mode for all tags for all milestones" - spyder-ide/loghub - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#61)

by Gonzalo

created 2017-01-31 17:05 UTC

GitHubRepo._filer_closed_prs() A

↳ Parent: GitHubRepo

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	4
c	3
b	0
f	0
dl	0
loc	14
rs	9.2

# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright (c) The Spyder Development Team
#
# Licensed under the terms of the MIT License
# (See LICENSE.txt for details)
# -----------------------------------------------------------------------------
"""Github repo wrapper."""

from __future__ import print_function

# Standard library imports
import datetime
import sys
import time

# Local imports
from loghub.external.github import ApiError, ApiNotFoundError, GitHub


class GitHubRepo(object):
    """Github repository wrapper."""

    def __init__(self, username=None, password=None, token=None, repo=None):
        """Github repository wrapper."""
        self._username = username
        self._password = password
        self._token = token

        self.gh = GitHub(
            username=username,
            password=password,
            access_token=token, )
        repo_organization, repo_name = repo.split('/')
        self._repo_organization = repo_organization
        self._repo_name = repo_name
        self.repo = self.gh.repos(repo_organization)(repo_name)

        # Check username and repo name
        self._check_user()
        self._check_repo_name()

    def _check_user(self):
        """Check if the supplied username is valid."""
        try:
            self.gh.users(self._repo_organization).get()
        except ApiNotFoundError:
            print('LOGHUB: Organization/user `{}` seems to be '
                  'invalid.\n'.format(self._repo_organization))
            sys.exit(1)
        except ApiError:
            self._check_rate()
            print('LOGHUB: The credentials seems to be invalid!\n')
            sys.exit(1)

    def _check_repo_name(self):
        """Check if the supplied repository exists."""
        try:
            self.repo.get()
        except ApiNotFoundError:
            print('LOGHUB: Repository `{0}` for organization/username `{1}` '
                  'seems to be invalid.\n'.format(self._repo_name,
                                                  self._repo_organization))
            sys.exit(1)
        except ApiError:
            self._check_rate()

    def _check_rate(self):
        """Check and handle if api rate limit has been exceeded."""
        if self.gh.x_ratelimit_remaining == 0:
            reset_struct = time.gmtime(self.gh.x_ratelimit_reset)
            reset_format = time.strftime('%Y/%m/%d %H:%M', reset_struct)
            print('LOGHUB: GitHub API rate limit exceeded!')
            print('LOGHUB: GitHub API rate limit resets on '
                  '{}'.format(reset_format))
            if not self._username and not self._password or not self._token:
                print('LOGHUB: Try running loghub with user/password or '
                      'a valid token.\n')
            sys.exit(1)

    def _filter_milestone(self, issues, milestone):
        """Filter out all issues in milestone."""
        if milestone:
            for issue in issues[:]:
                milestone_data = issue.get('milestone', {})
                if milestone_data:
                    issue_milestone_title = milestone_data.get('title')
                else:
                    issue_milestone_title = ''

                if issue_milestone_title != milestone:
                    issues.remove(issue)
        return issues

    def _filter_since(self, issues, since):
        """Filter out all issues before `since` date."""
        if since:
            since_date = self.str_to_date(since)
            for issue in issues[:]:
                close_date = self.str_to_date(issue['closed_at'])
                if close_date < since_date and issue in issues:
                    issues.remove(issue)
        return issues

    def _filter_until(self, issues, until):
        """Filter out all issues after `until` date."""
        if until:
            until_date = self.str_to_date(until)
            for issue in issues[:]:
                close_date = self.str_to_date(issue['closed_at'])
                if close_date > until_date and issue in issues:
                    issues.remove(issue)
        return issues

    def _filter_by_branch(self, issues, issue, branch):
        """Filter prs by the branch they were merged into."""
        number = issue['number']

        if not self.is_merged(number) and issue in issues:
            issues.remove(issue)

        if branch:
            # Get PR info and get base branch
            pr_data = self.pr(number)
            base_ref = pr_data['base']['ref']

            if base_ref != branch and issue in issues:
                issues.remove(issue)

        return issues

    def _filer_closed_prs(self, issues, branch):
        """Filter out closed PRs."""
        for issue in issues[:]:
            pr = issue.get('pull_request', '')

            # Add label names inside additional key
            issue['loghub_label_names'] = [
                l['name'] for l in issue.get('labels')
            ]

            if pr:
                issues = self._filter_by_branch(issues, issue, branch)

        return issues

    def tags(self):
        """Return all tags."""
        self._check_rate()
        return self.repo('git')('refs')('tags').get()

    def tag(self, tag_name):
        """Get tag information."""
        self._check_rate()
        refs = self.repo('git')('refs')('tags').get()
        sha = -1

        tags = []
        for ref in refs:
            ref_name = 'refs/tags/{tag}'.format(tag=tag_name)
            if 'object' in ref and ref['ref'] == ref_name:
                sha = ref['object']['sha']
            tags.append(ref['ref'].split('/')[-1])

        if sha == -1:
            print("LOGHUB: You didn't pass a valid tag name!")
            print('LOGHUB: The available tags are: {0}\n'.format(tags))
            sys.exit(1)

        return self.repo('git')('tags')(sha).get()

    def labels(self):
        """Return labels for the repo."""
        self._check_rate()
        return self.repo.labels.get()

    def set_labels(self, labels):
        """Return labels for the repo."""
        self._check_rate()
        for label in labels:
            new_name = label['new_name']
            old_name = label['old_name']
            color = label['color']
            try:
                self.repo.labels(old_name).patch(name=new_name, color=color)
                print('Updated label: "{0}" -> "{1}" (#{2})'.format(
                    old_name, new_name, color))
            except ApiError:
                try:
                    self.repo.labels.post(name=new_name, color=color)
                    print('Created label: "{0}" (#{1})'.format(new_name,
                                                               color))
                except ApiError:
                    print('\nLabel "{0}" already exists!'.format(new_name))

    def milestones(self):
        """Return all milestones."""
        self._check_rate()
        return self.repo.milestones.get(state='all')

    def milestone(self, milestone_title):
        """Return milestone with given title."""
        self._check_rate()
        milestones = self.milestones()
        milestone_number = -1

        milestone_titles = [milestone['title'] for milestone in milestones]
        for milestone in milestones:
            if milestone['title'] == milestone_title:
                milestone_number = milestone['number']
                break

        if milestone_number == -1:
            print("LOGHUB: You didn't pass a valid milestone name!")
            print('LOGHUB: The available milestones are: {0}\n'
                  ''.format(milestone_titles))
            sys.exit(1)

        return milestone

    def pr(self, pr_number):
        """Get PR information."""
        self._check_rate()
        return self.repo('pulls')(str(pr_number)).get()

    def issues(self,
               milestone_number=None,
               milestone=None,
               state=None,
               assignee=None,
               creator=None,
               mentioned=None,
               labels=None,
               sort=None,
               direction=None,
               since=None,
               until=None,
               branch=None,
               cache=False,
               base_issues=None):
        """Return Issues and Pull Requests."""
        self._check_rate()
        page = 1

        if not base_issues:
            issues = []
            while True:
                result = self.repo.issues.get(page=page,
                                              per_page=100,
                                              milestone=milestone_number,
                                              state=state,
                                              assignee=assignee,
                                              creator=creator,
                                              mentioned=mentioned,
                                              labels=labels,
                                              sort=sort,
                                              direction=direction,
                                              since=since)
                if len(result) > 0:
                    issues += result
                    page = page + 1
                else:
                    break
        else:
            issues = base_issues

        # If since was provided, filter the issue
        issues = self._filter_since(issues, since)

        # If until was provided, filter the issue
        issues = self._filter_until(issues, until)

        # If milestone was provided, filter the issue
        issues = self._filter_milestone(issues, milestone)

        # If it is a pr check if it is merged or closed, removed closed ones
        issues = self._filer_closed_prs(issues, branch)

        return issues

    def is_merged(self, pr):
        """
        Return wether a PR was merged, or if it was closed and discarded.

        https://developer.github.com/v3/pulls/#get-if-a-pull-request-has-been-merged
        """
        self._check_rate()
        merged = True
        try:
            self.repo('pulls')(str(pr))('merge').get()
        except Exception:
            merged = False
        return merged

    @staticmethod
    def str_to_date(string):
        """Convert ISO date string to datetime object."""
        parts = string.split('T')
        date_parts = parts[0]
        time_parts = parts[1][:-1]
        year, month, day = [int(i) for i in date_parts.split('-')]
        hour, minutes, seconds = [int(i) for i in time_parts.split(':')]
        return datetime.datetime(year, month, day, hour, minutes, seconds)


1			# -- coding: utf-8 --
2			# -----------------------------------------------------------------------------
3			# Copyright (c) The Spyder Development Team
4			#
5			# Licensed under the terms of the MIT License
6			# (See LICENSE.txt for details)
7			# -----------------------------------------------------------------------------
8			"""Github repo wrapper."""
9
10			from __future__ import print_function
11
12			# Standard library imports
13			import datetime
14			import sys
15			import time
16
17			# Local imports
18			from loghub.external.github import ApiError, ApiNotFoundError, GitHub
19
20
21			class GitHubRepo(object):
22			"""Github repository wrapper."""
23
24			def __init__(self, username=None, password=None, token=None, repo=None):
25			"""Github repository wrapper."""
26			self._username = username
27			self._password = password
28			self._token = token
29
30			self.gh = GitHub(
31			username=username,
32			password=password,
33			access_token=token, )
34			repo_organization, repo_name = repo.split('/')
35			self._repo_organization = repo_organization
36			self._repo_name = repo_name
37			self.repo = self.gh.repos(repo_organization)(repo_name)
38
39			# Check username and repo name
40			self._check_user()
41			self._check_repo_name()
42
43			def _check_user(self):
44			"""Check if the supplied username is valid."""
45			try:
46			self.gh.users(self._repo_organization).get()
47			except ApiNotFoundError:
48			print('LOGHUB: Organization/user `{}` seems to be '
49			'invalid.\n'.format(self._repo_organization))
50			sys.exit(1)
51			except ApiError:
52			self._check_rate()
53			print('LOGHUB: The credentials seems to be invalid!\n')
54			sys.exit(1)
55
56			def _check_repo_name(self):
57			"""Check if the supplied repository exists."""
58			try:
59			self.repo.get()
60			except ApiNotFoundError:
61			print('LOGHUB: Repository `{0}` for organization/username `{1}` '
62			'seems to be invalid.\n'.format(self._repo_name,
63			self._repo_organization))
64			sys.exit(1)
65			except ApiError:
66			self._check_rate()
67
68			def _check_rate(self):
69			"""Check and handle if api rate limit has been exceeded."""
70			if self.gh.x_ratelimit_remaining == 0:
71			reset_struct = time.gmtime(self.gh.x_ratelimit_reset)
72			reset_format = time.strftime('%Y/%m/%d %H:%M', reset_struct)
73			print('LOGHUB: GitHub API rate limit exceeded!')
74			print('LOGHUB: GitHub API rate limit resets on '
75			'{}'.format(reset_format))
76			if not self._username and not self._password or not self._token:
77			print('LOGHUB: Try running loghub with user/password or '
78			'a valid token.\n')
79			sys.exit(1)
80
81			def _filter_milestone(self, issues, milestone):
82			"""Filter out all issues in milestone."""
83			if milestone:
84			for issue in issues[:]:
85			milestone_data = issue.get('milestone', {})
86			if milestone_data:
87			issue_milestone_title = milestone_data.get('title')
88			else:
89			issue_milestone_title = ''
90
91			if issue_milestone_title != milestone:
92			issues.remove(issue)
93			return issues
94
95			def _filter_since(self, issues, since):
96			"""Filter out all issues before `since` date."""
97			if since:
98			since_date = self.str_to_date(since)
99			for issue in issues[:]:
100			close_date = self.str_to_date(issue['closed_at'])
101			if close_date < since_date and issue in issues:
102			issues.remove(issue)
103			return issues
104
105			def _filter_until(self, issues, until):
106			"""Filter out all issues after `until` date."""
107			if until:
108			until_date = self.str_to_date(until)
109			for issue in issues[:]:
110			close_date = self.str_to_date(issue['closed_at'])
111			if close_date > until_date and issue in issues:
112			issues.remove(issue)
113			return issues
114
115			def _filter_by_branch(self, issues, issue, branch):
116			"""Filter prs by the branch they were merged into."""
117			number = issue['number']
118
119			if not self.is_merged(number) and issue in issues:
120			issues.remove(issue)
121
122			if branch:
123			# Get PR info and get base branch
124			pr_data = self.pr(number)
125			base_ref = pr_data['base']['ref']
126
127			if base_ref != branch and issue in issues:
128			issues.remove(issue)
129
130			return issues
131
132			def _filer_closed_prs(self, issues, branch):
133			"""Filter out closed PRs."""
134			for issue in issues[:]:
135			pr = issue.get('pull_request', '')
136
137			# Add label names inside additional key
138			issue['loghub_label_names'] = [
139			l['name'] for l in issue.get('labels')
140			]
141
142			if pr:
143			issues = self._filter_by_branch(issues, issue, branch)
144
145			return issues
146
147			def tags(self):
148			"""Return all tags."""
149			self._check_rate()
150			return self.repo('git')('refs')('tags').get()
151
152			def tag(self, tag_name):
153			"""Get tag information."""
154			self._check_rate()
155			refs = self.repo('git')('refs')('tags').get()
156			sha = -1
157
158			tags = []
159			for ref in refs:
160			ref_name = 'refs/tags/{tag}'.format(tag=tag_name)
161			if 'object' in ref and ref['ref'] == ref_name:
162			sha = ref['object']['sha']
163			tags.append(ref['ref'].split('/')[-1])
164
165			if sha == -1:
166			print("LOGHUB: You didn't pass a valid tag name!")
167			print('LOGHUB: The available tags are: {0}\n'.format(tags))
168			sys.exit(1)
169
170			return self.repo('git')('tags')(sha).get()
171
172			def labels(self):
173			"""Return labels for the repo."""
174			self._check_rate()
175			return self.repo.labels.get()
176
177			def set_labels(self, labels):
178			"""Return labels for the repo."""
179			self._check_rate()
180			for label in labels:
181			new_name = label['new_name']
182			old_name = label['old_name']
183			color = label['color']
184			try:
185			self.repo.labels(old_name).patch(name=new_name, color=color)
186			print('Updated label: "{0}" -> "{1}" (#{2})'.format(
187			old_name, new_name, color))
188			except ApiError:
189			try:
190			self.repo.labels.post(name=new_name, color=color)
191			print('Created label: "{0}" (#{1})'.format(new_name,
192			color))
193			except ApiError:
194			print('\nLabel "{0}" already exists!'.format(new_name))
195
196			def milestones(self):
197			"""Return all milestones."""
198			self._check_rate()
199			return self.repo.milestones.get(state='all')
200
201			def milestone(self, milestone_title):
202			"""Return milestone with given title."""
203			self._check_rate()
204			milestones = self.milestones()
205			milestone_number = -1
206
207			milestone_titles = [milestone['title'] for milestone in milestones]
208			for milestone in milestones:
209			if milestone['title'] == milestone_title:
210			milestone_number = milestone['number']
211			break
212
213			if milestone_number == -1:
214			print("LOGHUB: You didn't pass a valid milestone name!")
215			print('LOGHUB: The available milestones are: {0}\n'
216			''.format(milestone_titles))
217			sys.exit(1)
218
219			return milestone
220
221			def pr(self, pr_number):
222			"""Get PR information."""
223			self._check_rate()
224			return self.repo('pulls')(str(pr_number)).get()
225
226			def issues(self,
227			milestone_number=None,
228			milestone=None,
229			state=None,
230			assignee=None,
231			creator=None,
232			mentioned=None,
233			labels=None,
234			sort=None,
235			direction=None,
236			since=None,
237			until=None,
238			branch=None,
239			cache=False,
240			base_issues=None):
241			"""Return Issues and Pull Requests."""
242			self._check_rate()
243			page = 1
244
245			if not base_issues:
246			issues = []
247			while True:
248			result = self.repo.issues.get(page=page,
249			per_page=100,
250			milestone=milestone_number,
251			state=state,
252			assignee=assignee,
253			creator=creator,
254			mentioned=mentioned,
255			labels=labels,
256			sort=sort,
257			direction=direction,
258			since=since)
259			if len(result) > 0:
260			issues += result
261			page = page + 1
262			else:
263			break
264			else:
265			issues = base_issues
266
267			# If since was provided, filter the issue
268			issues = self._filter_since(issues, since)
269
270			# If until was provided, filter the issue
271			issues = self._filter_until(issues, until)
272
273			# If milestone was provided, filter the issue
274			issues = self._filter_milestone(issues, milestone)
275
276			# If it is a pr check if it is merged or closed, removed closed ones
277			issues = self._filer_closed_prs(issues, branch)
278
279			return issues
280
281			def is_merged(self, pr):
282			"""
283			Return wether a PR was merged, or if it was closed and discarded.
284
285			https://developer.github.com/v3/pulls/#get-if-a-pull-request-has-been-merged
286			"""
287			self._check_rate()
288			merged = True
289			try:
290			self.repo('pulls')(str(pr))('merge').get()
291			except Exception:
292			merged = False
293			return merged
294
295			@staticmethod
296			def str_to_date(string):
297			"""Convert ISO date string to datetime object."""
298			parts = string.split('T')
299			date_parts = parts[0]
300			time_parts = parts[1][:-1]
301			year, month, day = [int(i) for i in date_parts.split('-')]
302			hour, minutes, seconds = [int(i) for i in time_parts.split(':')]
303			return datetime.datetime(year, month, day, hour, minutes, seconds)
304

spyder-ide / loghub

Pull Request — master (#61)

GitHubRepo._filer_closed_prs() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like