Commits

Jason R. Coombs  committed 2798312

Adding IntervalGovernor

  • Participants
  • Parent commits af5e3cc

Comments (0)

Files changed (5)

 Changes
 -------
 
+6.2dev
+~~~
+
+* Added IntervalGovernor to `timing` module. Allows one to decorate a
+  function, causing that function to only be called once per interval, despite
+  the number of calls attempted.
+
 6.1
 ~~~
 

File jaraco/util/timing.py

-from __future__ import unicode_literals
+from __future__ import unicode_literals, absolute_import
 
 import datetime
+import functools
 
 class Stopwatch(object):
 	"""
 
 	def __exit__(self, exc_type, exc_value, traceback):
 		self.stop()
+
+class IntervalGovernor(object):
+	"""
+	Decorate a function to only allow it to be called once per
+	min_interval. Otherwise, it returns None.
+	"""
+	def __init__(self, min_interval):
+		if isinstance(min_interval, (int, long)):
+			min_interval = datetime.timedelta(seconds=min_interval)
+		self.min_interval = min_interval
+		self.last_call = None
+
+	def decorate(self, func):
+		@functools.wraps(func)
+		def wrapper(*args, **kwargs):
+			allow = (
+				not self.last_call
+				or self.last_call.split() > self.min_interval
+			)
+			if allow:
+				self.last_call = Stopwatch()
+				return func(*args, **kwargs)
+		return wrapper
+
+	__call__ = decorate
 with-doctest=1
 
 [pytest]
-norecursedirs = build bugs sandbox
+norecursedirs = build bugs sandbox *.egg
 addopts = --doctest-modules
 	],
 	tests_require=[
 		'pytest>=2',
+		'mock',
 	],
 	setup_requires=[
 		'hgtools',

File tests/test_timing.py

+import datetime
+
+import mock
+
+from jaraco.util import timing
+
+
+def test_IntervalGovernor():
+	"""
+	IntervalGovernor should prevent a function from being called more than
+	once per interval.
+	"""
+	func_under_test = mock.MagicMock()
+	# to look like a function, it needs a __name__ attribute
+	func_under_test.__name__ = 'func_under_test'
+	interval = datetime.timedelta(seconds=1)
+	governed = timing.IntervalGovernor(interval)(func_under_test)
+	governed('a')
+	governed('b')
+	governed(3, 'sir')
+	func_under_test.assert_called_once_with('a')