1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
#
# author: David Medina
import requests
import json
from errors import GithubError
from handlers import users, gists
RESOURCES_PER_PAGE = 100
#TODO: refactor: loads json in request editing Response object
class GithubCore(object):
"""
Wrapper to github api requests
Methods: get, head, post, patch, put, delete
"""
requests_remaining = None
base_url = 'https://api.github.com/'
def __init__(self):
"""
Init `requests.session`
Init JSON parser
"""
self.session = requests.session()
self.session.params = {'per_page': RESOURCES_PER_PAGE}
self._parser = json
def get(self, request, paginate=False, **kwargs):
"""
GET request
:param paginate: Boolean to return link header to paginate
"""
response = self._request('GET', request, **kwargs)
content = self._parser.loads(response.content)
if paginate:
return response.headers.get('link'), content
else:
return content
def head(self, request, **kwargs):
""" HEAD request """
return self._request('HEAD', request, **kwargs)
def post(self, request, data=None, **kwargs):
"""
POST request
:param data: raw python object to send
"""
kwargs['data'] = self._parser.dumps(data)
response = self._request('POST', request, **kwargs)
assert response.status_code == 201
return self._parser.loads(response.content)
def patch(self, request, data=None, **kwargs):
"""
PATCH request
:param data: raw python object to send
"""
kwargs['data'] = self._parser.dumps(data)
response = self._request('PATCH', request, **kwargs)
assert response.status_code == 200
return self._parser.loads(response.content)
def put(self, request, **kwargs):
""" PUT request """
response = self._request('PUT', request,
headers={'Content-length': '0'}, **kwargs)
assert response.status_code == 204
return response
def delete(self, request, **kwargs):
""" DELETE request """
data = kwargs.get('data')
if data:
kwargs['data'] = self._parser.dumps(data)
response = self._request('DELETE', request, **kwargs)
assert response.status_code == 204
return response
def _parse_args(self, request_args):
"""
Arg's parser to `_request` method
Put extra request_args in params
"""
request_core = (
'params', 'data', 'headers', 'cookies', 'files', 'auth', 'tiemout',
'allow_redirects', 'proxies', 'return_response', 'config')
request_params = request_args.get('params')
extra_params = {}
for k, v in request_args.items():
if k in request_core:
continue
extra_params.update({k: v})
del request_args[k]
if request_params and getattr(request_params, 'update'):
request_args['params'].update(extra_params)
elif extra_params:
request_args['params'] = extra_params
return request_args
def _request(self, verb, request, **kwargs):
"""
Http request wrapper
:param verb: Http method
:param request : Url query request
:param kwargs: Keyword args to request
"""
request = self.base_url + request
self._parse_args(kwargs)
response = self.session.request(verb, request, **kwargs)
self.requests_remaining = response.headers.get(
'x-ratelimit-remaining', -1)
error = GithubError(response)
error.process()
return response
class Github(GithubCore):
""" Library enter """
def __init__(self, *args):
super(Github, self).__init__()
self.authenticated = False
auth = len(args)
if auth == 2: # Basic auth
self.session.auth = tuple(map(str, args))
self.authenticated = True
elif auth == 1: # Token oauth
raise NotImplementedError
elif auth > 2:
raise TypeError("user, password or token")
@property
def users(self):
if self.authenticated:
return users.AuthUser(self)
else:
return users.User(self)
@property
def gists(self):
if self.authenticated:
return gists.AuthGist(self)
else:
return gists.Gist(self)
|
