forked from ask/python-github2
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrequest.py
More file actions
executable file
·132 lines (116 loc) · 4.74 KB
/
request.py
File metadata and controls
executable file
·132 lines (116 loc) · 4.74 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
import sys, time, datetime, math
import httplib
try:
import json as simplejson # For Python 2.6
except ImportError:
import simplejson
from urlparse import urlparse, urlunparse
try:
from urlparse import parse_qs
except ImportError:
from cgi import parse_qs
from urllib import urlencode
GITHUB_URL = "https://api.github.com"
#URL_PREFIX = "https://github.com/api/v2/json"
URL_PREFIX = "api.github.com/"
class GithubError(Exception):
"""An error occured when making a request to the Github API."""
class GithubRequest(object):
github_url = GITHUB_URL
#url_format = "%(github_url)s/api/%(api_version)s/%(api_format)s"
url_format = "%(github_url)s"
api_version = "v2"
api_format = "json"
GithubError = GithubError
connector_for_scheme = {
"http": httplib.HTTPConnection,
"https": httplib.HTTPSConnection,
}
def __init__(self, username=None, api_token=None, url_prefix=None,
debug=False, requests_per_second=None):
"""
Make an API request.
"""
self.username = username
self.api_token = api_token
self.url_prefix = url_prefix
self.debug = debug
if requests_per_second is None:
self.delay = 0
else:
self.delay = 1.0 / requests_per_second
self.last_request = datetime.datetime(1900,1,1)
if not self.url_prefix:
self.url_prefix = self.url_format % {
"github_url": self.github_url,
"api_version": self.api_version,
"api_format": self.api_format,
}
def encode_authentication_data(self, extra_post_data):
if self.username and self.api_token:
post_data = {"login": self.username,
"token": self.api_token}
else:
post_data = {}
post_data.update(extra_post_data)
return urlencode(post_data)
def get(self, *path_components,**extra_get_data):
path_components = filter(None, path_components)
return self.make_request("/".join(path_components),extra_post_data=extra_get_data)
def post(self, *path_components, **extra_post_data):
path_components = filter(None, path_components)
return self.make_request("/".join(path_components), extra_post_data,
method="POST")
def make_request(self, path, extra_post_data=None, method="GET"):
if self.delay:
since_last = (datetime.datetime.now() - self.last_request)
since_last_in_seconds = since_last.days * 24 * 60 * 60 + since_last.seconds
if since_last_in_seconds < self.delay:
duration = self.delay - since_last_in_seconds
if self.debug:
sys.stderr.write("delaying API call %s\n" % duration)
time.sleep(duration)
extra_post_data = extra_post_data or {}
url = "/".join([self.url_prefix, path])
result = self.raw_request(url, extra_post_data, method=method)
if self.delay:
self.last_request = datetime.datetime.now()
return result
def raw_request(self, url, extra_post_data, method="GET"):
resource = urlparse(url)
scheme, netloc, path, params, query, fragment = urlparse(url)
hostname = netloc.split(':')[0]
post_data = None
headers = self.http_headers
headers["Accept"] = "application/json"
method = method.upper()
if method == "POST":
post_data = self.encode_authentication_data(extra_post_data)
headers["Content-Length"] = str(len(post_data))
else:
query_dict= parse_qs(query)
query_dict.update(extra_post_data)
path = urlunparse((scheme, netloc, path, params,
self.encode_authentication_data(query_dict),
fragment))
connector = self.connector_for_scheme[scheme]
connection = connector(hostname)
connection.request(method, path, post_data, headers)
response = connection.getresponse()
response_text = response.read()
if self.debug:
sys.stderr.write("URL:[%s] POST_DATA:%s RESPONSE_TEXT: [%s]\n" % (
path, post_data, response_text))
if response.status >= 400:
raise RuntimeError("unexpected response from github.com %d: %r" % (
response.status, response_text))
json = simplejson.loads(response_text)
if type(json) == list:
return json
if json.get("error"):
raise self.GithubError(json["error"][0]["error"])
return json
@property
def http_headers(self):
return {"User-Agent": "pygithub2 v1",
"Accept-Encoding": "application/json"}