Feat: Compatibility with NiceGUI bindings

[LeonMusCoden]

Hi,

Thank you for the package! It is already integrating nicely in my projects.
I do have one feature that would improve the DX quite a lot when working in NiceGUI apps: Having Signal and Computed be BindableProperties.

Use case example:
Currently, I'm using the following work around quite a lot:

class State:
    def __init__(self):
        self.count_text = Computed(...)

state = State()
a_label = ui.label()

_label_effect = Effect(lambda: a_label.set_text(state.count_text()))

This could be made easier with support for BindableProperties:

class State:
    def __init__(self):
        self.count_text = Computed(...)

state = State()
ui.label().bind_text_from(state.count_text)

[buiapp]

Great to hear that it's getting used in your projects! :)
I will look how I can make the integration into NiceGUI nicer. But I think that will be a separate package because I want to keep reaktiv framework-agnostic.

[LeonMusCoden]

I've created a proxy, which does what I requested (and a bit more):

from typing import Any, TypeVar
from collections.abc import Callable
from nicegui import binding
from typing_extensions import override
from reaktiv import Signal, Computed, Effect

T = TypeVar("T")


class ReactiveProxy:
    """A proxy object that makes Reaktiv signals compatible with NiceGUI's binding system.

    This class bridges Reaktiv's reactive state management with NiceGUI's UI binding system,
    allowing reactive updates to propagate to the UI components automatically.
    """

    def __init__(self):
        """Initialize a new ReactiveProxy instance.

        Attributes:
            _signals: Dictionary mapping names to Signal or Computed instances.
            _effects: Dictionary of Effect instances used for reactivity.
            _propagation_callbacks: Dictionary mapping property names to callback lists.
        """
        self._signals: dict[str, Signal[Any] | Computed[Any]] = {}  # pyright: ignore[reportExplicitAny]
        self._effects: dict[str, Effect] = {}
        self._propagation_callbacks: dict[str, list[Callable[[], None]]] = {}

    @classmethod
    def create(cls) -> "ReactiveProxy":
        """Create a new empty ReactiveProxy.

        Returns:
            ReactiveProxy: A new ReactiveProxy instance with no signals.
        """
        return cls()

    @classmethod
    def from_object(cls, obj: object, recursive: bool = True) -> "ReactiveProxy":
        """Convert an object containing Reaktiv signals into a NiceGUI-compatible proxy.

        Args:
            obj: The source object containing Signal or Computed instances.
            recursive: Whether to recursively process nested objects. Defaults to True.

        Returns:
            ReactiveProxy: A proxy containing all discovered signals from the object.
        """
        proxy = cls()

        # Helper function to recursively find signals
        def add_signals_from_obj(current_obj: object, prefix: str = ""):
            for attr_name in dir(current_obj):
                if attr_name.startswith("_"):
                    continue

                try:
                    attr_value = getattr(current_obj, attr_name)
                    full_name = f"{prefix}{attr_name}" if prefix else attr_name

                    if isinstance(attr_value, (Signal, Computed)):
                        proxy.add_signal(full_name, attr_value)
                    elif (
                        recursive
                        and hasattr(attr_value, "__dict__")
                        and not isinstance(attr_value, type)
                    ):
                        # Recursively process nested objects (excluding classes)
                        add_signals_from_obj(attr_value, f"{full_name}.")
                except Exception:
                    # Skip attributes that can't be accessed
                    continue

        add_signals_from_obj(obj)
        return proxy

    def add_signal(self, name: str, signal: Signal[T] | Computed[T]) -> "ReactiveProxy":
        """Add a signal to this proxy with the given name.

        Args:
            name: The property name to expose the signal as.
            signal: The Reaktiv Signal or Computed to bind.

        Returns:
            ReactiveProxy: The proxy instance for method chaining.
        """
        self._signals[name] = signal
        self._setup_property(name)
        return self

    def add_signals(self, **signals: Signal[Any] | Computed[Any]) -> "ReactiveProxy":  # pyright: ignore[reportExplicitAny]
        """Add multiple signals at once using keyword arguments.

        Args:
            **signals: Keyword arguments where keys are property names and values are signals.

        Returns:
            ReactiveProxy: The proxy instance for method chaining.
        """
        for name, signal in signals.items():
            self.add_signal(name, signal)
        return self

    def watch(self, name: str, callback: Callable[[], None]) -> "ReactiveProxy":
        """Watch a signal for changes and trigger a callback when it changes.

        Args:
            name: The name of the signal to watch.
            callback: Function to call when the signal changes.

        Returns:
            ReactiveProxy: The proxy instance for method chaining.

        Raises:
            ValueError: If the signal with the given name doesn't exist.
        """
        if name not in self._signals:
            raise ValueError(f"Signal '{name}' not found in ReactiveProxy")

        # Create a unique key for this watcher effect
        effect_key = f"watch_{name}_{id(callback)}"

        # Create an effect that triggers the callback when the signal changes
        def watcher_effect():
            # Access the signal value to create a dependency
            _ = self._signals[name]()
            # Call the callback
            callback()

        # Store the effect to keep it alive
        self._effects[effect_key] = Effect(watcher_effect)

        return self

    def remove_signal(self, name: str) -> "ReactiveProxy":
        """Remove a signal from this proxy by name.

        Args:
            name: The name of the signal to remove.

        Returns:
            ReactiveProxy: The proxy instance for method chaining.
        """
        if name in self._signals:
            # Clean up any related effects
            effects_to_remove = [
                key
                for key in self._effects.keys()
                if key == name or key.startswith(f"watch_{name}_")
            ]
            for effect_key in effects_to_remove:
                if hasattr(self._effects[effect_key], "dispose"):
                    self._effects[effect_key].dispose()
                del self._effects[effect_key]

            # Remove the signal
            del self._signals[name]

            # Clean up any propagation callbacks
            if name in self._propagation_callbacks:
                del self._propagation_callbacks[name]

        return self

    def remove_signals(self, *names: str) -> "ReactiveProxy":
        """Remove multiple signals at once by name.

        Args:
            *names: Names of the signals to remove.

        Returns:
            ReactiveProxy: The proxy instance for method chaining.
        """
        for name in names:
            self.remove_signal(name)
        return self

    def unwatch(
        self, name: str, callback: Callable[[], None] | None = None
    ) -> "ReactiveProxy":
        """Stop watching a signal for changes.

        Args:
            name: The name of the signal to stop watching.
            callback: The specific callback to remove. If None, removes all watchers for this signal.

        Returns:
            ReactiveProxy: The proxy instance for method chaining.

        Raises:
            ValueError: If the signal with the given name doesn't exist.
        """
        if name not in self._signals:
            raise ValueError(f"Signal '{name}' not found in ReactiveProxy")

        if callback is None:
            # Remove all watchers for this signal
            effects_to_remove = [
                key for key in self._effects.keys() if key.startswith(f"watch_{name}_")
            ]
            for effect_key in effects_to_remove:
                if hasattr(self._effects[effect_key], "dispose"):
                    self._effects[effect_key].dispose()
                del self._effects[effect_key]
        else:
            # Remove only the specific callback
            effect_key = f"watch_{name}_{id(callback)}"
            if effect_key in self._effects:
                if hasattr(self._effects[effect_key], "dispose"):
                    self._effects[effect_key].dispose()
                del self._effects[effect_key]

        return self

    def _setup_property(self, name: str) -> None:
        """Set up a property that NiceGUI can bind to.

        Args:
            name: The name of the signal property to set up.
        """

        # Create an Effect to watch for changes and trigger NiceGUI's propagation
        def update_cache() -> None:
            # Trigger NiceGUI's propagation system
            self._trigger_propagation(name)

        # Store the effect to keep it alive
        self._effects[name] = Effect(update_cache)

    def _trigger_propagation(self, name: str) -> None:
        """Trigger NiceGUI's binding propagation for a property.

        Args:
            name: The name of the property to trigger propagation for.
        """
        binding._propagate(self, name)  # pyright: ignore[reportPrivateUsage]

    def __getattr__(self, name: str) -> Any:  # pyright: ignore[reportExplicitAny]
        """Get attribute value for NiceGUI's binding system.

        Args:
            name: The name of the attribute to get.

        Returns:
            The current value of the signal.

        Raises:
            AttributeError: If the attribute doesn't exist.
        """
        if name.startswith("_"):
            # Private attributes bypass our system
            raise AttributeError(
                f"'{type(self).__name__}' object has no attribute '{name}'"
            )

        if name in self._signals:
            # Get current signal value
            return self._signals[name].get()
        else:
            raise AttributeError(
                f"'{type(self).__name__}' object has no attribute '{name}'"
            )

    @override
    def __setattr__(self, name: str, value: Any) -> None:  # pyright: ignore[reportExplicitAny]
        """Set attribute value for NiceGUI's binding system.

        Args:
            name: The name of the attribute to set.
            value: The new value to set.

        Raises:
            AttributeError: If trying to set a computed signal.
        """
        if name.startswith("_"):
            # Private attributes use normal setting
            super().__setattr__(name, value)
        elif name in self._signals:
            signal = self._signals[name]
            if hasattr(signal, "set"):
                signal.set(value)
            else:
                raise AttributeError(f"Cannot set value of computed signal '{name}'")
        else:
            # For unknown attributes, use default behavior
            super().__setattr__(name, value)
            # Trigger propagation
            self._trigger_propagation(name)

    def __hasattr__(self, name: str) -> bool:
        """Check if attribute exists for NiceGUI's binding system.

        Args:
            name: The name of the attribute to check.

        Returns:
            bool: True if the attribute exists, False otherwise.
        """
        return name in self._signals

    def dispose(self) -> None:
        """Clean up effects and references to prevent memory leaks."""
        for effect in self._effects.values():
            if hasattr(effect, "dispose"):
                effect.dispose()
        self._effects.clear()
        self._signals.clear()

