Skip to content
Navigation Menu
{{ message }}
forked from sigmavirus24/github3.py
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmodels.py
More file actions
338 lines (269 loc) · 10.6 KB
/
Copy pathmodels.py
File metadata and controls
338 lines (269 loc) · 10.6 KB
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
# -*- coding: utf-8 -*-
"""This module provides the basic models used in github3.py."""
from __future__ import unicode_literals
import json as jsonlib
import logging
import dateutil.parser
import requests
import requests.compat
from . import exceptions
from . import session
LOG = logging.getLogger(__package__)
class GitHubCore(object):
"""The base object for all objects that require a session.
The :class:`GitHubCore <GitHubCore>` object provides some
basic attributes and methods to other sub-classes that are very useful to
have.
"""
_ratelimit_resource = "core"
_refresh_to = None
def __init__(self, json, session):
"""Initialize our basic object.
Pretty much every object will pass in decoded JSON and a Session.
"""
# Either or 'session' is an instance of a GitHubCore sub-class or it
# is a session. In the former case it will have a 'session' attribute.
# If it doesn't, we can just default to using the passed in session.
self.session = getattr(session, "session", session)
# set a sane default
self._github_url = "https://api.github.com"
if json is not None:
self.etag = json.pop("ETag", None)
self.last_modified = json.pop("Last-Modified", None)
self._uniq = json.get("url", None)
self._json_data = json
try:
self._update_attributes(json)
except KeyError as kerr:
raise exceptions.IncompleteResponse(json, kerr)
def _update_attributes(self, json):
pass
def __getattr__(self, attribute):
"""Proxy access to stored JSON."""
if attribute not in self._json_data:
raise AttributeError(attribute)
value = self._json_data.get(attribute)
setattr(self, attribute, value)
return value
def as_dict(self):
"""Return the attributes for this object as a dictionary.
This is equivalent to calling::
json.loads(obj.as_json())
:returns: this object's attributes serialized to a dictionary
:rtype: dict
"""
return self._json_data
def as_json(self):
"""Return the json data for this object.
This is equivalent to calling::
json.dumps(obj.as_dict())
:returns: this object's attributes as a JSON string
:rtype: str
"""
return jsonlib.dumps(self._json_data)
@classmethod
def _strptime(cls, time_str):
"""Convert an ISO 8601 formatted string to a datetime object.
We assume that the ISO 8601 formatted string is in UTC and we create
the datetime object so that it is timezone-aware.
:param str time_str: ISO 8601 formatted string
:returns: timezone-aware datetime object
:rtype: datetime or None
"""
if time_str:
return dateutil.parser.parse(time_str)
return None
def __repr__(self):
repr_string = self._repr()
if requests.compat.is_py2:
return repr_string.encode("utf-8")
return repr_string
@classmethod
def from_dict(cls, json_dict, session):
"""Return an instance of this class formed from ``json_dict``."""
return cls(json_dict, session)
@classmethod
def from_json(cls, json, session):
"""Return an instance of this class formed from ``json``."""
return cls(jsonlib.loads(json), session)
def __eq__(self, other):
return self._uniq == other._uniq
def __ne__(self, other):
return self._uniq != other._uniq
def __hash__(self):
return hash(self._uniq)
def _repr(self):
return "<github3-core at 0x{0:x}>".format(id(self))
@staticmethod
def _remove_none(data):
if not data:
return
for (k, v) in list(data.items()):
if v is None:
del (data[k])
def _instance_or_null(self, instance_class, json):
if json is not None and not isinstance(json, dict):
raise exceptions.UnprocessableResponseBody(
"GitHub's API returned a body that could not be handled", json
)
if not json:
return None
return instance_class(json, self)
def _json(self, response, expected_status_code, include_cache_info=True):
ret = None
if response is None:
return None
actual_status_code = response.status_code
if actual_status_code != expected_status_code:
if actual_status_code >= 400:
raise exceptions.error_for(response)
if actual_status_code == 304:
# Received a response from someone passing in `etag=`
return None
LOG.warning(
"Expected status_code %d but got %d",
expected_status_code,
actual_status_code,
)
try:
ret = response.json()
except ValueError:
raise exceptions.UnexpectedResponse(response)
headers = response.headers
if (
include_cache_info
and (headers.get("Last-Modified") or headers.get("ETag"))
and isinstance(ret, dict)
):
ret["Last-Modified"] = response.headers.get("Last-Modified", "")
ret["ETag"] = response.headers.get("ETag", "")
LOG.info("JSON was %sreturned", "not " if ret is None else "")
return ret
def _boolean(self, response, true_code, false_code):
if response is not None:
status_code = response.status_code
if status_code == true_code:
return True
if status_code != false_code and status_code >= 400:
raise exceptions.error_for(response)
return False
def _request(self, method, *args, **kwargs):
try:
request_method = getattr(self.session, method)
return request_method(*args, **kwargs)
except (
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
) as exc:
raise exceptions.ConnectionError(exc)
except requests.exceptions.RequestException as exc:
raise exceptions.TransportError(exc)
def _delete(self, url, **kwargs):
LOG.debug("DELETE %s with %s", url, kwargs)
return self._request("delete", url, **kwargs)
def _get(self, url, **kwargs):
LOG.debug("GET %s with %s", url, kwargs)
return self._request("get", url, **kwargs)
def _patch(self, url, **kwargs):
LOG.debug("PATCH %s with %s", url, kwargs)
return self._request("patch", url, **kwargs)
def _post(self, url, data=None, json=True, **kwargs):
if json:
data = jsonlib.dumps(data) if data is not None else data
LOG.debug("POST %s with %s, %s", url, data, kwargs)
return self._request("post", url, data, **kwargs)
def _put(self, url, **kwargs):
LOG.debug("PUT %s with %s", url, kwargs)
return self._request("put", url, **kwargs)
def _build_url(self, *args, **kwargs):
"""Build a new API url from scratch."""
return self.session.build_url(*args, **kwargs)
@property
def _api(self):
value = "{0.scheme}://{0.netloc}{0.path}".format(self._uri)
if self._uri.query:
value += "?{}".format(self._uri.query)
return value
@staticmethod
def _uri_parse(uri):
return requests.compat.urlparse(uri)
@_api.setter
def _api(self, uri):
if uri:
self._uri = self._uri_parse(uri)
self.url = uri
def _iter(
self,
count,
url,
cls,
params=None,
etag=None,
headers=None,
list_key=None,
):
"""Generic iterator for this project.
:param int count: How many items to return.
:param int url: First URL to start with
:param class cls: cls to return an object of
:param params dict: (optional) Parameters for the request
:param str etag: (optional), ETag from the last call
:param dict headers: (optional) HTTP Headers for the request
:param str list_key: (optional) Key for extracting the list of items
from a dict response
:returns: A lazy iterator over the pagianted resource
:rtype: :class:`GitHubIterator <github3.structs.GitHubIterator>`
"""
from .structs import GitHubIterator
return GitHubIterator(
count, url, cls, self, params, etag, headers, list_key
)
@property
def ratelimit_remaining(self):
"""Number of requests before GitHub imposes a ratelimit.
:returns: int
"""
json = self._json(self._get(self._build_url("rate_limit")), 200)
core = json.get("resources", {}).get(self._ratelimit_resource, {})
self._remaining = core.get("remaining", 0)
return self._remaining
def refresh(self, conditional=False):
"""Re-retrieve the information for this object.
The reasoning for the return value is the following example: ::
repos = [r.refresh() for r in g.repositories_by('kennethreitz')]
Without the return value, that would be an array of ``None``'s and you
would otherwise have to do: ::
repos = [r for i in g.repositories_by('kennethreitz')]
[r.refresh() for r in repos]
Which is really an anti-pattern.
.. versionchanged:: 0.5
.. _Conditional Requests:
http://developer.github.com/v3/#conditional-requests
:param bool conditional: If True, then we will search for a stored
header ('Last-Modified', or 'ETag') on the object and send that
as described in the `Conditional Requests`_ section of the docs
:returns: self
"""
headers = getattr(self, "CUSTOM_HEADERS", {})
if conditional:
if self.last_modified:
headers["If-Modified-Since"] = self.last_modified
elif self.etag:
headers["If-None-Match"] = self.etag
headers = headers or None
json = self._json(self._get(self._api, headers=headers), 200)
if json is not None:
if self._refresh_to is None:
self._json_data = json
self._update_attributes(json)
else:
return self._refresh_to(json, self)
return self
def new_session(self):
"""Generate a new session.
:returns:
A brand new session
:rtype:
:class:`~github3.session.GitHubSession`
"""
return session.GitHubSession()
You can’t perform that action at this time.