Guest User

Untitled

a guest
Oct 18th, 2018
82
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
text 1.62 KB | None | 0 0
  1. import collections
  2. import functools
  3. import time
  4.  
  5.  
  6. class RateLimiter(object):
  7.  
  8. def __init__(self, max_calls, period=1.0, callback=None):
  9. self.iplist = {}
  10. self.calls = collections.deque()
  11. self.period = period
  12. self.max_calls = max_calls
  13. self.callback = callback
  14.  
  15. def __call__(self, f):
  16.  
  17. @functools.wraps(f)
  18. def wrapped(*args, **kwargs):
  19. ip = kwargs.get("ip")
  20. self.nowip = ip
  21. with self:
  22. if not self.exceed:
  23. return f(*args, **kwargs)
  24. # raise Exception("频率过快")
  25. self.callback(ip)
  26.  
  27. return wrapped
  28.  
  29. def __enter__(self):
  30. ip = self.nowip
  31. _r = self.iplist.get(ip,None)
  32. if not _r:
  33. self.iplist[ip] = collections.deque()
  34.  
  35. if len(self.iplist[ip]) < self.max_calls:
  36. self.exceed = False
  37. else:
  38. self.exceed = True
  39.  
  40. def __exit__(self, exc_type, exc_val, exc_tb):
  41. ip = self.nowip
  42. self.iplist[ip].append(time.time())
  43. while self._timespan >= self.period:
  44. self.iplist[ip].popleft()
  45.  
  46. @property
  47. def _timespan(self):
  48. ip = self.nowip
  49. return self.iplist[ip][-1] - self.iplist[ip][0]
  50.  
  51. def over_speed_waring(ip):
  52. print(ip,"速度过快")
  53.  
  54. @RateLimiter(3,1.0,over_speed_waring)
  55. def func_testcall(ip):
  56. print('run,ip is',ip)
  57. # time.sleep(1)
  58.  
  59.  
  60. func_testcall(ip=1)
  61. func_testcall(ip=2)
  62. func_testcall(ip=2)
  63. func_testcall(ip=2)
  64. func_testcall(ip=1)
  65. func_testcall(ip=2)
  66. # func_testcall(ip=1)
  67. func_testcall(ip=2)
  68. # func_testcall(ip=1)
  69. func_testcall(ip=2)
Add Comment
Please, Sign In to add comment