perf($Async): add async debounce decorator

home_guardian/common/debounce_throttle.py

@@ -1,3 +1,5 @@
+import asyncio
+from datetime import datetime
 from threading import Timer
 from time import time
 
@@ -33,7 +35,7 @@ def call_it():
     return decorator
 
 
-class Wrapper:
+class _Wrapper:
     def __init__(self, initial_value=None) -> None:
         self._value = initial_value
 
@@ -44,6 +46,52 @@ def set(self, v):
         self._value = v
 
 
+def async_debounce(interval: float):
+    """
+    Async debounce decorator.
+
+    Python 中的防抖与节流 https://www.bilibili.com/read/cv13257868/
+    Python 中的防抖与节流 https://www.moyu.moe/articles/25/
+
+    :param interval: interval time in seconds
+    """
+
+    # 用于存储 asyncio.Task 实例，这里是闭包
+    task_wrapper = _Wrapper()
+
+    def decorator(fn):
+        # Task 协程函数
+        async def f(*args, **kwargs):
+            # 进行休眠
+            logger.debug("Set debounced function delay in {} seconds", interval)
+            await asyncio.sleep(interval)
+
+            # 调用函数
+            f1 = fn(*args, **kwargs)
+            logger.debug("Called debounced function, {}", fn)
+
+            # 支持回调函数是异步函数的情况
+            if asyncio.iscoroutine(f1):
+                await f1
+
+            # 清除 task_wrapper
+            task_wrapper.set(None)
+
+        def wrapper(*args, **kwargs):
+            # 如果 task_wrapper 存在，说明在 delay 秒内调用过一次函数，此次调用应当重置计时器，因此取消先前的 Task
+            if task_wrapper.get() is not None:
+                task_wrapper.get().cancel()
+                logger.debug("Task cancelled, {}", task_wrapper.get())
+
+            # 创建 Task 并赋值变量 task_wrapper
+            task_wrapper.set(asyncio.create_task(f(*args, **kwargs)))
+            logger.debug(f"Created task: {task_wrapper.get()}")
+
+        return wrapper
+
+    return decorator
+
+
 def throttle(interval: float):
     """
     Throttle decorator.
@@ -54,7 +102,7 @@ def throttle(interval: float):
     """
 
     # 下一次许可调用的时间，初始化为 0
-    next_t = Wrapper(0)
+    next_t = _Wrapper(0)
 
     def decorator(fn):
         def throttled(*args, **kwargs):
@@ -73,3 +121,8 @@ def throttled(*args, **kwargs):
         return throttled
 
     return decorator
+
+
+@async_debounce(1)
+def debounced_example_1():
+    logger.info(f"debounced_example_1 function: {datetime.now()}")

test/test_debounce_and_throttle.py

@@ -2,7 +2,7 @@
 
 from loguru import logger
 
-from home_guardian.common.debounce_throttle import debounce, throttle
+from home_guardian.common.debounce_throttle import async_debounce, debounce, throttle
 
 
 def test_debounce() -> None:
@@ -16,6 +16,14 @@ def test_debounce() -> None:
     sleep(2)
 
 
+# @pytest.mark.asyncio
+# async def test_async_debounce():
+#     try:
+#         await asyncio.new_event_loop().run_until_complete(async_debounce_function() for _ in range(3))
+#     except Exception:
+#         assert False, "Failed to test throttle_function()"
+
+
 def test_throttle() -> None:
     call_count: int = 5
     try:
@@ -32,6 +40,11 @@ def debounce_function() -> None:
     logger.warning("'debounce_function' was called")
 
 
+@async_debounce(0.25)
+def async_debounce_function():
+    logger.warning("'async_debounce_function' was called")
+
+
 @throttle(0.25)
 def throttle_function() -> None:
     logger.warning("'throttle_function' was called")