Computed Signal API

reaktiv.Computed(func=None, /, *, equal=None)

Computed(func: Callable[[], T]) -> ComputeSignal[T]

Computed(func: Callable[[], T], /, *, equal: Callable[[T, T], bool]) -> ComputeSignal[T]

Computed(*, equal: Callable[[T, T], bool]) -> Callable[[Callable[[], T]], ComputeSignal[T]]

Create a computed signal that derives its value from other signals.

Can be used as a function or as a decorator (with or without parameters).

Parameters:

Name	Type	Description	Default
func	Optional[Callable[[], T]]	The computation function (when used as factory or decorator without params)	None
equal	Optional[Callable[[T, T], bool]]	Optional custom equality function for change detection	None

Returns:

Type	Description
Union[ComputeSignal[T], Callable[[Callable[[], T]], ComputeSignal[T]]]	A ComputeSignal instance or a decorator function

Examples:

As a function:

from reaktiv import Signal, Computed

x = Signal(10)
result = Computed(lambda: x() * 2)
print(result())  # 20

As a decorator (no parameters):

from reaktiv import Signal, Computed

price = Signal(100)
quantity = Signal(2)

@Computed
def total():
    return price() * quantity()

print(total())  # 200

As a decorator (with custom equality):

from reaktiv import Signal, Computed

items = Signal([1, 2, 3])

@Computed(equal=lambda a, b: a == b)
def sorted_items():
    return sorted(items())

print(sorted_items())  # [1, 2, 3]

reaktiv.ComputeSignal

A computed signal that derives its value from other signals.

ComputeSignal automatically tracks dependencies on other signals and recomputes its value when any dependency changes. Computations are lazy and cached - they only run when accessed and dependencies have changed.

Parameters:

Name	Type	Description	Default
compute_fn	Callable[[], T]	A function that computes the signal's value from other signals	required
equal	Optional[Callable[[T, T], bool]]	Optional custom equality function for change detection	None

Examples:

Basic computed signal:

from reaktiv import Signal, Computed

first_name = Signal("John")
last_name = Signal("Doe")

# Create computed signal
full_name = Computed(lambda: f"{first_name()} {last_name()}")

print(full_name())  # "John Doe"

first_name.set("Jane")
print(full_name())  # "Jane Doe"

Lazy computation:

from reaktiv import Signal, Computed

x = Signal(10)
y = Signal(20)

def expensive_computation():
    print("Computing...")
    return x() * y()

result = Computed(expensive_computation)

# Nothing happens yet - computation is lazy

# First access - computation runs
print(result())  # Prints: "Computing..." then "200"

# Second access - no computation (cached)
print(result())  # Just prints "200"

# Change a dependency
x.set(5)

# Next access will recompute
print(result())  # Prints: "Computing..." then "100"

Decorator pattern:

from reaktiv import Signal, Computed

price = Signal(100)
quantity = Signal(2)

@Computed
def total():
    return price() * quantity()

print(total())  # 200

Error handling:

from reaktiv import Signal, Computed

x = Signal(10)

# Computed signal with potential error
result = Computed(lambda: 100 / x())

print(result())  # 10.0 (100 / 10)

# Set x to 0, causing division by zero
x.set(0)

# Exception is propagated to caller
try:
    print(result())
except ZeroDivisionError as e:
    print(f"Error: {e}")

# After fixing, computation works again
x.set(5)
print(result())  # 20.0 (100 / 5)

Note

Computed signals are lazy - they only compute when accessed and cache the result until dependencies change. When a computation raises an exception, it is propagated to the caller for flexible error handling.