forked from bear/python-twitter
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathratelimit.py
More file actions
215 lines (173 loc) · 8.26 KB
/
ratelimit.py
File metadata and controls
215 lines (173 loc) · 8.26 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
from collections import namedtuple
import re
try:
from urllib.parse import urlparse
except ImportError:
from urlparse import urlparse
from twitter.twitter_utils import enf_type
EndpointRateLimit = namedtuple('EndpointRateLimit',
['limit', 'remaining', 'reset'])
ResourceEndpoint = namedtuple('ResourceEndpoint', ['regex', 'resource'])
GEO_ID_PLACE_ID = ResourceEndpoint(re.compile(r'/geo/id/\d+'), "/geo/id/:place_id")
SAVED_SEARCHES_DESTROY_ID = ResourceEndpoint(re.compile(r'/saved_searches/destroy/\d+'), "/saved_searches/destroy/:id")
SAVED_SEARCHES_SHOW_ID = ResourceEndpoint(re.compile(r'/saved_searches/show/\d+'), "/saved_searches/show/:id")
STATUSES_RETWEETS_ID = ResourceEndpoint(re.compile(r'/statuses/retweets/\d+'), "/statuses/retweets/:id")
STATUSES_SHOW_ID = ResourceEndpoint(re.compile(r'/statuses/show'), "/statuses/show/:id")
USERS_SHOW_ID = ResourceEndpoint(re.compile(r'/users/show'), "/users/show/:id")
USERS_SUGGESTIONS_SLUG = ResourceEndpoint(re.compile(r'/users/suggestions/\w+$'), "/users/suggestions/:slug")
USERS_SUGGESTIONS_SLUG_MEMBERS = ResourceEndpoint(re.compile(r'/users/suggestions/.+/members'), "/users/suggestions/:slug/members")
NON_STANDARD_ENDPOINTS = [
GEO_ID_PLACE_ID,
SAVED_SEARCHES_DESTROY_ID,
SAVED_SEARCHES_SHOW_ID,
STATUSES_RETWEETS_ID,
STATUSES_SHOW_ID,
USERS_SHOW_ID,
USERS_SUGGESTIONS_SLUG,
USERS_SUGGESTIONS_SLUG_MEMBERS,
]
class RateLimit(object):
""" Object to hold the rate limit status of various endpoints for
the twitter.Api object.
This object is generally attached to the API as Api.rate_limit, but is not
created until the user makes a method call that uses _RequestUrl() or calls
Api.InitializeRateLimit(), after which it get created and populated with
rate limit data from Twitter.
Calling Api.InitializeRateLimit() populates the object with all of the
rate limits for the endpoints defined by Twitter; more info is available
here:
https://dev.twitter.com/rest/public/rate-limits
https://dev.twitter.com/rest/public/rate-limiting
https://dev.twitter.com/rest/reference/get/application/rate_limit_status
Once a resource (i.e., an endpoint) has been requested, Twitter's response
will contain the current rate limit status as part of the headers, i.e.::
x-rate-limit-limit
x-rate-limit-remaining
x-rate-limit-reset
``limit`` is the generic limit for that endpoint, ``remaining`` is how many
more times you can make a call to that endpoint, and ``reset`` is the time
(in seconds since the epoch) until remaining resets to its default for that
endpoint.
Generally speaking, each endpoint has a 15-minute reset time and endpoints
can either make 180 or 15 requests per window. According to Twitter, any
endpoint not defined in the rate limit chart or the response from a GET
request to ``application/rate_limit_status.json`` should be assumed to be
15 requests per 15 minutes.
"""
def __init__(self, **kwargs):
""" Instantiates the RateLimitObject. Takes a json dict as
kwargs and maps to the object's dictionary. So for something like:
{"resources": {
"help": {
/help/privacy": {
"limit": 15,
"remaining": 15,
"reset": 1452254278
}
}
}
}
the RateLimit object will have an attribute 'resources' from which you
can perform a lookup like:
api.rate_limit.get('help').get('/help/privacy')
and a dictionary of limit, remaining, and reset will be returned.
"""
self.__dict__.update(kwargs)
@staticmethod
def url_to_resource(url):
""" Take a fully qualified URL and attempts to return the rate limit
resource family corresponding to it. For example:
>>> RateLimit.url_to_resource('https://api.twitter.com/1.1/statuses/lookup.json?id=317')
>>> '/statuses/lookup'
Args:
url (str): URL to convert to a resource family.
Returns:
string: Resource family corresponding to the URL.
"""
resource = urlparse(url).path.replace('/1.1', '').replace('.json', '')
for non_std_endpoint in NON_STANDARD_ENDPOINTS:
if re.match(non_std_endpoint.regex, resource):
return non_std_endpoint.resource
else:
return resource
def set_unknown_limit(self, url, limit, remaining, reset):
""" If a resource family is unknown, add it to the object's
dictionary. This is to deal with new endpoints being added to
the API, but not necessarily to the information returned by
``/account/rate_limit_status.json`` endpoint.
For example, if Twitter were to add an endpoint
``/puppies/lookup.json``, the RateLimit object would create a resource
family ``puppies`` and add ``/puppies/lookup`` as the endpoint, along
with whatever limit, remaining hits available, and reset time would be
applicable to that resource+endpoint pair.
Args:
url (str):
URL of the endpoint being fetched.
limit (int):
Max number of times a user or app can hit the endpoint
before being rate limited.
remaining (int):
Number of times a user or app can access the endpoint
before being rate limited.
reset (int):
Epoch time at which the rate limit window will reset.
"""
endpoint = self.url_to_resource(url)
resource_family = endpoint.split('/')[1]
self.__dict__['resources'].update(
{resource_family: {
endpoint: {
"limit": limit,
"remaining": remaining,
"reset": reset
}}})
def get_limit(self, url):
""" Gets a EndpointRateLimit object for the given url.
Args:
url (str, optional):
URL of the endpoint for which to return the rate limit
status.
Returns:
namedtuple: EndpointRateLimit object containing rate limit
information.
"""
endpoint = self.url_to_resource(url)
resource_family = endpoint.split('/')[1]
try:
family_rates = self.resources.get(resource_family).get(endpoint)
except AttributeError:
return EndpointRateLimit(limit=15, remaining=15, reset=0)
if not family_rates:
self.set_unknown_limit(url, limit=15, remaining=15, reset=0)
return EndpointRateLimit(limit=15, remaining=15, reset=0)
return EndpointRateLimit(family_rates['limit'],
family_rates['remaining'],
family_rates['reset'])
def set_limit(self, url, limit, remaining, reset):
""" Set an endpoint's rate limits. The data used for each of the
args should come from Twitter's ``x-rate-limit`` headers.
Args:
url (str):
URL of the endpoint being fetched.
limit (int):
Max number of times a user or app can hit the endpoint
before being rate limited.
remaining (int):
Number of times a user or app can access the endpoint
before being rate limited.
reset (int):
Epoch time at which the rate limit window will reset.
"""
endpoint = self.url_to_resource(url)
resource_family = endpoint.split('/')[1]
try:
family_rates = self.resources.get(resource_family).get(endpoint)
except AttributeError:
self.set_unknown_limit(url, limit, remaining, reset)
family_rates = self.resources.get(resource_family).get(endpoint)
family_rates['limit'] = enf_type('limit', int, limit)
family_rates['remaining'] = enf_type('remaining', int, remaining)
family_rates['reset'] = enf_type('reset', int, reset)
return EndpointRateLimit(family_rates['limit'],
family_rates['remaining'],
family_rates['reset'])