Adds throttle decorator

blinkpy/helpers/util.py

@@ -2,6 +2,7 @@
 
 import logging
 import time
+from functools import wraps
 from requests import Request, Session, exceptions
 from blinkpy.helpers.constants import BLINK_URL, TIMESTAMP_FORMAT
 import blinkpy.helpers.errors as ERROR
@@ -121,3 +122,33 @@ class BlinkURLHandler():
         self.networks_url = "{}/networks".format(self.base_url)
         self.video_url = "{}/api/v2/videos".format(self.base_url)
         _LOGGER.debug("Setting base url to %s.", self.base_url)
+
+
+class Throttle():
+    """Class for throttling api calls."""
+
+    def __init__(self, seconds=10):
+        """Initialize throttle class."""
+        self.throttle_time = seconds
+        self.last_call = 0
+
+    def __call__(self, method):
+        """Throttle caller method."""
+        def throttle_method():
+            """Call when method is throttled."""
+            return None
+
+        @wraps(method)
+        def wrapper(*args, **kwargs):
+            """Wrap that checks for throttling."""
+            force = kwargs.pop('force', False)
+            now = int(time.time())
+            last_call_delta = now - self.last_call
+            if force or last_call_delta > self.throttle_time:
+                result = method(*args, *kwargs)
+                self.last_call = now
+                return result
+
+            return throttle_method()
+
+        return wrapper

pylintrc

@@ -8,6 +8,8 @@ reports=no
 # too-many-* - are not enforced for the sake of readability
 # too-few-* - same as too-many-*
 # no-else-return - I don't see any reason to enforce this. both forms are readable
+# no-self-use - stupid and only annoying
+# unexpected-keyword-arg - doesn't allow for use of **kwargs, which is dumb
 
 disable=
   locally-disabled,
@@ -23,3 +25,5 @@ disable=
   too-many-lines,
   too-few-public-methods,
   no-else-return,
+  no-self-use,
+  unexpected-keyword-arg,

tests/test_util.py

@@ -0,0 +1,100 @@
+"""Test various api functions."""
+
+import unittest
+from unittest import mock
+import time
+from blinkpy.helpers.util import Throttle
+
+
+class TestUtil(unittest.TestCase):
+    """Test the helpers/util module."""
+
+    def setUp(self):
+        """Initialize the blink module."""
+
+    def tearDown(self):
+        """Tear down blink module."""
+
+    def test_throttle(self):
+        """Test the throttle decorator."""
+        calls = []
+
+        @Throttle(seconds=5)
+        def test_throttle():
+            calls.append(1)
+
+        now = int(time.time())
+        now_plus_four = now + 4
+        now_plus_six = now + 6
+
+        test_throttle()
+        self.assertEqual(1, len(calls))
+
+        # Call again, still shouldn't fire
+        test_throttle()
+        self.assertEqual(1, len(calls))
+
+        # Call with force
+        test_throttle(force=True)
+        self.assertEqual(2, len(calls))
+
+        # Call without throttle, shouldn't fire
+        test_throttle()
+        self.assertEqual(2, len(calls))
+
+        # Fake time as 4 seconds from now
+        with mock.patch('time.time', return_value=now_plus_four):
+            test_throttle()
+        self.assertEqual(2, len(calls))
+
+        # Fake time as 6 seconds from now
+        with mock.patch('time.time', return_value=now_plus_six):
+            test_throttle()
+        self.assertEqual(3, len(calls))
+
+    def test_throttle_per_instance(self):
+        """Test that throttle is done once per instance of class."""
+        class Tester:
+            """A tester class for throttling."""
+
+            def test(self):
+                """Test the throttle."""
+                return True
+
+        tester = Tester()
+        throttled = Throttle(seconds=1)(tester.test)
+        self.assertEqual(throttled(), True)
+        self.assertEqual(throttled(), None)
+
+    def test_throttle_on_two_methods(self):
+        """Test that throttle works for multiple methods."""
+        class Tester:
+            """A tester class for throttling."""
+
+            @Throttle(seconds=3)
+            def test1(self):
+                """Test function for throttle."""
+                return True
+
+            @Throttle(seconds=5)
+            def test2(self):
+                """Test function for throttle."""
+                return True
+
+        tester = Tester()
+        now = time.time()
+        now_plus_4 = now + 4
+        now_plus_6 = now + 6
+
+        self.assertEqual(tester.test1(), True)
+        self.assertEqual(tester.test2(), True)
+        self.assertEqual(tester.test1(), None)
+        self.assertEqual(tester.test2(), None)
+
+        with mock.patch('time.time', return_value=now_plus_4):
+            self.assertEqual(tester.test1(), True)
+            self.assertEqual(tester.test2(), None)
+
+        with mock.patch('time.time', return_value=now_plus_6):
+            self.assertEqual(tester.test1(), None)
+            self.assertEqual(tester.test2(), True)