utils/debounce.py

"""Debounce utility for Gradio event handlers."""

import asyncio
import time
from functools import wraps
from typing import Callable


def debounce(wait_seconds: float):
    """
    Decorator that debounces an async function.
    Only executes the function after `wait_seconds` of inactivity.
    """
    def decorator(fn: Callable) -> Callable:
        last_call: list[float] = [0.0]

        @wraps(fn)
        async def wrapper(*args, **kwargs):
            last_call[0] = time.monotonic()
            await asyncio.sleep(wait_seconds)
            if time.monotonic() - last_call[0] >= wait_seconds:
                return await fn(*args, **kwargs)

        return wrapper

    return decorator


class AsyncDebouncer:
    """Stateful debouncer for use in Gradio event handlers."""

    def __init__(self, wait_seconds: float = 0.5):
        self._wait = wait_seconds
        self._last_call = 0.0

    async def wait(self) -> None:
        """Sleep until debounce window expires."""
        self._last_call = time.monotonic()
        await asyncio.sleep(self._wait)