Skip to content
Navigation Menu
{{ message }}
forked from facebookarchive/python-instagram
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbind.py
More file actions
executable file
·196 lines (163 loc) · 8.62 KB
/
Copy pathbind.py
File metadata and controls
executable file
·196 lines (163 loc) · 8.62 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
import urllib.request, urllib.parse, urllib.error
from .oauth2 import OAuth2Request
import re
from .json_import import simplejson
import hmac
from hashlib import sha256
re_path_template = re.compile('{\w+}')
def encode_string(value):
return value.encode('utf-8') \
if isinstance(value, str) else str(value)
class InstagramClientError(Exception):
def __init__(self, error_message, status_code=None):
self.status_code = status_code
self.error_message = error_message
def __str__(self):
if self.status_code:
return "(%s) %s" % (self.status_code, self.error_message)
else:
return self.error_message
class InstagramAPIError(Exception):
def __init__(self, status_code, error_type, error_message, *args, **kwargs):
self.status_code = status_code
self.error_type = error_type
self.error_message = error_message
def __str__(self):
return "(%s) %s-%s" % (self.status_code, self.error_type, self.error_message)
def bind_method(**config):
class InstagramAPIMethod(object):
path = config['path']
method = config.get('method', 'GET')
accepts_parameters = config.get("accepts_parameters", [])
signature = config.get("signature", False)
requires_target_user = config.get('requires_target_user', False)
paginates = config.get('paginates', False)
root_class = config.get('root_class', None)
response_type = config.get("response_type", "list")
include_secret = config.get("include_secret", False)
objectify_response = config.get("objectify_response", True)
def __init__(self, api, *args, **kwargs):
self.api = api
self.as_generator = kwargs.pop("as_generator", False)
if self.as_generator:
self.pagination_format = 'next_url'
else:
self.pagination_format = kwargs.pop('pagination_format', 'next_url')
self.return_json = kwargs.pop("return_json", False)
self.max_pages = kwargs.pop("max_pages", 3)
self.with_next_url = kwargs.pop("with_next_url", None)
self.parameters = {}
self._build_parameters(args, kwargs)
self._build_path()
def _build_parameters(self, args, kwargs):
# via tweepy https://github.com/joshthecoder/tweepy/
for index, value in enumerate(args):
if value is None:
continue
try:
self.parameters[self.accepts_parameters[index]] = encode_string(value)
except IndexError:
raise InstagramClientError("Too many arguments supplied")
for key, value in list(kwargs.items()):
if value is None:
continue
if key in self.parameters:
raise InstagramClientError("Parameter %s already supplied" % key)
self.parameters[key] = encode_string(value)
if 'user_id' in self.accepts_parameters and not 'user_id' in self.parameters \
and not self.requires_target_user:
self.parameters['user_id'] = 'self'
def _build_path(self):
for variable in re_path_template.findall(self.path):
name = variable.strip('{}')
try:
value = urllib.parse.quote(self.parameters[name])
except KeyError:
raise Exception('No parameter value found for path variable: %s' % name)
del self.parameters[name]
self.path = self.path.replace(variable, value)
self.path = self.path + '.%s' % self.api.format
def _build_pagination_info(self, content_obj):
"""Extract pagination information in the desired format."""
pagination = content_obj.get('pagination', {})
if self.pagination_format == 'next_url':
return pagination.get('next_url')
if self.pagination_format == 'dict':
return pagination
raise Exception('Invalid value for pagination_format: %s' % self.pagination_format)
def _do_api_request(self, url, method="GET", body=None, headers=None):
headers = headers or {}
if self.signature and self.api.client_ips != None and self.api.client_secret != None:
secret = self.api.client_secret
ips = self.api.client_ips
signature = hmac.new(secret, ips, sha256).hexdigest()
headers['X-Insta-Forwarded-For'] = '|'.join([ips, signature])
response, content = OAuth2Request(self.api).make_request(url, method=method, body=body, headers=headers)
if response['status'] == '503' or response['status'] == '429':
raise InstagramAPIError(response['status'], "Rate limited", "Your client is making too many request per second")
try:
content_obj = simplejson.loads(content)
except ValueError:
raise InstagramClientError('Unable to parse response, not valid JSON.', status_code=response['status'])
# Handle OAuthRateLimitExceeded from Instagram's Nginx which uses different format to documented api responses
if 'meta' not in content_obj:
if content_obj.get('code') == 420 or content_obj.get('code') == 429:
error_message = content_obj.get('error_message') or "Your client is making too many request per second"
raise InstagramAPIError(content_obj.get('code'), "Rate limited", error_message)
raise InstagramAPIError('code' in content_obj, 'error_type' in content_obj, 'error_message' in content_obj)
api_responses = []
status_code = content_obj['meta']['code']
self.api.x_ratelimit_remaining = response.get("x-ratelimit-remaining",None)
self.api.x_ratelimit = response.get("x-ratelimit-limit",None)
if status_code == 200:
if not self.objectify_response:
return content_obj, None
if self.response_type == 'list':
for entry in content_obj['data']:
if self.return_json:
api_responses.append(entry)
else:
obj = self.root_class.object_from_dictionary(entry)
api_responses.append(obj)
elif self.response_type == 'entry':
data = content_obj['data']
if self.return_json:
api_responses = data
else:
api_responses = self.root_class.object_from_dictionary(data)
elif self.response_type == 'empty':
pass
return api_responses, self._build_pagination_info(content_obj)
else:
raise InstagramAPIError(status_code, content_obj['meta']['error_type'], content_obj['meta']['error_message'])
def _paginator_with_url(self, url, method="GET", body=None, headers=None):
headers = headers or {}
pages_read = 0
while url and pages_read < self.max_pages:
api_responses, url = self._do_api_request(url, method, body, headers)
pages_read += 1
yield api_responses, url
return
def _get_with_next_url(self, url, method="GET", body=None, headers=None):
headers = headers or {}
content, next = self._do_api_request(url, method, body, headers)
return content, next
def execute(self):
url, method, body, headers = OAuth2Request(self.api).prepare_request(self.method,
self.path,
self.parameters,
include_secret=self.include_secret)
if self.with_next_url:
return self._get_with_next_url(self.with_next_url, method, body, headers)
if self.as_generator:
return self._paginator_with_url(url, method, body, headers)
else:
content, next = self._do_api_request(url, method, body, headers)
if self.paginates:
return content, next
else:
return content
def _call(api, *args, **kwargs):
method = InstagramAPIMethod(api, *args, **kwargs)
return method.execute()
return _call
You can’t perform that action at this time.