Python-API-Ratenbegrenzung - Wie man API-Aufrufe global begrenzt

Question

Ich versuche, die API-Aufrufe in meinem Code zu beschränken. Ich habe bereits eine nette Python-Bibliothek gefunden. ratelimiter==1.0.2.post0https://pypi.python.org/pypi/ratelimiter

Diese Bibliothek kann jedoch nur die Rate im lokalen Bereich begrenzen. d.h) in Funktion und Schleifen

# Decorator 
@RateLimiter(max_calls=10, period=1) 
def do_something(): 
    pass 


# Context Manager 
rate_limiter = RateLimiter(max_calls=10, period=1) 

for i in range(100): 
    with rate_limiter: 
     do_something() 

Weil ich mehrere Funktionen haben, die API-Anrufe zu tätigen, an verschiedenen Orten, ich die API-Aufrufe in globalen Umfang begrenzen wollen.

Angenommen, ich möchte den API-Aufruf auf einmal pro Sekunde beschränken. Und angenommen, ich habe die Funktionen x und y, in denen zwei API-Aufrufe gemacht werden.

@rate(...) 
def x(): 
    ... 

@rate(...) 
def y(): 
    ... 

Durch die Funktionen mit den limiter dekorieren, ich bin in der Lage, die Rate gegen die beiden Funktionen zu begrenzen.

Wenn ich jedoch die obigen beiden Funktionen nacheinander ausführe, verliert es die Anzahl der API-Aufrufe in global Bereich, weil sie einander nicht bewusst sind. So wird y direkt nach der Ausführung von x aufgerufen, ohne eine weitere Sekunde zu warten. Und dies wird die Einmal pro Sekunde Einschränkung verletzen.

Gibt es eine Möglichkeit oder eine Bibliothek, die ich verwenden kann, um die Rate global in Python zu begrenzen?

Answer 1

Schließlich habe ich meine eigene Throttler Klasse implementiert. Indem wir jede API-Anfrage an die request-Methode weiterleiten, können wir alle API-Anfragen verfolgen. Unter Ausnutzung der Übergabe der Funktion als request Methodenparameter speichert es auch das Ergebnis, um API-Aufrufe zu reduzieren.

class TooManyRequestsError(Exception): 
    def __str__(self): 
     return "More than 30 requests have been made in the last five seconds." 


class Throttler(object): 
    cache = {} 

    def __init__(self, max_rate, window, throttle_stop=False, cache_age=1800): 
     # Dict of max number of requests of the API rate limit for each source 
     self.max_rate = max_rate 
     # Dict of duration of the API rate limit for each source 
     self.window = window 
     # Whether to throw an error (when True) if the limit is reached, or wait until another request 
     self.throttle_stop = throttle_stop 
     # The time, in seconds, for which to cache a response 
     self.cache_age = cache_age 
     # Initialization 
     self.next_reset_at = dict() 
     self.num_requests = dict() 

     now = datetime.datetime.now() 
     for source in self.max_rate: 
      self.next_reset_at[source] = now + datetime.timedelta(seconds=self.window.get(source)) 
      self.num_requests[source] = 0 

    def request(self, source, method, do_cache=False): 
     now = datetime.datetime.now() 

     # if cache exists, no need to make api call 
     key = source + method.func_name 
     if do_cache and key in self.cache: 
      timestamp, data = self.cache.get(key) 
      logging.info('{} exists in cached @ {}'.format(key, timestamp)) 

      if (now - timestamp).seconds < self.cache_age: 
       logging.info('retrieved cache for {}'.format(key)) 
       return data 

     # <--- MAKE API CALLS ---> # 

     # reset the count if the period passed 
     if now > self.next_reset_at.get(source): 
      self.num_requests[source] = 0 
      self.next_reset_at[source] = now + datetime.timedelta(seconds=self.window.get(source)) 

     # throttle request 
     def halt(wait_time): 
      if self.throttle_stop: 
       raise TooManyRequestsError() 
      else: 
       # Wait the required time, plus a bit of extra padding time. 
       time.sleep(wait_time + 0.1) 

     # if exceed max rate, need to wait 
     if self.num_requests.get(source) >= self.max_rate.get(source): 
      logging.info('back off: {} until {}'.format(source, self.next_reset_at.get(source))) 
      halt((self.next_reset_at.get(source) - now).seconds) 

     self.num_requests[source] += 1 
     response = method() # potential exception raise 

     # cache the response 
     if do_cache: 
      self.cache[key] = (now, response) 
      logging.info('cached instance for {}, {}'.format(source, method)) 

     return response