signal API

Use signal(...) to create writable reactive state in new code:

Editor (session: default) Run

from reaktiv import signal

count = signal(0)
count.update(lambda value: value + 1)

print(count())  # 1

Output Clear

Naming

Prefer signal(...) in application code. Signal is the concrete writable signal class and remains supported for existing class-style code, type references, and advanced use.

reaktiv.signal

Writable Signal implementation.

Migrated to Edge-based dependency tracking and minimal scheduler.

signal = Signal module-attribute

Lowercase factory-style spelling for user code that prefers function-based primitive creation. Equivalent to Signal.

ComputeSignal

Bases: Signal[T]

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 computed, signal

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

@computed
def full_name():
    return f"{first_name()} {last_name()}"

print(full_name())  # "John Doe"

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

Lazy computation:

from reaktiv import computed, signal

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 computed, signal

price = signal(100)
quantity = signal(2)

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

print(total())  # 200

Error handling:

from reaktiv import computed, signal

x = signal(10)

# Computed signal with potential error
@computed
def result():
    return 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.

ComputedDescriptor

Bases: PerInstanceDescriptor[ComputeSignal[T]], Generic[T]

Descriptor that creates one ComputeSignal per owner instance.

ReadonlySignal

Bases: Generic[T]

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

Signal

Bases: Generic[T]

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

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'

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}
value = signal(obj)
value.set(obj)  # No notification
value.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"

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}
value = signal(obj)
value.set(obj)  # No notification
value.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