Examples:

from nicegui import ui
from reaktiv import Signal, Computed
from reaktiv_nicegui_bridge.proxy import ReactiveProxy


# Example 1: Basic usage with ReactiveProxy
def example_basic():
    """Basic example showing how to use ReactiveProxy directly."""

    proxy = ReactiveProxy()
    proxy.add_signal("count", Signal[int](0))
    proxy.add_signal("doubled", Computed(lambda: proxy.count * 2))
    proxy.add_signal("message", Computed(lambda: f"Count: {proxy.count}"))

    # Create UI with bindings
    with ui.column():
        ui.label("Simple Counter Example")

        # Bind label text to computed signal
        ui.label().bind_text_from(proxy, "message")

        # Show doubled value
        ui.label().bind_text_from(proxy, "doubled")

        # Buttons to modify state
        ui.button(
            "Increment", on_click=lambda: setattr(proxy, "count", proxy.count + 1)
        )
        ui.button(
            "Decrement", on_click=lambda: setattr(proxy, "count", proxy.count - 1)
        )
        ui.button("Reset", on_click=lambda: setattr(proxy, "count", 0))


# Example 2: Two-way binding
def example_two_way_binding():
    """Example showing bidirectional binding between UI and signals."""

    # Create proxy with inline signal definitions
    proxy = ReactiveProxy().add_signals(
        name=Signal[str](""), email=Signal[str](""), age=Signal[int](0)
    )

    proxy.add_signal(
        "is_valid",
        Computed(
            lambda: (len(proxy.name) > 0 and "@" in proxy.email and 0 < proxy.age < 150)
        ),
    )

    proxy.add_signal(
        "summary",
        Computed(
            lambda: (
                f"✅ {proxy.name} ({proxy.age} years) - {proxy.email}"
                if proxy.is_valid
                else "❌ Please fill in all fields correctly"
            )
        ),
    )

    with ui.column():
        ui.label("User Form").classes("text-h6")

        # Input fields with two-way binding
        ui.input("Name").bind_value(proxy, "name")
        ui.input("Email").bind_value(proxy, "email")
        ui.number("Age", min=0, max=150).bind_value(proxy, "age")

        # Display summary
        ui.label().bind_text_from(proxy, "summary")


