Last active
May 7, 2016 06:17
-
-
Save AdoHaha/6306372 to your computer and use it in GitHub Desktop.
A throttle function decorator in Python similar to the one in underscore.js, inspired by debounce decorator from: https://gist.github.com/walkermatt/2871026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python | |
import unittest | |
import time | |
from throttle import throttle | |
class TestThrottle(unittest.TestCase): | |
@throttle(1) | |
def increment(self): | |
""" Simple function that | |
increments a counter when | |
called, used to test the | |
throttle function decorator """ | |
self.count += 1 | |
def setUp(self): | |
self.count = 0 | |
def test_throttle(self): | |
""" Test that the increment | |
function is being throttled. | |
Function should be used only once a second (and used at start)""" | |
self.assertTrue(self.count == 0) | |
self.increment() | |
self.assertTrue(self.count == 1) | |
self.increment() | |
self.increment() | |
self.increment() | |
self.increment() | |
self.increment() | |
time.sleep(0.25) | |
self.increment() | |
self.increment() | |
self.increment() | |
self.increment() | |
self.increment() | |
self.increment() | |
self.assertTrue(self.count == 1) | |
self.increment() | |
self.increment() | |
self.increment() | |
self.increment() | |
self.assertTrue(self.count == 1) | |
time.sleep(1) | |
self.assertTrue(self.count == 2) | |
time.sleep(10) | |
self.assertTrue(self.count == 2) | |
if __name__ == '__main__': | |
unittest.main() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from threading import Timer | |
import time | |
def throttle(mindelta): | |
def decorator(fn): | |
def throttled(*args,**kwargs): | |
def call_it(): | |
throttled.lastTimeExecuted=time.time() | |
fn(*args, **kwargs) | |
if hasattr(throttled,"lastTimeExecuted"): | |
lasttime=throttled.lastTimeExecuted | |
else: #just execute fction | |
try: | |
throttled.t.cancel() | |
except(AttributeError): | |
pass | |
call_it() | |
return throttled | |
delta=time.time()-throttled.lastTimeExecuted | |
try: | |
throttled.t.cancel() | |
except(AttributeError): | |
pass | |
if delta>mindelta: | |
call_it() | |
else: | |
timespot=mindelta-delta | |
timespot=timespot | |
throttled.t=Timer(timespot,call_it) | |
throttled.t.start() | |
return throttled | |
return decorator |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment