Debounce across multiple components to reduce frequent callbacks

Dash Python
1

While a lot of input components support debouncing it is sometimes desirable to debounce “across” different components, e.g. when a user can select from a lot of different filters and you do not want to update until the user has “finished” setting the filters.

If dcc.Store had a debounce parameter, it could be used for this purpose, but it has not in the current release.

I implemented an AIO component that can be used for this. I thought others could find this useful as well…

import uuid
from typing import Literal

from dash import MATCH, Input, Output, State, clientside_callback, dcc, html
from dash.dependencies import _Wildcard


class DebounceStoreAIO(html.Div):
    """AIO component that provides a debounced dcc.Store.

    This component wraps a dcc.Store and adds debouncing functionality to it.
    
    It can be used as a proxy for values that update frequently, but should only be propagated
    after a specified debounce delay to avoid excessive processing or API calls.

    This comes in handy when the input component itself does not support debouncing natively or when
    the debouncing behaviour should be coordinated across multiple components.
    
    Usage:
        # Create the debounce store with a 500ms debounce delay
        debounce_store = DebounceStoreAIO(aio_id="my-input", debounce_ms=500)
        
        # Write to the immediate store from any callback
        @callback(
            Output(DebounceStoreAIO.ids.immediate("my-input"), "data"),
            Input("some-component", "value")
        )
        def update_immediate(value):
            return value
            
        # Read from the debounced store
        @callback(
            Output("result", "children"),
            Input(DebounceStoreAIO.ids.debounced("my-input"), "data")
        )
        def use_debounced_value(debounced_value):
            return f"Debounced: {debounced_value}"
    """

    class ids:  # noqa: N801
        @staticmethod
        def immediate(aio_id: str | _Wildcard) -> dict:
            """Store that receives immediate updates."""
            return {"component": "DebounceStoreAIO", "subcomponent": "immediate", "aio_id": aio_id}

        @staticmethod
        def debounced(aio_id: str | _Wildcard) -> dict:
            """Store that contains debounced values."""
            return {"component": "DebounceStoreAIO", "subcomponent": "debounced", "aio_id": aio_id}

        @staticmethod
        def interval(aio_id: str | _Wildcard) -> dict:
            """Interval component for checking updates."""
            return {"component": "DebounceStoreAIO", "subcomponent": "interval", "aio_id": aio_id}

    def __init__(
        self,
        aio_id: str | None = None,
        debounce_ms: int = 500,
        storage_type: Literal["memory", "session", "local"] = "memory",
    ):
        """Initialize the DebounceStoreAIO component.

        Args:
            aio_id: Unique identifier for this instance. If None, a random ID is generated.
            debounce_ms: Debounce delay in milliseconds. Default is 500ms.
                         Note: Interval checks at debounce_ms/2 with a minimum of 50ms,
                         resulting in an effective minimum debounce of 100ms.
            storage_type: Storage type for the stores ('memory', 'session', or 'local').
                         Default is 'memory'.
        """
        if aio_id is None:
            aio_id = uuid.uuid4().hex

        # Interval checks at half the debounce time for better responsiveness
        # Minimum interval of 50ms means effective minimum debounce of 100ms
        interval_ms = max(50, debounce_ms // 2)

        super().__init__(
            style={"display": "none"},
            children=[
                # Immediate store - receives immediate updates
                # modified_timestamp tracks when data was last changed
                dcc.Store(
                    id=self.ids.immediate(aio_id),
                    storage_type=storage_type,
                ),
                # Debounced store - contains debounced values
                dcc.Store(
                    id=self.ids.debounced(aio_id),
                    storage_type=storage_type,
                ),
                # Interval for periodic checks
                dcc.Interval(
                    id=self.ids.interval(aio_id),
                    interval=interval_ms,
                    n_intervals=0,
                ),
            ],
        )


# Client-side callback to handle debounce logic
clientside_callback(
    """
    function(n_intervals, immediate_value, immediate_timestamp, debounced_timestamp, interval_ms) {
        // If immediate timestamp is before or equal to debounced timestamp, debounced is already up-to-date
        // Note: We allow null/undefined values to propagate, so we don't check immediate_value here
        if (immediate_timestamp <= debounced_timestamp) {
            return window.dash_clientside.no_update;
        }
        
        // Immediate value has changed - check if enough time has passed since the last change
        const now = Date.now();
        const time_elapsed = now - immediate_timestamp;
        // Interval is set to debounce_ms / 2, so multiply by 2 to get the actual debounce delay
        const debounce_ms = interval_ms * 2;
        
        if (time_elapsed < debounce_ms) {
            // Still within debounce period - keep waiting
            return window.dash_clientside.no_update;
        }
        
        // Debounce period has passed - update debounced store (including null/undefined values)
        return immediate_value;
    }
    """,
    Output(DebounceStoreAIO.ids.debounced(MATCH), "data"),
    Input(DebounceStoreAIO.ids.interval(MATCH), "n_intervals"),
    State(DebounceStoreAIO.ids.immediate(MATCH), "data"),
    State(DebounceStoreAIO.ids.immediate(MATCH), "modified_timestamp"),
    State(DebounceStoreAIO.ids.debounced(MATCH), "modified_timestamp"),
    State(DebounceStoreAIO.ids.interval(MATCH), "interval"),
    prevent_initial_call=True,
)
1 Like