豆瓣API限制次数
API限制为每分钟40次,一不小心就超了,马上KEY就被封了,等了半个多小时居然还没有解封,只能用不带KEY的API,每分钟只有可怜的10次。
马上写了一个限制类,还挺好用的
用法:
1.
client = Limit(DoubanService(server=SERVER), 60, 9)
调用client.some_services_function()的时候,如果超过了访问限制,就会自动sleep,到指定的时刻再继续干活
2.
用来限制函数也可以
urlopen = Limit(urllib2.urlopen, 60, 40)
$ cat limit.py
import time
import unittest
import logging
logger = logging.getLogger('Limit')
class Limit(object):
def __init__(self, obj_or_func, seconds=None, maxn=None):
self.obj_or_func = obj_or_func
self.seconds = seconds
self.maxn = maxn
self.hist = []
def check(self):
now = time.time()
hist = [ i for i in self.hist if i>=now-self.seconds ]
if len(hist) >= self.maxn:
t = hist[0]+self.seconds-now
logger.debug('exceed limit, go to sleep for %.1f seconds'%t)
time.sleep(t)
self.hist = hist+[time.time()]
def __getattr__(self, name):
if self.seconds and self.maxn:
self.check()
return object.__getattribute__(self.obj_or_func, name)
def __call__(self, *args, **kw):
if self.seconds and self.maxn:
self.check()
return self.obj_or_func(*args, **kw)
class LimitTest(unittest.TestCase):
def setUp(self):
self.obj = {}
self.func = lambda:0
def test_obj1(self):
l = Limit(self.obj, .1, 2)
t1 = time.time()
for i in range(1):
l.keys
self.assertTrue(time.time()-t1<.1)
def test_obj2(self):
l = Limit(self.obj, .1, 2)
t1 = time.time()
for i in range(3):
l.keys
self.assertTrue(time.time()-t1>.1)
def test_obj3(self):
l = Limit(self.obj, .1, 2)
t1 = time.time()
for i in range(5):
l.keys
self.assertTrue(time.time()-t1>.2)
def test_func(self):
l = Limit(self.func, .1, 2)
t1 = time.time()
for i in range(3):
l()
self.assertTrue(time.time()-t1>.1)
马上写了一个限制类,还挺好用的
用法:
1.
client = Limit(DoubanService(server=SERVER), 60, 9)
调用client.some_services_function()的时候,如果超过了访问限制,就会自动sleep,到指定的时刻再继续干活
2.
用来限制函数也可以
urlopen = Limit(urllib2.urlopen, 60, 40)
$ cat limit.py
import time
import unittest
import logging
logger = logging.getLogger('Limit')
class Limit(object):
def __init__(self, obj_or_func, seconds=None, maxn=None):
self.obj_or_func = obj_or_func
self.seconds = seconds
self.maxn = maxn
self.hist = []
def check(self):
now = time.time()
hist = [ i for i in self.hist if i>=now-self.seconds ]
if len(hist) >= self.maxn:
t = hist[0]+self.seconds-now
logger.debug('exceed limit, go to sleep for %.1f seconds'%t)
time.sleep(t)
self.hist = hist+[time.time()]
def __getattr__(self, name):
if self.seconds and self.maxn:
self.check()
return object.__getattribute__(self.obj_or_func, name)
def __call__(self, *args, **kw):
if self.seconds and self.maxn:
self.check()
return self.obj_or_func(*args, **kw)
class LimitTest(unittest.TestCase):
def setUp(self):
self.obj = {}
self.func = lambda:0
def test_obj1(self):
l = Limit(self.obj, .1, 2)
t1 = time.time()
for i in range(1):
l.keys
self.assertTrue(time.time()-t1<.1)
def test_obj2(self):
l = Limit(self.obj, .1, 2)
t1 = time.time()
for i in range(3):
l.keys
self.assertTrue(time.time()-t1>.1)
def test_obj3(self):
l = Limit(self.obj, .1, 2)
t1 = time.time()
for i in range(5):
l.keys
self.assertTrue(time.time()-t1>.2)
def test_func(self):
l = Limit(self.func, .1, 2)
t1 = time.time()
for i in range(3):
l()
self.assertTrue(time.time()-t1>.1)