Skip to content

Instantly share code, notes, and snippets.

@AdoHaha
Last active May 7, 2016 06:17
Show Gist options
  • Star 2 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save AdoHaha/6306372 to your computer and use it in GitHub Desktop.
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
#!/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()
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