_Callable - Code Metrics - Inspection of "Merge pull request #38 from goanpeca/fix/checkigno..." - spyder-ide/loghub - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (702f68)

by Gonzalo

created 2016-12-16 00:06 UTC

_Callable A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	29
Duplicated Lines	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
c	2
b	0
f	0
dl	0
loc	29
rs	10
wmc	11

#!/usr/bin/env python
# -*-coding: utf8 -*-
'''
GitHub API Python SDK. (Python >= 2.6)

Apache License

Michael Liao ([email protected])

Usage:

>>> gh = GitHub(username='githubpy', password='test-githubpy-1234')
>>> L = gh.users('githubpy').followers.get()
>>> L[0].id
470058
>>> L[0].login == u'michaelliao'
True
>>> x_ratelimit_remaining = gh.x_ratelimit_remaining
>>> x_ratelimit_limit = gh.x_ratelimit_limit
>>> x_ratelimit_reset = gh.x_ratelimit_reset
>>> L = gh.users('githubpy').following.get()
>>> L[0].url == u'https://api.github.com/users/michaelliao'
True
>>> L = gh.repos('githubpy')('testgithubpy').issues.get(state='closed', sort='created')
>>> L[0].title == u'sample issue for test'
True
>>> L[0].number
1
>>> I = gh.repos('githubpy')('testgithubpy').issues(1).get()
>>> I.url == u'https://api.github.com/repos/githubpy/testgithubpy/issues/1'
True
>>> gh = GitHub(username='githubpy', password='test-githubpy-1234')
>>> r = gh.repos('githubpy')('testgithubpy').issues.post(title='test create issue', body='just a test')
>>> r.title == u'test create issue'
True
>>> r.state == u'open'
True
>>> gh.repos.thisisabadurl.get()
Traceback (most recent call last):
    ...
ApiNotFoundError: https://api.github.com/repos/thisisabadurl
>>> gh.users('github-not-exist-user').followers.get()
Traceback (most recent call last):
    ...
ApiNotFoundError: https://api.github.com/users/github-not-exist-user/followers
'''
# yapf: disable
__version__ = '1.1.1'

try:
    # Python 2
    from urllib2 import build_opener, HTTPSHandler, Request, HTTPError
    from urllib import quote as urlquote
    from StringIO import StringIO

    def bytes(string, encoding=None):
        return str(string)
except:
    # Python 3
    from urllib.request import build_opener, HTTPSHandler, HTTPError, Request
    from urllib.parse import quote as urlquote
    from io import StringIO

# Standard library imports
from collections import Iterable
from datetime import datetime, timedelta, tzinfo
import base64
import hashlib
import hmac
import json
import mimetypes
import os
import re
import time
import urllib

TIMEOUT = 60

_URL = 'https://api.github.com'
_METHOD_MAP = dict(
    GET=lambda: 'GET',
    PUT=lambda: 'PUT',
    POST=lambda: 'POST',
    PATCH=lambda: 'PATCH',
    DELETE=lambda: 'DELETE')

DEFAULT_SCOPE = None
RW_SCOPE = 'user,public_repo,repo,repo:status,gist'


def _encode_params(kw):
    '''
    Encode parameters.
    '''
    args = []
    for k, v in kw.items():
        # If value is None or empty, ignore
        if v:
            try:
                # Python 2
                if not isinstance(v, [unicode, str]):
                    v = str(v)
                qv = v.encode('utf-8') if isinstance(v, unicode) else str(v)
            except:
                # Make sure all values are strings
                qv = str(v)
            args.append('%s=%s' % (k, urlquote(qv)))
    return '&'.join(args)


def _encode_json(obj):
    '''
    Encode object as json str.
    '''

    def _dump_obj(obj):
        if isinstance(obj, dict):
            return obj
        d = dict()
        for k in dir(obj):
            if not k.startswith('_'):
                d[k] = getattr(obj, k)
        return d

    return json.dumps(obj, default=_dump_obj)


def _parse_json(jsonstr):
    def _obj_hook(pairs):
        o = JsonObject()
        for k, v in pairs.items():
            o[str(k)] = v
        return o

    return json.loads(jsonstr, object_hook=_obj_hook)


class _Executable(object):
    def __init__(self, _gh, _method, _path):
        self._gh = _gh
        self._method = _method
        self._path = _path

    def __call__(self, **kw):
        return self._gh._http(self._method, self._path, **kw)

    def __str__(self):
        return '_Executable (%s %s)' % (self._method, self._path)

    __repr__ = __str__


