GitHubRepo._check_user() - Code Metrics - Inspection of "PR: Break into smaller funcs" - spyder-ide/loghub - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#45)

by Gonzalo

created 2016-12-16 04:43 UTC

GitHubRepo._check_user() A

↳ Parent: GitHubRepo

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	3
c	1
b	0
f	0
dl	0
loc	12
rs	9.4285

# -*- 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

# 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 as error:
            print('LOGHUB: Organization/user `{}` seems to be '
                  'invalid.\n'.format(self._repo_organization))
            sys.exit(1)
        except ApiError as error:
            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 as error:
            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 as error:
            self._check_rate()

    def _check_rate(self):
        """Check and handle if api rate limit has been exceeded."""
        if self.gh.x_ratelimit_remaining == 0:
            print('LOGHUB: GitHub API rate limit exceeded!\n')
            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_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):
        """"""
        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 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 = []
        for milestone in milestones:
            if milestone['title'] == milestone_title:
                milestone_number = milestone['number']
            milestone_titles.append(milestone['title'])

        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=None,
               state=None,
               assignee=None,
               creator=None,
               mentioned=None,
               labels=None,
               sort=None,
               direction=None,
               since=None,
               until=None,
               branch=None):
        """Return Issues and Pull Requests."""
        self._check_rate()
        page = 1
        issues = []
        while True:
            result = self.repo.issues.get(page=page,
                                          per_page=100,
                                          milestone=milestone,
                                          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

        # 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 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
16			# Local imports
17			from loghub.external.github import ApiError, ApiNotFoundError, GitHub
18
19
20			class GitHubRepo(object):
21			"""Github repository wrapper."""
22
23			def __init__(self, username=None, password=None, token=None, repo=None):
24			"""Github repository wrapper."""
25			self._username = username
26			self._password = password
27			self._token = token
28
29			self.gh = GitHub(
30			username=username,
31			password=password,
32			access_token=token, )
33			repo_organization, repo_name = repo.split('/')
34			self._repo_organization = repo_organization
35			self._repo_name = repo_name
36			self.repo = self.gh.repos(repo_organization)(repo_name)
37
38			# Check username and repo name
39			self._check_user()
40			self._check_repo_name()
41
42			def _check_user(self):
43			"""Check if the supplied username is valid."""
44			try:
45			self.gh.users(self._repo_organization).get()
46			except ApiNotFoundError as error:
47			print('LOGHUB: Organization/user `{}` seems to be '
48			'invalid.\n'.format(self._repo_organization))
49			sys.exit(1)
50			except ApiError as error:
51			self._check_rate()
52			print('LOGHUB: The credentials seems to be invalid!\n')
53			sys.exit(1)
54
55			def _check_repo_name(self):
56			"""Check if the supplied repository exists."""
57			try:
58			self.repo.get()
59			except ApiNotFoundError as error:
60			print('LOGHUB: Repository `{0}` for organization/username `{1}` '
61			'seems to be invalid.\n'.format(self._repo_name,
62			self._repo_organization))
63			sys.exit(1)
64			except ApiError as error:
65			self._check_rate()
66
67			def _check_rate(self):
68			"""Check and handle if api rate limit has been exceeded."""
69			if self.gh.x_ratelimit_remaining == 0:
70			print('LOGHUB: GitHub API rate limit exceeded!\n')
71			if not self._username and not self._password or not self._token:
72			print('LOGHUB: Try running loghub with user/password or '
73			'a valid token.\n')
74			sys.exit(1)
75
76			def _filter_since(self, issues, since):
77			"""Filter out all issues before `since` date."""
78			if since:
79			since_date = self.str_to_date(since)
80			for issue in issues[:]:
81			close_date = self.str_to_date(issue['closed_at'])
82			if close_date < since_date and issue in issues:
83			issues.remove(issue)
84			return issues
85
86			def _filter_until(self, issues, until):
87			"""Filter out all issues after `until` date."""
88			if until:
89			until_date = self.str_to_date(until)
90			for issue in issues[:]:
91			close_date = self.str_to_date(issue['closed_at'])
92			if close_date > until_date and issue in issues:
93			issues.remove(issue)
94			return issues
95
96			def _filter_by_branch(self, issues, issue, branch):
97			""""""
98			number = issue['number']
99
100			if not self.is_merged(number) and issue in issues:
101			issues.remove(issue)
102
103			if branch:
104			# Get PR info and get base branch
105			pr_data = self.pr(number)
106			base_ref = pr_data['base']['ref']
107
108			if base_ref != branch and issue in issues:
109			issues.remove(issue)
110
111			return issues
112
113			def _filer_closed_prs(self, issues, branch):
114			"""Filter out closed PRs."""
115			for issue in issues[:]:
116			pr = issue.get('pull_request', '')
117
118			# Add label names inside additional key
119			issue['loghub_label_names'] = [
120			l['name'] for l in issue.get('labels')
121			]
122
123			if pr:
124			issues = self._filter_by_branch(issues, issue, branch)
125
126			return issues
127
128			def tags(self):
129			"""Return all tags."""
130			self._check_rate()
131			return self.repo('git')('refs')('tags').get()
132
133			def tag(self, tag_name):
134			"""Get tag information."""
135			self._check_rate()
136			refs = self.repo('git')('refs')('tags').get()
137			sha = -1
138
139			tags = []
140			for ref in refs:
141			ref_name = 'refs/tags/{tag}'.format(tag=tag_name)
142			if 'object' in ref and ref['ref'] == ref_name:
143			sha = ref['object']['sha']
144			tags.append(ref['ref'].split('/')[-1])
145
146			if sha == -1:
147			print("LOGHUB: You didn't pass a valid tag name!")
148			print('LOGHUB: The available tags are: {0}\n'.format(tags))
149			sys.exit(1)
150
151			return self.repo('git')('tags')(sha).get()
152
153			def milestones(self):
154			"""Return all milestones."""
155			self._check_rate()
156			return self.repo.milestones.get(state='all')
157
158			def milestone(self, milestone_title):
159			"""Return milestone with given title."""
160			self._check_rate()
161			milestones = self.milestones()
162			milestone_number = -1
163
164			milestone_titles = []
165			for milestone in milestones:
166			if milestone['title'] == milestone_title:
167			milestone_number = milestone['number']
168			milestone_titles.append(milestone['title'])
169
170			if milestone_number == -1:
171			print("LOGHUB: You didn't pass a valid milestone name!")
172			print('LOGHUB: The available milestones are: {0}\n'
173			''.format(milestone_titles))
174			sys.exit(1)
175
176			return milestone
177
178			def pr(self, pr_number):
179			"""Get PR information."""
180			self._check_rate()
181			return self.repo('pulls')(str(pr_number)).get()
182
183			def issues(self,
184			milestone=None,
185			state=None,
186			assignee=None,
187			creator=None,
188			mentioned=None,
189			labels=None,
190			sort=None,
191			direction=None,
192			since=None,
193			until=None,
194			branch=None):
195			"""Return Issues and Pull Requests."""
196			self._check_rate()
197			page = 1
198			issues = []
199			while True:
200			result = self.repo.issues.get(page=page,
201			per_page=100,
202			milestone=milestone,
203			state=state,
204			assignee=assignee,
205			creator=creator,
206			mentioned=mentioned,
207			labels=labels,
208			sort=sort,
209			direction=direction,
210			since=since)
211			if len(result) > 0:
212			issues += result
213			page = page + 1
214			else:
215			break
216
217			# If since was provided, filter the issue
218			issues = self._filter_since(issues, since)
219
220			# If until was provided, filter the issue
221			issues = self._filter_until(issues, until)
222
223			# If it is a pr check if it is merged or closed, removed closed ones
224			issues = self._filer_closed_prs(issues, branch)
225
226			return issues
227
228			def is_merged(self, pr):
229			"""
230			Return wether a PR was merged, or if it was closed and discarded.
231
232			https://developer.github.com/v3/pulls/#get-if-a-pull-request-has-been-merged
233			"""
234			self._check_rate()
235			merged = True
236			try:
237			self.repo('pulls')(str(pr))('merge').get()
238			except Exception:
239			merged = False
240			return merged
241
242			@staticmethod
243			def str_to_date(string):
244			"""Convert ISO date string to datetime object."""
245			parts = string.split('T')
246			date_parts = parts[0]
247			time_parts = parts[1][:-1]
248			year, month, day = [int(i) for i in date_parts.split('-')]
249			hour, minutes, seconds = [int(i) for i in time_parts.split(':')]
250			return datetime.datetime(year, month, day, hour, minutes, seconds)
251

spyder-ide / loghub

Pull Request — master (#45)

GitHubRepo._check_user() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like