feat(utils): add debounce helper utilities

New Features
- Introduce `debounce` async decorator to delay coroutine execution until a specified delay elapses.
- Add `sync_debounce` wrapper that uses GLib timeout for synchronous functions, supporting immediate execution and repeat counts.
This commit is contained in:
sankalp committed 2026-02-26 06:42:00 +05:45
1 parent deb9b1082f
commit 934b62ebff
1 file changed
+70
+70
View File
@@ -0,0 +1,70 @@
import asyncio
import functools
import typing as t
from fabric.utils import GLib
def debounce(delay: float) -> t.Callable[..., t.Any]:
def decorator(func: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]:
task: asyncio.Task[None] | None = None
@functools.wraps(func)
async def wrapper(*args: t.Any, **kwargs: t.Any) -> None:
nonlocal task
if task is not None and not task.done():
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
async def delayed_call() -> None:
await asyncio.sleep(delay)
await func(*args, **kwargs)
task = asyncio.create_task(delayed_call())
return wrapper
return decorator
def sync_debounce(
delay: int, min_n_times: int = 0, immediate: bool = False
) -> t.Callable[..., t.Any]:
def decorator(func: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]:
n_times = 0
timeout_id: int | None = None
last_args: tuple[t.Any, ...] = ()
last_kwargs: dict[str, t.Any] = {}
def call_func() -> bool:
nonlocal timeout_id, n_times
if n_times > min_n_times:
func(*last_args, **last_kwargs)
n_times = 0
timeout_id = None
return False
def wrapper(*args: t.Any, **kwargs: t.Any) -> None:
nonlocal timeout_id, last_args, last_kwargs, n_times
last_args = args
last_kwargs = kwargs
if timeout_id is not None:
GLib.source_remove(timeout_id)
if not immediate or n_times < min_n_times:
n_times += 1
timeout_id = GLib.timeout_add(delay, call_func)
else:
n_times = 0
timeout_id = None
func(*last_args, **last_kwargs)
return functools.wraps(func)(wrapper)
return decorator