A very naive implementation of reactive values and computed values

signal.py

'''
A very naive implementation of reactive values and computed values, inspired by
https://github.com/tc39/proposal-signals
https://github.com/buiapp/reaktiv
Most of the features described in the proposal are NOT implemented.

.. code-block::

    from signal import State, computed

    count = State(0)

    @computed
    def divisible_by_3(count=count):
        return count() % 3 == 0

    @computed
    def contains_the_digit_3(count=count):
        return "3" in str(count())

    @computed
    def nabeatsu(c1=divisible_by_3, c2=contains_the_digit_3):
        return "😜" if c1() or c2() else "😑"

    count.value = 11
    assert nabeatsu() == "😑"
    count.value = 12
    assert nabeatsu() == "😜"
    count.value = 13
    assert nabeatsu() == "😜"
    count.value = 14
    assert nabeatsu() == "😑"
'''
__all__ = ("State", "Computed", "computed", )
from collections.abc import Callable
import operator
from functools import partial


class State:
    def __init__(self, initial_value, /, *, equal=operator.eq):
        self._value = initial_value
        self._dependants: list["Computed"] = []
        self.add_dependant = self._dependants.append
        self.equal = equal

    def set(self, new_value):
        if self.equal(new_value, self._value):
            return
        self._value = new_value
        # NOTE: Computedではなくmethodの方を保持しておけば属性参照の手間を省けるだろう
        for d in self._dependants:
            d.invalidate_cache()

    def get(self):
        return self._value

    __call__ = get
    value = property(get, set)


class Computed:
    def __init__(self, compute: Callable, /, *, equal=operator.eq):
        self._value = None
        self._is_fresh = False
        self._dependants: list["Computed"] = []
        self.add_dependant = self._dependants.append
        self._compute = compute
        self.equal = equal
        _register_computed_as_dependant(compute, self)

    def invalidate_cache(self):
        self._is_fresh = False
        # NOTE: Computedではなくmethodの方を保持しておけば属性参照の手間を省けるだろう
        for d in self._dependants:
            d.invalidate_cache()

    def get(self):
        if self._is_fresh:
            return self._value
        old_value = self._value
        new_value = self._compute()
        self._is_fresh = True
        if self.equal(new_value, old_value):
            return old_value
        self._value = new_value
        return new_value

    __call__ = get
    value = property(get)


_DEPENDENCY_MARK = r"_Oh_yes_I_can_be_a_source_of_other_computations"
for cls in (State, Computed):
    setattr(cls, _DEPENDENCY_MARK, None)


# TODO: Support `functools.partial`-ed functions
def _register_computed_as_dependant(compute: Callable, c: Computed, MARK=_DEPENDENCY_MARK, hasattr=hasattr):
    for defaultvalue in compute.__defaults__:
        if hasattr(defaultvalue, MARK):
            defaultvalue.add_dependant(c)


def computed(compute: Callable=None, /, *, equal=operator.eq):
    if compute is None:
        return partial(computed, equal=equal)
    return Computed(compute, equal=equal)