_encode_params() - Code Metrics - Inspection of "Delete files" - spyder-ide/loghub - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 20f388...bc1964 )

by Gonzalo

created 2016-11-03 03:32 UTC

_encode_params() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	6
c	2
b	0
f	0
dl	0
loc	18
rs	8

#!/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
'''

__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

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

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

spyder-ide / loghub

Push — master ( 20f388...bc1964 )

_encode_params() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like