class _Callable(object):
    def __init__(self, _gh, _name):
        self._gh = _gh
        self._name = _name

    def __call__(self, *args):
        if len(args) == 0:
            return self
        name = '%s/%s' % (self._name, '/'.join([str(arg) for arg in args]))
        return _Callable(self._gh, name)

    def __getattr__(self, attr):
        if attr == 'get':
            return _Executable(self._gh, 'GET', self._name)
        if attr == 'put':
            return _Executable(self._gh, 'PUT', self._name)
        if attr == 'post':
            return _Executable(self._gh, 'POST', self._name)
        if attr == 'patch':
            return _Executable(self._gh, 'PATCH', self._name)
        if attr == 'delete':
            return _Executable(self._gh, 'DELETE', self._name)
        name = '%s/%s' % (self._name, attr)
        return _Callable(self._gh, name)

    def __str__(self):
        return '_Callable (%s)' % self._name

    __repr__ = __str__


class GitHub(object):
    '''
    GitHub client.
    '''

    def __init__(self,
                 username=None,
                 password=None,
                 access_token=None,
                 client_id=None,
                 client_secret=None,
                 redirect_uri=None,
                 scope=None):
        self.x_ratelimit_remaining = (-1)
        self.x_ratelimit_limit = (-1)
        self.x_ratelimit_reset = (-1)
        self._authorization = None
        if username and password:
            # roundabout hack for Python 3
            userandpass = base64.b64encode(
                bytes('%s:%s' % (username, password), 'utf-8'))
            userandpass = userandpass.decode('ascii')
            self._authorization = 'Basic %s' % userandpass
        elif access_token:
            self._authorization = 'token %s' % access_token
        self._client_id = client_id
        self._client_secret = client_secret
        self._redirect_uri = redirect_uri
        self._scope = scope

    def authorize_url(self, state=None):
        '''
        Generate authorize_url.

        >>> GitHub(client_id='3ebf94c5776d565bcf75').authorize_url()
        'https://github.com/login/oauth/authorize?client_id=3ebf94c5776d565bcf75'
        '''
        if not self._client_id:
            raise ApiAuthError('No client id.')
        kw = dict(client_id=self._client_id)
        if self._redirect_uri:
            kw['redirect_uri'] = self._redirect_uri
        if self._scope:
            kw['scope'] = self._scope
        if state:
            kw['state'] = state
        return 'https://github.com/login/oauth/authorize?%s' % _encode_params(
            kw)

    def get_access_token(self, code, state=None):
        '''
        In callback url: http://host/callback?code=123&state=xyz

        use code and state to get an access token.        
        '''
        kw = dict(
            client_id=self._client_id,
            client_secret=self._client_secret,
            code=code)
        if self._redirect_uri:
            kw['redirect_uri'] = self._redirect_uri
        if state:
            kw['state'] = state
        opener = build_opener(HTTPSHandler)
        request = Request(
            'https://github.com/login/oauth/access_token',
            data=_encode_params(kw))
        request.get_method = _METHOD_MAP['POST']
        request.add_header('Accept', 'application/json')
        try:
            response = opener.open(request, timeout=TIMEOUT)
            r = _parse_json(response.read())
            if 'error' in r:
                raise ApiAuthError(str(r.error))
            return str(r.access_token)
        except HTTPError as e:
            raise ApiAuthError('HTTPError when get access token')

    def __getattr__(self, attr):
        return _Callable(self, '/%s' % attr)

    def _http(self, _method, _path, **kw):
        data = None
        params = None
        if _method == 'GET' and kw:
            _path = '%s?%s' % (_path, _encode_params(kw))
        if _method in ['POST', 'PATCH', 'PUT']:
            data = bytes(_encode_json(kw), 'utf-8')
        url = '%s%s' % (_URL, _path)
        opener = build_opener(HTTPSHandler)
        request = Request(url, data=data)
        request.get_method = _METHOD_MAP[_method]
        if self._authorization:
            request.add_header('Authorization', self._authorization)
        if _method in ['POST', 'PATCH', 'PUT']:
            request.add_header('Content-Type',
                               'application/x-www-form-urlencoded')
        try:
            response = opener.open(request, timeout=TIMEOUT)
            is_json = self._process_resp(response.headers)
            if is_json:
                return _parse_json(response.read().decode('utf-8'))
        except HTTPError as e:
            is_json = self._process_resp(e.headers)
            if is_json:
                json = _parse_json(e.read().decode('utf-8'))
            else:
                json = e.read().decode('utf-8')
            req = JsonObject(method=_method, url=url)
            resp = JsonObject(code=e.code, json=json)
            if resp.code == 404:
                raise ApiNotFoundError(url, req, resp)
            raise ApiError(url, req, resp)

    def _process_resp(self, headers):
        is_json = False
        if headers:
            for k in headers:
                h = k.lower()
                if h == 'x-ratelimit-remaining':
                    self.x_ratelimit_remaining = int(headers[k])
                elif h == 'x-ratelimit-limit':
                    self.x_ratelimit_limit = int(headers[k])
                elif h == 'x-ratelimit-reset':
                    self.x_ratelimit_reset = int(headers[k])
                elif h == 'content-type':
                    is_json = headers[k].startswith('application/json')
        return is_json


class JsonObject(dict):
    '''
    general json object that can bind any fields but also act as a dict.
    '''

    def __getattr__(self, key):
        try:
            return self[key]
        except KeyError:
            raise AttributeError(r"'Dict' object has no attribute '%s'" % key)

    def __setattr__(self, attr, value):
        self[attr] = value


class ApiError(Exception):
    def __init__(self, url, request, response):
        super(ApiError, self).__init__(url)
        self.request = request
        self.response = response


class ApiAuthError(ApiError):
    def __init__(self, msg):
        super(ApiAuthError, self).__init__(msg, None, None)


class ApiNotFoundError(ApiError):
    pass


if __name__ == '__main__':
    import doctest
    doctest.testmod()


1			#!/usr/bin/env python
2			# --coding: utf8 --
3			'''
4			GitHub API Python SDK. (Python >= 2.6)
5
6			Apache License
7
8			Michael Liao ([email protected])
9
10			Usage:
11
12			>>> gh = GitHub(username='githubpy', password='test-githubpy-1234')
13			>>> L = gh.users('githubpy').followers.get()
14			>>> L[0].id
15			470058
16			>>> L[0].login == u'michaelliao'
17			True
18			>>> x_ratelimit_remaining = gh.x_ratelimit_remaining
19			>>> x_ratelimit_limit = gh.x_ratelimit_limit
20			>>> x_ratelimit_reset = gh.x_ratelimit_reset
21			>>> L = gh.users('githubpy').following.get()
22			>>> L[0].url == u'https://api.github.com/users/michaelliao'
23			True
24			>>> L = gh.repos('githubpy')('testgithubpy').issues.get(state='closed', sort='created')
25			>>> L[0].title == u'sample issue for test'
26			True
27			>>> L[0].number
28			1
29			>>> I = gh.repos('githubpy')('testgithubpy').issues(1).get()
30			>>> I.url == u'https://api.github.com/repos/githubpy/testgithubpy/issues/1'
31			True
32			>>> gh = GitHub(username='githubpy', password='test-githubpy-1234')
33			>>> r = gh.repos('githubpy')('testgithubpy').issues.post(title='test create issue', body='just a test')
34			>>> r.title == u'test create issue'
35			True
36			>>> r.state == u'open'
37			True
38			>>> gh.repos.thisisabadurl.get()
39			Traceback (most recent call last):
40			...
41			ApiNotFoundError: https://api.github.com/repos/thisisabadurl
42			>>> gh.users('github-not-exist-user').followers.get()
43			Traceback (most recent call last):
44			...
45			ApiNotFoundError: https://api.github.com/users/github-not-exist-user/followers
46			'''
47			# yapf: disable
48			__version__ = '1.1.1'
49
50			try:
51			# Python 2
52			from urllib2 import build_opener, HTTPSHandler, Request, HTTPError
53			from urllib import quote as urlquote
54			from StringIO import StringIO
55
56			def bytes(string, encoding=None):
57			return str(string)
58			except:
59			# Python 3
60			from urllib.request import build_opener, HTTPSHandler, HTTPError, Request
61			from urllib.parse import quote as urlquote
62			from io import StringIO
63
64			# Standard library imports
65			from collections import Iterable
66			from datetime import datetime, timedelta, tzinfo
67			import base64
68			import hashlib
69			import hmac
70			import json
71			import mimetypes
72			import os
73			import re
74			import time
75			import urllib
76
77			TIMEOUT = 60
78
79			_URL = 'https://api.github.com'
80			_METHOD_MAP = dict(
81			GET=lambda: 'GET',
82			PUT=lambda: 'PUT',
83			POST=lambda: 'POST',
84			PATCH=lambda: 'PATCH',
85			DELETE=lambda: 'DELETE')
86
87			DEFAULT_SCOPE = None
88			RW_SCOPE = 'user,public_repo,repo,repo:status,gist'
89
90
91			def _encode_params(kw):
92			'''
93			Encode parameters.
94			'''
95			args = []
96			for k, v in kw.items():
97			# If value is None or empty, ignore
98			if v:
99			try:
100			# Python 2
101			if not isinstance(v, [unicode, str]):
102			v = str(v)
103			qv = v.encode('utf-8') if isinstance(v, unicode) else str(v)
104			except:
105			# Make sure all values are strings
106			qv = str(v)
107			args.append('%s=%s' % (k, urlquote(qv)))
108			return '&'.join(args)
109
110
111			def _encode_json(obj):
112			'''
113			Encode object as json str.
114			'''
115
116			def _dump_obj(obj):
117			if isinstance(obj, dict):
118			return obj
119			d = dict()
120			for k in dir(obj):
121			if not k.startswith('_'):
122			d[k] = getattr(obj, k)
123			return d
124
125			return json.dumps(obj, default=_dump_obj)
126
127
128			def _parse_json(jsonstr):
129			def _obj_hook(pairs):
130			o = JsonObject()
131			for k, v in pairs.items():
132			o[str(k)] = v
133			return o
134
135			return json.loads(jsonstr, object_hook=_obj_hook)
136
137
138			class _Executable(object):
139			def __init__(self, _gh, _method, _path):
140			self._gh = _gh
141			self._method = _method
142			self._path = _path
143
144			def __call__(self, **kw):
145			return self._gh._http(self._method, self._path, **kw)
146
147			def __str__(self):
148			return '_Executable (%s %s)' % (self._method, self._path)
149
150			__repr__ = __str__
151
152
153			class _Callable(object):
154			def __init__(self, _gh, _name):
155			self._gh = _gh
156			self._name = _name
157
158			def __call__(self, *args):
159			if len(args) == 0:
160			return self
161			name = '%s/%s' % (self._name, '/'.join([str(arg) for arg in args]))
162			return _Callable(self._gh, name)
163
164			def __getattr__(self, attr):
165			if attr == 'get':
166			return _Executable(self._gh, 'GET', self._name)
167			if attr == 'put':
168			return _Executable(self._gh, 'PUT', self._name)
169			if attr == 'post':
170			return _Executable(self._gh, 'POST', self._name)
171			if attr == 'patch':
172			return _Executable(self._gh, 'PATCH', self._name)
173			if attr == 'delete':
174			return _Executable(self._gh, 'DELETE', self._name)
175			name = '%s/%s' % (self._name, attr)
176			return _Callable(self._gh, name)
177
178			def __str__(self):
179			return '_Callable (%s)' % self._name
180
181			__repr__ = __str__
182
183
184			class GitHub(object):
185			'''
186			GitHub client.
187			'''
188
189			def __init__(self,
190			username=None,
191			password=None,
192			access_token=None,
193			client_id=None,
194			client_secret=None,
195			redirect_uri=None,
196			scope=None):
197			self.x_ratelimit_remaining = (-1)
198			self.x_ratelimit_limit = (-1)
199			self.x_ratelimit_reset = (-1)
200			self._authorization = None
201			if username and password:
202			# roundabout hack for Python 3
203			userandpass = base64.b64encode(
204			bytes('%s:%s' % (username, password), 'utf-8'))
205			userandpass = userandpass.decode('ascii')
206			self._authorization = 'Basic %s' % userandpass
207			elif access_token:
208			self._authorization = 'token %s' % access_token
209			self._client_id = client_id
210			self._client_secret = client_secret
211			self._redirect_uri = redirect_uri
212			self._scope = scope
213
214			def authorize_url(self, state=None):
215			'''
216			Generate authorize_url.
217
218			>>> GitHub(client_id='3ebf94c5776d565bcf75').authorize_url()
219			'https://github.com/login/oauth/authorize?client_id=3ebf94c5776d565bcf75'
220			'''
221			if not self._client_id:
222			raise ApiAuthError('No client id.')
223			kw = dict(client_id=self._client_id)
224			if self._redirect_uri:
225			kw['redirect_uri'] = self._redirect_uri
226			if self._scope:
227			kw['scope'] = self._scope
228			if state:
229			kw['state'] = state
230			return 'https://github.com/login/oauth/authorize?%s' % _encode_params(
231			kw)
232
233			def get_access_token(self, code, state=None):
234			'''
235			In callback url: http://host/callback?code=123&state=xyz
236
237			use code and state to get an access token.
238			'''
239			kw = dict(
240			client_id=self._client_id,
241			client_secret=self._client_secret,
242			code=code)
243			if self._redirect_uri:
244			kw['redirect_uri'] = self._redirect_uri
245			if state:
246			kw['state'] = state
247			opener = build_opener(HTTPSHandler)
248			request = Request(
249			'https://github.com/login/oauth/access_token',
250			data=_encode_params(kw))
251			request.get_method = _METHOD_MAP['POST']
252			request.add_header('Accept', 'application/json')
253			try:
254			response = opener.open(request, timeout=TIMEOUT)
255			r = _parse_json(response.read())
256			if 'error' in r:
257			raise ApiAuthError(str(r.error))
258			return str(r.access_token)
259			except HTTPError as e:
260			raise ApiAuthError('HTTPError when get access token')
261
262			def __getattr__(self, attr):
263			return _Callable(self, '/%s' % attr)
264
265			def _http(self, _method, _path, **kw):
266			data = None
267			params = None
268			if _method == 'GET' and kw:
269			_path = '%s?%s' % (_path, _encode_params(kw))
270			if _method in ['POST', 'PATCH', 'PUT']:
271			data = bytes(_encode_json(kw), 'utf-8')
272			url = '%s%s' % (_URL, _path)
273			opener = build_opener(HTTPSHandler)
274			request = Request(url, data=data)
275			request.get_method = _METHOD_MAP[_method]
276			if self._authorization:
277			request.add_header('Authorization', self._authorization)
278			if _method in ['POST', 'PATCH', 'PUT']:
279			request.add_header('Content-Type',
280			'application/x-www-form-urlencoded')
281			try:
282			response = opener.open(request, timeout=TIMEOUT)
283			is_json = self._process_resp(response.headers)
284			if is_json:
285			return _parse_json(response.read().decode('utf-8'))
286			except HTTPError as e:
287			is_json = self._process_resp(e.headers)
288			if is_json:
289			json = _parse_json(e.read().decode('utf-8'))
290			else:
291			json = e.read().decode('utf-8')
292			req = JsonObject(method=_method, url=url)
293			resp = JsonObject(code=e.code, json=json)
294			if resp.code == 404:
295			raise ApiNotFoundError(url, req, resp)
296			raise ApiError(url, req, resp)
297
298			def _process_resp(self, headers):
299			is_json = False
300			if headers:
301			for k in headers:
302			h = k.lower()
303			if h == 'x-ratelimit-remaining':
304			self.x_ratelimit_remaining = int(headers[k])
305			elif h == 'x-ratelimit-limit':
306			self.x_ratelimit_limit = int(headers[k])
307			elif h == 'x-ratelimit-reset':
308			self.x_ratelimit_reset = int(headers[k])
309			elif h == 'content-type':
310			is_json = headers[k].startswith('application/json')
311			return is_json
312
313
314			class JsonObject(dict):
315			'''
316			general json object that can bind any fields but also act as a dict.
317			'''
318
319			def __getattr__(self, key):
320			try:
321			return self[key]
322			except KeyError:
323			raise AttributeError(r"'Dict' object has no attribute '%s'" % key)
324
325			def __setattr__(self, attr, value):
326			self[attr] = value
327
328
329			class ApiError(Exception):
330			def __init__(self, url, request, response):
331			super(ApiError, self).__init__(url)
332			self.request = request
333			self.response = response
334
335
336			class ApiAuthError(ApiError):
337			def __init__(self, msg):
338			super(ApiAuthError, self).__init__(msg, None, None)
339
340
341			class ApiNotFoundError(ApiError):
342			pass
343
344
345			if __name__ == '__main__':
346			import doctest
347			doctest.testmod()
348

spyder-ide / loghub

Branch — master (702f68)

_Callable A

Complexity

Size/Duplication

Importance

Duplication Side-by-Side

Filter issues like