Add debounce_interval to throttles

This commit is contained in:
Aarni Koskela 2019-09-26 18:36:20 +03:00
parent 0bcb275281
commit 81c150b5e8
3 changed files with 97 additions and 1 deletions

View File

@ -195,8 +195,36 @@ The following is an example of a rate throttle, that will randomly throttle 1 in
def allow_request(self, request, view):
return random.randint(1, 10) != 1
# Debouncing
[Debouncing][debounce] refers to disallowing rapid invocations of the throttled view regardless of its actual throttling rate.
When deriving a custom throttle from `throttling.SimpleRateThrottle` (or one of its concrete subclasses described above), you may override the `get_debounce_interval()` function, or set the
class-level `debounce_interval` to a value in seconds.
For instance,
class FiveSecondDebounce(throttling.UserRateThrottle):
debounce_interval = 5
rate = '10/min'
would allow each user/IP address (as described above) to invoke the view 10 times in a minute, but only after 5 seconds have
passed from the previous successful invocation.
Using `get_debounce_interval()`, one can e.g. ensure superusers can always invoke the view regardless of debouncing.
class FiveSecondDebounce(throttling.UserRateThrottle):
rate = '10/min'
def get_debounce_interval(self, request, view):
if request.user.is_superuser:
return 0
return 5
[cite]: https://developer.twitter.com/en/docs/basics/rate-limiting
[permissions]: permissions.md
[identifying-clients]: http://oxpedia.org/wiki/index.php?title=AppSuite:Grizzly#Multiple_Proxies_in_front_of_the_cluster
[cache-setting]: https://docs.djangoproject.com/en/stable/ref/settings/#caches
[cache-docs]: https://docs.djangoproject.com/en/stable/topics/cache/#setting-up-the-cache
[debounce]: https://en.wikipedia.org/wiki/Switch#Contact_bounce

View File

@ -64,6 +64,7 @@ class SimpleRateThrottle(BaseThrottle):
cache_format = 'throttle_%(scope)s_%(ident)s'
scope = None
THROTTLE_RATES = api_settings.DEFAULT_THROTTLE_RATES
debounce_interval = None
def __init__(self):
if not getattr(self, 'rate', None):
@ -106,6 +107,18 @@ class SimpleRateThrottle(BaseThrottle):
duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
return (num_requests, duration)
def get_debounce_interval(self, request, view):
"""
Get the debounce interval for a request and view, in units returned
by the throttle's `timer` (usually seconds).
At least this much time must pass between successful (non-throttled)
invocations of the action.
If a falsy value is returned, debounce is disabled.
"""
return self.debounce_interval
def allow_request(self, request, view):
"""
Implement the check to see if the request should be throttled.
@ -113,7 +126,9 @@ class SimpleRateThrottle(BaseThrottle):
On success calls `throttle_success`.
On failure calls `throttle_failure`.
"""
if self.rate is None:
debounce_interval = self.get_debounce_interval(request, view)
if self.rate is None and not debounce_interval:
return True
self.key = self.get_cache_key(request, view)
@ -123,6 +138,12 @@ class SimpleRateThrottle(BaseThrottle):
self.history = self.cache.get(self.key, [])
self.now = self.timer()
# Perform debounce, i.e. limiting of subsequent (rapid) invocations.
# Basically, if there is history of requests and the latest (first)
# entry has taken place less than `debounce_interval` seconds ago, fail.
if debounce_interval and self.history and self.now - self.history[0] < debounce_interval:
return self.debounce_failure()
# Drop any requests from the history which have now passed the
# throttle duration
while self.history and self.history[-1] <= self.now - self.duration:
@ -146,9 +167,17 @@ class SimpleRateThrottle(BaseThrottle):
"""
return False
def debounce_failure(self):
"""
Called when a request to the API has failed due to debouncing.
"""
return False
def wait(self):
"""
Returns the recommended next request time in seconds.
Does not take debounce time into account.
"""
if self.history:
remaining_duration = self.duration - (self.now - self.history[-1])

View File

@ -512,3 +512,42 @@ class AnonRateThrottleTests(TestCase):
request = Request(HttpRequest())
cache_key = self.throttle.get_cache_key(request, view={})
assert cache_key == 'throttle_anon_None'
def test_debounce_throttle():
class DebounceThrottle(ThrottleTestTimerMixin, AnonRateThrottle):
debounce_interval = 1
rate = '1000/min' # More often than the debounce would allow
scope = 'debounce'
def get_debounce_interval(self, request, view):
# This is here to test that requests can affect the debounce interval
if request.query_params.get('secret'):
return 0
return super().get_debounce_interval(request, view)
class DebouncedAPIView(APIView):
throttle_classes = (DebounceThrottle,)
def get(self, request):
return Response('foo')
factory = APIRequestFactory()
request = factory.get('/')
view = DebouncedAPIView().as_view()
DebounceThrottle.TIMER_SECONDS = 0
assert view(request).status_code == 200 # first request works fine
DebounceThrottle.TIMER_SECONDS = 0.1
assert view(request).status_code == 429 # 0.1 seconds after the last one should not
assert view(factory.get('/', {'secret': '1'})).status_code == 200 # (unless we pass the secret parameter)
DebounceThrottle.TIMER_SECONDS = 1.1
assert view(request).status_code == 200 # after 1.1 seconds (since the secret reset the throttle), things work
assert view(request).status_code == 429 # but a second request at the same instant shouldn't work
DebounceThrottle.TIMER_SECONDS = 1.6
assert view(request).status_code == 429 # nor 0.5 seconds after that.
DebounceThrottle.debounce_interval = 0.2 # However after changing the interval...
assert view(request).status_code == 200 # ... things should be fine again