Core

signified.core

Core reactive programming functionality.

Observer

Bases: Protocol

update

update()

Variable

Bases: ABC, _HasValue[Y], ReactiveMixIn[T]

An abstract base class for reactive values.

A reactive value is an object that can be observed by observer for changes and can notify observers when its value changes. This class implements both the observer and observable patterns.

This class implements both the observer and observable pattern.

Subclasses should implement the update method.

Attributes:

Name	Type	Description
_observers	list[Observer]	List of observers subscribed to this variable.

value property

value

subscribe

subscribe(observer)

Subscribe an observer to this variable.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to subscribe.	required

unsubscribe

unsubscribe(observer)

Unsubscribe an observer from this variable.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to unsubscribe.	required

observe

observe(items)

Subscribe the observer (self) to all items that are Observable.

This method handles arbitrarily nested iterables.

Parameters:

Name	Type	Description	Default
items	Any	A single item, an iterable, or a nested structure of items to potentially subscribe to.	required

Returns:

Type	Description
Self	self

unobserve

unobserve(items)

Unsubscribe the observer (self) from all items that are Observable.

Parameters:

Name	Type	Description	Default
items	Any	A single item or an iterable of items to potentially unsubscribe from.	required

Returns:

Type	Description
Self	self

notify

notify()

Notify all observers by calling their update method.

update abstractmethod

update()

Update method to be overridden by subclasses.

Raises:

Type	Description
NotImplementedError	If not overridden by a subclass.

add_name

add_name(name)

bool

bool()

Return a reactive value for the boolean value of self.

Note

__bool__ cannot be implemented to return a non-bool, so it is provided as a method.

Returns:

Type	Description
Computed[bool]	A reactive value for bool(self.value).

Example

>>> s = Signal(1)
>>> result = s.bool()
>>> result.value
True
>>> s.value = 0
>>> result.value
False

contains

contains(other)

Return a reactive value for whether other is in self.

Parameters:

Name	Type	Description	Default
other	Any	The value to check for containment.	required

Returns:

Type	Description
Computed[bool]	A reactive value for other in self.value.

Example

>>> s = Signal([1, 2, 3, 4])
>>> result = s.contains(3)
>>> result.value
True
>>> s.value = [5, 6, 7, 8]
>>> result.value
False

is_not

is_not(other)

Return a reactive value for whether self is not other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value is not other.

Example

>>> s = Signal(10)
>>> other = None
>>> result = s.is_not(other)
>>> result.value
True
>>> s.value = None
>>> result.value
False

eq

eq(other)

Return a reactive value for whether self equals other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value == other.

Note

We can't overload __eq__ because it interferes with basic Python operations.

Example

>>> from signified import Signal
>>> s = Signal(10)
>>> result = s.eq(10)
>>> result.value
True
>>> s.value = 25
>>> result.value
False

where

where(a, b)

Return a reactive value for a if self is True, else b.

Parameters:

Name	Type	Description	Default
a	HasValue[A]	The value to return if self is True.	required
b	HasValue[B]	The value to return if self is False.	required

Returns:

Type	Description
Computed[A | B]	A reactive value for a if self.value else b.

Example

>>> condition = Signal(True)
>>> result = condition.where("Yes", "No")
>>> result.value
'Yes'
>>> condition.value = False
>>> result.value
'No'

Signal

Bases: Variable[NestedValue[T], T]

A container that holds a reactive value.

value property writable

value

Get or set the current value.

at

at(value)

Temporarily set the signal to a given value within a context.

update

update()

Update the signal and notify subscribers.

notify

notify()

Notify all observers by calling their update method.

bool

bool()

Return a reactive value for the boolean value of self.

Note

__bool__ cannot be implemented to return a non-bool, so it is provided as a method.

Returns:

Type	Description
Computed[bool]	A reactive value for bool(self.value).

Example

>>> s = Signal(1)
>>> result = s.bool()
>>> result.value
True
>>> s.value = 0
>>> result.value
False

contains

contains(other)

Return a reactive value for whether other is in self.

Parameters:

Name	Type	Description	Default
other	Any	The value to check for containment.	required

Returns:

Type	Description
Computed[bool]	A reactive value for other in self.value.

Example

>>> s = Signal([1, 2, 3, 4])
>>> result = s.contains(3)
>>> result.value
True
>>> s.value = [5, 6, 7, 8]
>>> result.value
False

is_not

is_not(other)

Return a reactive value for whether self is not other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value is not other.

Example

>>> s = Signal(10)
>>> other = None
>>> result = s.is_not(other)
>>> result.value
True
>>> s.value = None
>>> result.value
False

eq

eq(other)

Return a reactive value for whether self equals other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value == other.

Note

We can't overload __eq__ because it interferes with basic Python operations.

Example

>>> from signified import Signal
>>> s = Signal(10)
>>> result = s.eq(10)
>>> result.value
True
>>> s.value = 25
>>> result.value
False

where

where(a, b)

Return a reactive value for a if self is True, else b.

Parameters:

Name	Type	Description	Default
a	HasValue[A]	The value to return if self is True.	required
b	HasValue[B]	The value to return if self is False.	required

Returns:

Type	Description
Computed[A | B]	A reactive value for a if self.value else b.

Example

>>> condition = Signal(True)
>>> result = condition.where("Yes", "No")
>>> result.value
'Yes'
>>> condition.value = False
>>> result.value
'No'

subscribe

subscribe(observer)

Subscribe an observer to this variable.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to subscribe.	required

unsubscribe

unsubscribe(observer)

Unsubscribe an observer from this variable.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to unsubscribe.	required

observe

observe(items)

Subscribe the observer (self) to all items that are Observable.

This method handles arbitrarily nested iterables.

Parameters:

Name	Type	Description	Default
items	Any	A single item, an iterable, or a nested structure of items to potentially subscribe to.	required

Returns:

Type	Description
Self	self

unobserve

unobserve(items)

Unsubscribe the observer (self) from all items that are Observable.

Parameters:

Name	Type	Description	Default
items	Any	A single item or an iterable of items to potentially unsubscribe from.	required

Returns:

Type	Description
Self	self

add_name

add_name(name)

Computed

Bases: Variable[T, T]

A reactive value defined by a function.

f instance-attribute

f = f

value property

value

Get the current value.

update

update()

Update the value by re-evaluating the function.

notify

notify()

Notify all observers by calling their update method.

bool

bool()

Return a reactive value for the boolean value of self.

Note

__bool__ cannot be implemented to return a non-bool, so it is provided as a method.

Returns:

Type	Description
Computed[bool]	A reactive value for bool(self.value).

Example

>>> s = Signal(1)
>>> result = s.bool()
>>> result.value
True
>>> s.value = 0
>>> result.value
False

contains

contains(other)

Return a reactive value for whether other is in self.

Parameters:

Name	Type	Description	Default
other	Any	The value to check for containment.	required

Returns:

Type	Description
Computed[bool]	A reactive value for other in self.value.

Example

>>> s = Signal([1, 2, 3, 4])
>>> result = s.contains(3)
>>> result.value
True
>>> s.value = [5, 6, 7, 8]
>>> result.value
False

is_not

is_not(other)

Return a reactive value for whether self is not other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value is not other.

Example

>>> s = Signal(10)
>>> other = None
>>> result = s.is_not(other)
>>> result.value
True
>>> s.value = None
>>> result.value
False

eq

eq(other)

Return a reactive value for whether self equals other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value == other.

Note

We can't overload __eq__ because it interferes with basic Python operations.

Example

>>> from signified import Signal
>>> s = Signal(10)
>>> result = s.eq(10)
>>> result.value
True
>>> s.value = 25
>>> result.value
False

where

where(a, b)

Return a reactive value for a if self is True, else b.

Parameters:

Name	Type	Description	Default
a	HasValue[A]	The value to return if self is True.	required
b	HasValue[B]	The value to return if self is False.	required

Returns:

Type	Description
Computed[A | B]	A reactive value for a if self.value else b.

Example

>>> condition = Signal(True)
>>> result = condition.where("Yes", "No")
>>> result.value
'Yes'
>>> condition.value = False
>>> result.value
'No'

subscribe

subscribe(observer)

Subscribe an observer to this variable.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to subscribe.	required

unsubscribe

unsubscribe(observer)

Unsubscribe an observer from this variable.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to unsubscribe.	required

observe

observe(items)

Subscribe the observer (self) to all items that are Observable.

This method handles arbitrarily nested iterables.

Parameters:

Name	Type	Description	Default
items	Any	A single item, an iterable, or a nested structure of items to potentially subscribe to.	required

Returns:

Type	Description
Self	self

unobserve

unobserve(items)

Unsubscribe the observer (self) from all items that are Observable.

Parameters:

Name	Type	Description	Default
items	Any	A single item or an iterable of items to potentially unsubscribe from.	required

Returns:

Type	Description
Self	self

add_name

add_name(name)

signified.utils

Utility functions for reactive programming.

InstanceMethod module-attribute

InstanceMethod = Callable[Concatenate[Any, P], T]

ReactiveMethod module-attribute

ReactiveMethod = Callable[Concatenate[Any, P], 'Computed[T]']

unref

unref(value)

Dereference a value, resolving any nested reactive variables.

deep_unref

deep_unref(value)

Recursively unref values potentially within containers.

computed

computed(func)

Decorate the function to return a reactive value.

reactive_method

reactive_method(*dep_names)

Decorate the method to return a reactive value.

as_signal

as_signal(val)

Convert a value to a Signal if it's not already a reactive value.

signified.ops

Reactive operators and methods for reactive values.

ReactiveMixIn

Bases: Generic[T]

Methods for easily creating reactive values.

value property

value

The current value of the reactive object.

notify

notify()

Notify all observers by calling their update method.

bool

bool()

Return a reactive value for the boolean value of self.

Note

__bool__ cannot be implemented to return a non-bool, so it is provided as a method.

Returns:

Type	Description
Computed[bool]	A reactive value for bool(self.value).

Example

>>> s = Signal(1)
>>> result = s.bool()
>>> result.value
True
>>> s.value = 0
>>> result.value
False

contains

contains(other)

Return a reactive value for whether other is in self.

Parameters:

Name	Type	Description	Default
other	Any	The value to check for containment.	required

Returns:

Type	Description
Computed[bool]	A reactive value for other in self.value.

Example

>>> s = Signal([1, 2, 3, 4])
>>> result = s.contains(3)
>>> result.value
True
>>> s.value = [5, 6, 7, 8]
>>> result.value
False

is_not

is_not(other)

Return a reactive value for whether self is not other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value is not other.

Example

>>> s = Signal(10)
>>> other = None
>>> result = s.is_not(other)
>>> result.value
True
>>> s.value = None
>>> result.value
False

eq

eq(other)

Return a reactive value for whether self equals other.

Parameters:

Name	Type	Description	Default
other	Any	The value to compare against.	required

Returns:

Type	Description
Computed[bool]	A reactive value for self.value == other.

Note

We can't overload __eq__ because it interferes with basic Python operations.

Example

>>> from signified import Signal
>>> s = Signal(10)
>>> result = s.eq(10)
>>> result.value
True
>>> s.value = 25
>>> result.value
False

where

where(a, b)

Return a reactive value for a if self is True, else b.

Parameters:

Name	Type	Description	Default
a	HasValue[A]	The value to return if self is True.	required
b	HasValue[B]	The value to return if self is False.	required

Returns:

Type	Description
Computed[A | B]	A reactive value for a if self.value else b.

Example

>>> condition = Signal(True)
>>> result = condition.where("Yes", "No")
>>> result.value
'Yes'
>>> condition.value = False
>>> result.value
'No'