# Example 3: Using ReactiveProxy.from_object
def example_from_object():
    """Example showing how to use ReactiveProxy.from_object() to create a proxy from an existing class."""

    class TodoModel:
        def __init__(self):
            self.task = Signal[str]("")
            self.tasks = Signal[list[str]]([])
            self.task_count = Computed(lambda: len(self.tasks()))
            self.has_tasks = Computed(lambda: self.task_count() > 0)
            self.status_message = Computed(
                lambda: f"{self.task_count()} task(s) in the list"
                if self.has_tasks()
                else "No tasks yet"
            )

        def add_task(self):
            if self.task() and self.task() not in self.tasks():
                self.tasks.set(self.tasks() + [self.task()])
                self.task.set("")

        def remove_task(self, task):
            self.tasks.set([t for t in self.tasks() if t != task])

    # Create model instance
    model = TodoModel()

    # Create proxy from the model object
    proxy = ReactiveProxy.from_object(model)

    with ui.column():
        ui.label("Todo List Example").classes("text-h6")

        # Status message from computed property
        ui.label().bind_text_from(proxy, "status_message")

        # Add task input with two-way binding
        with ui.row().classes("w-full"):
            ui.input(placeholder="Enter a task").bind_value(proxy, "task")
            ui.button("Add", on_click=model.add_task).bind_enabled_from(
                proxy, "task", lambda v: v.strip() != ""
            )

        # Task list
        with ui.column().classes("w-full"):

            @ui.refreshable
            def create_task_list():
                with ui.element("div").classes("w-full") as container:
                    if proxy.has_tasks:
                        for task in proxy.tasks:
                            with ui.row().classes("w-full items-center"):
                                ui.label(task).classes("flex-grow")
                                ui.button(
                                    "✖", on_click=lambda t=task: model.remove_task(t)
                                )
                    else:
                        ui.label("Add some tasks to get started").classes(
                            "italic text-gray-500"
                        )
                return container

            create_task_list()

        proxy.watch("tasks", create_task_list.refresh)


def main():
    """Run all examples in a tabbed interface."""

    @ui.page("/")
    def example_page():
        with ui.tabs() as tabs:
            tab_basic = ui.tab("Basic")
            tab_two_way = ui.tab("Two-Way")
            tab_from_obj = ui.tab("From Object")

        with ui.tab_panels(tabs, value=tab_basic):
            with ui.tab_panel(tab_basic):
                example_basic()

            with ui.tab_panel(tab_two_way):
                example_two_way_binding()

            with ui.tab_panel(tab_from_obj):
                example_from_object()

    ui.run(port=5000)


main()

[buiapp]

Thank you for sharing your ReaktivProxy! This will surely be helpful for other NiceGUI developers.

I am currently looking into the Watcher mechanism from the TC39-Signals-Proposal to help Framework, Library authors to integrate reaktiv more neatly.

[Kaleidosium]

I am currently looking into the Watcher mechanism from the TC39-Signals-Proposal to help Framework, Library authors to integrate reaktiv more neatly.

I would love to see this implemented in reaktiv! Currently building a library for declarative UI with pyglet that uses reaktiv.

[buiapp]

@Kaleidosium Could you help me to understand how you would use the watcher mechanism? Is it for custom scheduling or performance reasons? Just trying to collect different views to get some nuances before implementing the API. :) Thanks!

[Kaleidosium]

I've had some issues with scheduling while building the library, so I think the Watcher mechanism could help with that.