🐍
Code Snippet — July 22, 2026Debounce & Throttle DecoratorsPrevent high-frequency handlers from firing on every event. Auto async/sync detection, leading-edge mode,
max_wait throttle floor,
.cancel() &
.flush().
import asyncio, functools, threading
def debounce(delay=0.3, *, leading=False):
def decorator(func):
is_async = asyncio.iscoroutinefunction(func)
if is_async:
timer = None; pending = False; last_a = (); last_k = {}
@functools.wraps(func)
async def wrapper(*a, **k):
nonlocal timer, pending, last_a, last_k
if timer: timer.cancel(); timer = None
last_a, last_k = a, k
if leading and not pending:
pending = True; return func(*a, **k)
pending = True
async def _fire():
nonlocal timer, pending
timer = None; pending = False
await func(*last_a, **last_k)
timer = asyncio.get_event_loop().call_later(delay, lambda: asyncio.ensure_future(_fire()))
wrapper.cancel = lambda: (timer.cancel() if timer else None)
else:
t = None; p = False; la = (); lk = {}
@functools.wraps(func)
def wrapper(*a, **k):
nonlocal t, p, la, lk
if t: t.cancel(); t = None
la, lk = a, k
if leading and not p:
p = True; return func(*a, **k)
p = True
def _fire():
nonlocal t, p; t = None; p = False; func(*la, **lk)
t = threading.Timer(delay, _fire); t.start()
wrapper.cancel = lambda: (t.cancel() or (t:=None) if t else None)
return wrapper
return decorator
@debounce(0.3)
def search(q): print(f"🔍 {q}")
💡
Drop in any project — no third-party library needed.