Signal API

reaktiv.Signal

A reactive writable signal that notifies dependents when its value changes.

Signal is the core building block in reaktiv. It creates a container for values that can change over time and automatically notify effects and computed signals that depend on it.

Parameters:

Name	Type	Description	Default
value	T	The initial value of the signal	required
equal	Optional[Callable[[T, T], bool]]	Optional custom equality function to determine if two values should be considered equal. By default, identity (is) is used.	None

Examples:

Basic usage:

from reaktiv import Signal

# Create a signal with an initial value
counter = Signal(0)

# Get the current value
value = counter()  # 0

# Set a new value
counter.set(5)

# Update using a function
counter.update(lambda x: x + 1)  # Now 6

Custom equality:

from reaktiv import Signal

# Custom equality function for dictionaries
def dict_equal(a, b):
    return isinstance(a, dict) and isinstance(b, dict) and a == b

user = Signal({"name": "Alice", "age": 30}, equal=dict_equal)

# This won't trigger updates (same key-value pairs)
user.set({"name": "Alice", "age": 30})

# This will trigger updates (different age)
user.set({"name": "Alice", "age": 31})

Readonly wrapper:

from reaktiv import Signal

# Internal writable signal
_counter = Signal(0)

# Expose readonly view
counter = _counter.as_readonly()

def increment():
    _counter.update(lambda x: x + 1)

print(counter())  # 0
increment()
print(counter())  # 1
# counter.set(5)  # Error: ReadonlySignal has no 'set' method

__init__(value, *, equal=None)

Initialize a new Signal with an initial value.

Parameters:

Name	Type	Description	Default
value	T	The initial value of the signal	required
equal	Optional[Callable[[T, T], bool]]	Optional custom equality function. If provided, it will be used to determine if a new value is different from the current value. If not provided, identity comparison (is) is used.	None

get()

Get the current value of the signal.

When called within an active effect or computed signal, it establishes a dependency relationship, so changes to this signal will notify the dependent.

Returns:

Type	Description
T	The current value of the signal

Examples:

counter = Signal(42)
value = counter.get()  # 42
# Or use the callable syntax
value = counter()      # 42

set(new_value)

Set a new value for the signal and notify subscribers if it changed.

A notification is triggered only if the new value is considered different from the current value. By default, identity comparison (is) is used unless a custom equality function was provided.

Parameters:

Name	Type	Description	Default
new_value	T	The new value to set	required

Raises:

Type	Description
RuntimeError	If called from within a ComputeSignal computation

Examples:

counter = Signal(0)
counter.set(5)
print(counter())  # 5

# Setting the same value (by identity) won't trigger updates
obj = {"x": 1}
signal = Signal(obj)
signal.set(obj)  # No notification
signal.set({"x": 1})  # Notification (different object)

update(update_fn)

Atomically update the signal using a function of its current value.

This method is useful for updates that depend on the current value, such as incrementing a counter or modifying an object.

Parameters:

Name	Type	Description	Default
update_fn	Callable[[T], T]	A function that takes the current value and returns the new value	required

Examples:

counter = Signal(0)
counter.update(lambda x: x + 1)
print(counter())  # 1

counter.update(lambda x: x * 2)
print(counter())  # 2

# Works with any type
name = Signal("Alice")
name.update(lambda s: s.upper())
print(name())  # "ALICE"

as_readonly()

Return a readonly wrapper that exposes only read access to this signal.

Useful for encapsulation when you want to share a value but prevent external code from mutating it.

Returns:

Type	Description
'ReadonlySignal[T]'	A ReadonlySignal that wraps this signal

Examples:

# Keep internal signal private
_counter = Signal(0)

# Expose readonly view
counter = _counter.as_readonly()

def increment():
    _counter.update(lambda x: x + 1)

print(counter())  # 0
increment()
print(counter())  # 1
# counter.set(5)  # AttributeError: 'ReadonlySignal' has no attribute 'set'

reaktiv.ReadonlySignal

A readonly wrapper around a Signal that prevents modification.

ReadonlySignal provides only read access (get() and __call__()) to the underlying signal, preventing direct modification. Useful for encapsulation and API design.

Note

You typically don't create ReadonlySignal directly. Use Signal.as_readonly() instead.

Examples:

from reaktiv import Signal

_counter = Signal(0)
counter = _counter.as_readonly()

# Can read
print(counter())  # 0

# Cannot write
# counter.set(5)  # AttributeError

get()

Get the current value of the underlying signal.

Returns:

Type	Description
T	The current value