[Feature Proposal] Preprocessed / Plugin callbacks

[rubenthoms]

Is your feature request related to a problem? Please describe. Sometimes, callbacks can become quite large and complex. Moreover, code might get repeated in multiple callbacks. Of course, repeated parts of code can be moved to a common function but callback inputs/states cannot.

One possible workaround is to move the repeated code to its own callback. However, the result would need to be stored in an output or a server-side caching mechanism would need to be implemented. When e.g. storing the result in a dcc.Store, data would be sent to the client and back to the server - an unnecessary loop.

Finding a better way of reusing whole callback implementations (i.e. callback decorator arguments as well as the wrapped function) could improve code quality, reduce code repetitions, and make complex callbacks allocable and, hence, better comprehensible.

Describe the solution you'd like A possible solution would be to implement support for something I'd call Preprocessed Callbacks or Plugin Callbacks (I haven't found a perfect name yet 😉).

Plugin Callbacks are registered by using a new decorator, e.g. @plugin_callback, which takes dash.Inputs and dash.States as arguments as well as a name. After the decorator, a regular Python function can be defined which takes the Inputs and States as arguments and returns a value (/tuple of values). In an advanced Dash @callback one could refer to this Plugin Callback by using a new PluginCallback class and providing the name of the requested Plugin Callback. If such a callback is registered, it, i.e. its registered inputs and states, get injected in a regular Dash callback. The function defined after the @callback decorator gets provided one argument per PluginCallback (as it is for Input / State). When the callback is triggered, the function associated to the PluginCallback gets called first with the injected inputs/states and the result is forwarded together with the other inputs/states to the wrapped function (that follows the @callback decorator).

Here is a very simple first example implementation to illustrate this proposal.

NOTE:

• Inputs/States already used in a Plugin Callback cannot be reused in the main callback yet (but easy to implement)
• PluginCallbacks should probably be able to use other PluginCallbacks in their callback decorator

from typing import Any, Callable, Dict, List, Tuple, Union
from dataclasses import dataclass

from dash import Dash, html, Input, Output, State, callback
from dash_core_components import Slider


@dataclass(frozen=True)
class RegisteredPluginCallback:
    inputs_and_states: List[Union[Input, State]]
    func: Callable

_REGISTERED_PLUGIN_CALLBACKS: Dict[str, RegisteredPluginCallback] = {}

class PluginCallback:
    def __init__(self, name: str) -> None:
        self._name = name

    @property
    def name(self) -> str:
        return self._name

def advanced_callback(*args: Any) -> Callable:
    outputs = []
    inputs_and_states = []
    plugin_callbacks: List[RegisteredPluginCallback] = []

    for arg in args:
        if isinstance(arg, Output):
            outputs.append(arg)
        elif isinstance(arg, (Input, State)):
            inputs_and_states.append(arg)
        elif isinstance(arg, PluginCallback):
            plugin_callbacks.append(_REGISTERED_PLUGIN_CALLBACKS[arg.name])

    def wrapper(func: Callable) -> None:
        @callback(
            *outputs,
            *inputs_and_states,
            *[fc.inputs_and_states for fc in plugin_callbacks]
        )
        def _callback_func(*args):
            adjusted_args = []
            for index in range(len(inputs_and_states)):
                adjusted_args.append(args[index])
            
            fc_index = len(inputs_and_states)
            for fc in plugin_callbacks:
                adjusted_args.append(fc.func(*[args[index] for index in range(fc_index, fc_index + len(fc.inputs_and_states))]))
                fc_index += len(fc.inputs_and_states)

            return func(*adjusted_args)
        
    return wrapper



def plugin_callback(*inputs_states: Tuple[Union[Input, State]], name: str) -> Callable:
    def wrapper(func: Callable) -> None:
        _REGISTERED_PLUGIN_CALLBACKS[name] = RegisteredPluginCallback(list(inputs_states), func)

    return wrapper


app = Dash(
    name="TestApp",
)

app.layout = html.Div([Slider(min=0, max=20, step=5, value=10, id="slider"), html.Div(id="output")])

@plugin_callback(Input("slider", "value"), name="my-cb")
def _square(value: int) -> int:
    return value * value

@advanced_callback(
    Output("output", "children"),
    PluginCallback("my-cb")
)
def _calc(value: int) -> str:
    return str(value)

app.run_server(host="localhost", port=5008, debug=False)

Looking forward to hearing your opinions and/or alternative solutions/approaches to this problem. 😄