feat : Added new decorator - rate_limiter.

Author: manas-shinde

decorators/rate_limiter.py

@@ -0,0 +1,22 @@
+import time
+import threading
+import functools
+
+
+def rate_limiter(calls_per_second):
+    interval = 1 / calls_per_second
+    lock = threading.Lock()
+    last_called = [0.0]
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            with lock:
+                now = time.time()
+                wait = interval - (now - last_called[0])
+                if wait > 0:
+                    time.sleep(wait)
+                last_called[0] = time.time()
+            return func(*args, **kwargs)
+        return wrapper
+    return decorator

tests/test_rate_limiter.py

@@ -0,0 +1,21 @@
+import time
+from decorators.rate_limiter import rate_limiter
+
+calls = []
+
+
+# Max 2 calls per second = 0.5s delay between calls
+@rate_limiter(calls_per_second=2)
+def limited_func(x):
+    calls.append(time.time())
+    return x * 2
+
+
+def test_rate_limiter_timing():
+    calls.clear()
+    for i in range(3):
+        limited_func(i)
+
+    assert len(calls) == 3
+    assert (calls[1] - calls[0]) >= 0.49
+    assert (calls[2] - calls[1]) >= 0.49