Skip to main content
Version: 0.3.1

09: Signal chain

A pipeline of three signals where each callback's output is the next one's input. Click a button → signal 1 fires → signal 2 enriches it → signal 3 enriches it again → final consumer renders. Shows how to express sequential async-like work in Dash.

The script​

import random
from dash import html
from dash_extensions.enrich import Output, Trigger
from weaverlet.base import SignalOutput, SignalInput, WeaverletComponent, WeaverletApp, Identifier
from weaverlet.components import SignalComponent


class SignalChainComponent(WeaverletComponent):

    trigger_button_id = Identifier()
    label_p_id = Identifier()

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.signal_1_component = SignalComponent()
        self.signal_2_component = SignalComponent()
        self.signal_3_component = SignalComponent()

    def get_layout(self):
        return html.Div(
            [
                self.signal_1_component(),
                self.signal_2_component(),
                self.signal_3_component(),
                html.Button('Click to trigger signal chain',
                            id=self.trigger_button_id),
                html.P(id=self.label_p_id)
            ]
        )

    def register_callbacks(self, app):

        @app.callback(
            SignalOutput(self.signal_1_component),
            Trigger(self.trigger_button_id, 'n_clicks')
        )
        def button_click():
            return {}

        @app.callback(
            SignalOutput(self.signal_2_component),
            SignalInput(self.signal_1_component)
        )
        def signal_1(signal_input):
            signal_input['random_number_1'] = random.random()
            return signal_input

        @app.callback(
            SignalOutput(self.signal_3_component),
            SignalInput(self.signal_2_component)
        )
        def signal_2(signal_input):
            signal_input['random_number_2'] = random.random()
            return signal_input

        @app.callback(
            Output(self.label_p_id, 'children'),
            SignalInput(self.signal_3_component)
        )
        def signal_3(signal_input):
            return f'Signal chain triggered! data: {signal_input}'


signal_trigger_component = SignalChainComponent()

wapp = WeaverletApp(root_component=signal_trigger_component)
wapp.app.run(port=8089)

Expected output​

  • Clicking the button triggers four callbacks in sequence:
    1. Button click → emits {} to signal 1.
    2. Signal 1 → enriches with random_number_1 → emits to signal 2.
    3. Signal 2 → enriches with random_number_2 → emits to signal 3.
    4. Signal 3 → renders the enriched dict as text.
  • The paragraph shows something like: Signal chain triggered! data: {'random_number_1': 0.42, 'random_number_2': 0.81}

Notes​

  • Each chain stage costs one browser round-trip (each emission writes to a dcc.Store, which the browser sends back to Dash). Long chains (5+ stages) start to feel sluggish.
  • Chains are useful when stages are independently observable (you want to fan out at stage 2 to a logger and at stage 3 to the next step), or when you want pluggable stages you can reorder/insert later.
  • If a chain is just a sequence of pure dict transformations with no fan-out, consider collapsing it into one callback.
  • For large payloads in chains, see Serverside signals. Keeps data on the server, sends only a token to the browser between stages.

Try it locally​

pip install weaverlet
python signal_chain_app.py

See also​