Observable Operators¶

Special operators for composing and transforming observables.

FynX Operators - observable operator implementations and mixins¶

FynX overloads Python operators so reactive code reads like an expression rather than a method chain. The operators handle dependency tracking and updates behind the scenes.

Three mixins provide this:

OperatorMixin gives every observable type the core operators (+, >>, &, @, ~, |).
TupleMixin adds tuple-like behavior to merged observables - iteration, indexing, length.
ValueMixin gives ObservableValue transparent value access, so reactive attributes behave like regular values while keeping reactive capabilities.

Operator Semantics¶

Transform (>>): Apply functions to create derived values

from fynx.observable import Observable

counter = Observable("counter", 5)
doubled = counter >> (lambda x: x * 2)
print(doubled.value)  # 10

Boolean AND (&): Combine boolean conditions

authenticated = observable(True)
connected = observable(True)
loading = observable(False)
ready = authenticated & connected & ~loading

Gate (@): Only emit values when conditions are met

data = Observable("data", "hello")
is_ready = Observable("ready", False)
filtered = data @ is_ready  # Only emits when is_ready is True

Combine (+): Merge multiple observables into tuples

x = Observable("x", 1)
y = Observable("y", 2)
z = Observable("z", 3)
coordinates = x + y + z
print(coordinates.value)  # (1, 2, 3)

These operators compose to create complex reactive pipelines:

result = ((x + y) >> (lambda a, b: a + b)) @ (total >> (lambda t: t > 10))

Implementation Architecture¶

Operators delegate to OperationsMixin rather than implementing logic directly, which keeps imports lazy and avoids circular-import issues. obs >> func calls __rshift__, which delegates to obs.then(func); then creates a computed observable through _create_computed.

Transformation functions receive unpacked tuple values as separate arguments for merged observables, and a single argument for regular ones.

Performance Characteristics¶

Operators create computed or conditional observables that evaluate lazily, recalculating only when accessed after a dependency changes. Chained operators fuse into a single composed function rather than creating intermediate objects, and they reuse existing infrastructure instead of new classes, keeping memory overhead low.

Common Patterns¶

Data Processing Pipeline:

from fynx import observable

raw_data = observable([1, -2, 3, -4, 5])
processed = (raw_data
    >> (lambda d: [x for x in d if x > 0])  # Filter positive values
    >> (lambda d: sorted(d))                # Sort results
    >> (lambda d: sum(d) / len(d) if d else 0))  # Calculate average
print(processed.value)  # 3.0

Conditional UI Updates:

user_input = observable("")
is_valid = user_input >> (lambda s: len(s) >= 3)
has_input = user_input >> bool
show_error = user_input @ (has_input & ~is_valid)  # Show error once invalid input exists

Reactive Calculations:

price = observable(10.0)
quantity = observable(1)
tax_rate = observable(0.08)

subtotal = (price + quantity) >> (lambda p, q: p * q)
tax = (subtotal + tax_rate) >> (lambda s, rate: s * rate)
total = (subtotal + tax) >> (lambda s, t: s + t)
print(total.value)  # 10.8

Error Handling¶

Errors from transformation functions propagate normally rather than being swallowed. Invalid operator usage raises TypeError with a descriptive message. A transform that reads .value or calls .set() on an observable raises TransformPurityError, with a hint to combine inputs explicitly or move the effect to a subscription. Circular dependencies are caught during .set() and raise RuntimeError before they can loop.

Best Practices¶

Keep transformation functions pure: use only the values passed in as arguments, with no side effects and no hidden observable reads - combine every reactive input first with + / .alongside(). Prefer named functions over long lambdas for complex operations, and break long chains into intermediate variables for clarity.

OperatorMixin ¶

Mixin class providing common reactive operators for observable classes.

Consolidates the operator overloading logic (__add__, __rshift__, __and__, __matmul__, __or__, __invert__) that would otherwise be duplicated across every observable class.

Classes inheriting from this mixin get automatic support for: - Merging with + operator - Transformation with >> operator - Boolean AND with & operator - Conditional gating with @ operator - Boolean negation with ~ operator

This mixin should be used by classes that represent reactive values and need to support reactive composition operations.

add ¶

__add__(other: ObservableOperand[U]) -> 'MergedObservable[T, U]'

__add__(other: Any) -> Any

__add__(other)

Combine observables, or delegate plain additions to the wrapped value.

Observable-like operands create a merged observable containing an ordered tuple of both values. Plain Python values use the underlying value's own +, so items + [new_item] works for immutable updates.

Parameters:

Name	Type	Description	Default
`other`	`Any`	Another Observable to combine with, or a plain value to add	required

Returns:

Type	Description
`Any`	A MergedObservable for observable operands; otherwise the plain
`Any`	Python addition result.

and ¶

__and__(other)

Create a total boolean AND observable using the & operator.

This creates a computed boolean observable that is True when both operands are truthy and False otherwise.

Parameters:

Name	Type	Description	Default
`other`	`BooleanOperand`	Another observable-like boolean value	required

Returns:

Type	Description
`'Observable[bool]'`	A computed Observable[bool] containing the AND result.

invert ¶

__invert__()

Create a negated boolean observable using the ~ operator.

This creates a computed observable that returns the logical negation of the current boolean value, useful for creating inverse conditions. The negation updates automatically when the source changes.

Returns:

Type	Description
`'Observable[bool]'`	A computed Observable[bool] with negated boolean value

matmul ¶

__matmul__(condition)

Gate this observable using the @ operator for conditional reactivity.

Creates a ConditionalObservable that only emits this observable's values while every condition is True.

Parameters:

Name	Type	Description	Default
`condition`	`ConditionOperand[T]`	A boolean Observable, callable, or compound condition	required

Returns:

Type	Description
`'ConditionalObservable[T]'`	A ConditionalObservable that gates values based on the condition.

or ¶

__or__(other)

Create a logical OR condition using the | operator.

This creates a computed boolean observable that is True when either operand is truthy and False otherwise. The operation combines boolean observables with logical disjunction.

Parameters:

Name	Type	Description	Default
`other`	`ObservableOperand[Any]`	Another boolean observable to OR with	required

Returns:

Type	Description
`'Observable[bool]'`	A computed Observable[bool] containing the OR result.

radd ¶

__radd__(other: ObservableOperand[U]) -> 'MergedObservable[U, T]'

__radd__(other: Any) -> Any

__radd__(other)

Support right-side addition for merging observables.

This enables expressions like other + self to work correctly, ensuring that merged observables can be chained properly. Python calls this method when the left operand doesn't support __add__.

Parameters:

Name	Type	Description	Default
`other`	`Any`	Another Observable to combine with	required

Returns:

Type	Description
`Any`	A MergedObservable containing both values as a tuple

rshift ¶

__rshift__(func)

Apply a transformation function using the >> operator to create computed observables.

This implements the functorial map operation over observables, allowing you to transform observable values through pure functions while preserving reactivity. The operation satisfies the functor laws: identity and composition preservation.

Parameters:

Name	Type	Description	Default
`func`	`Callable[[T], U]`	A pure function to apply to the observable's value(s)	required

Returns:

Type	Description
`'Observable[U]'`	A new computed Observable containing the transformed values

TupleMixin ¶

Mixin class providing tuple-like operators for merged observables.

Adds iteration, indexing, and length operators so a MergedObservable behaves like a tuple of its component values.

Classes inheriting from this mixin get automatic support for: - Iteration with for item in merged: - Length with len(merged) - Indexing with merged[0], merged[-1], etc. - Setting values by index with merged[0] = new_value

getitem ¶

__getitem__(index)

Allow indexing into the merged observable like a tuple.

iter ¶

__iter__()

Allow iteration over the tuple value.

len ¶

__len__()

Return the number of combined observables.

setitem ¶

__setitem__(index, value)

Allow setting values by index, updating the corresponding source observable.

ValueMixin ¶

Mixin class providing value wrapper operators for ObservableValue.

Adds the magic methods (equality, string conversion, iteration, indexing) that let an observable value behave like its underlying value in most Python contexts, alongside the reactive operators.

Classes inheriting from this mixin get automatic support for: - Value-like behavior (equality, string conversion, etc.) - Reactive operators (add, and, matmul, or, invert, rshift) - Transparent access to the wrapped observable

add ¶

__add__(other: ObservableOperand[U]) -> 'MergedObservable[T, U]'

__add__(other: list[V]) -> list[V]

__add__(other: Any) -> Any

__add__(other)

Support merging observables with + operator.

and ¶

__and__(other)

Support boolean AND with & operator.

invert ¶

__invert__()

Support negating conditions with ~ operator.

matmul ¶

__matmul__(condition)

Support conditional gating with @ operator.

or ¶

__or__(other)

Support logical OR conditions with | operator.

radd ¶

__radd__(other: ObservableOperand[U]) -> 'MergedObservable[U, T]'

__radd__(other: list[V]) -> list[V]

__radd__(other: Any) -> Any

__radd__(other)

Support right-side addition for merging observables.

rshift ¶

__rshift__(func)

Support computed observables with >> operator.

all ¶

all(*others)

Combine this observable value with conditions using AND.

alongside ¶

alongside(other)

Combine this observable value with another observable-like value.

either ¶

either(other)

Combine this observable value with another condition using OR.

negate ¶

negate()

Negate this observable value as a boolean condition.

requiring ¶

requiring(*conditions)

Gate this observable value behind one or more conditions.

then ¶

then(func)

Transform this observable value with a typed natural-language method.

and_operator ¶

and_operator(obs, other)

Implement the & operator for total boolean AND observables.

Parameters:

Name	Type	Description	Default
`obs`	`'Observable[Any]'`	The first boolean-like observable.	required
`other`	`BooleanOperand`	Another boolean-like observable.	required

Returns:

Type	Description
`'Observable[bool]'`	A computed Observable[bool] that updates whenever either input changes.

matmul_operator ¶

matmul_operator(obs, condition)

Implement the @ operator for gated conditional observables.

rshift_operator ¶

rshift_operator(obs, func)

Implement the >> operator for pure reactive transforms.

Sequential pure transforms compose over the original source when no observed boundary requires an intermediate notification: obs >> f >> g behaves like mapping through f then g, while the runtime represents it as one composed transform.

For merged observables (created with +), the function receives multiple arguments corresponding to the tuple values. For single observables, it receives one argument.

Parameters:

Name	Type	Description	Default
`obs`	`'Observable[T]'`	The source observable(s) to transform. Can be a single Observable or a MergedObservable (from `+` operator).	required
`func`	`Callable[[T], U]`	A pure function that transforms the observable value(s). For merged observables, receives unpacked tuple values as separate arguments.	required

Returns:

Type	Description
`'Observable[U]'`	A new computed observable. Unobserved computed values are cached and
`'Observable[U]'`	version-validated lazily; observed values are maintained eagerly enough
`'Observable[U]'`	to deliver subscriber notifications.

Examples:

from fynx.observable import Observable

# Single observable with automatic optimization
counter = Observable("counter", 5)
result = counter >> (lambda x: x * 2) >> (lambda x: x + 10) >> str
# Automatically optimized to single fused computation

# Repeated products reuse the same ordered product while live
width = Observable("width", 10)
height = Observable("height", 20)
area = (width + height) >> (lambda w, h: w * h)
volume = (width + height + Observable("depth", 5)) >> (lambda w, h, d: w * h * d)

Performance

Transform fusion: reduces intermediate reactive-node overhead
Canonical products: reuses repeated ordered products while live
Version invalidation: refreshes lazy values only when inputs changed
Demand frontier: subscribers create eager maintenance where needed

Observable Operators¶

FynX Operators - observable operator implementations and mixins¶

Operator Semantics¶

Implementation Architecture¶

Performance Characteristics¶

Common Patterns¶

Error Handling¶

Best Practices¶

See Also¶

OperatorMixin ¶

__add__ ¶

__and__ ¶

__invert__ ¶

__matmul__ ¶

__or__ ¶

__radd__ ¶

__rshift__ ¶

TupleMixin ¶

__getitem__ ¶

__iter__ ¶

__len__ ¶

__setitem__ ¶

ValueMixin ¶

__add__ ¶

__and__ ¶

__invert__ ¶

__matmul__ ¶

__or__ ¶

__radd__ ¶

__rshift__ ¶

all ¶

alongside ¶

either ¶

negate ¶

requiring ¶

then ¶

and_operator ¶

matmul_operator ¶

rshift_operator ¶

add ¶

and ¶

invert ¶

matmul ¶

or ¶

radd ¶

rshift ¶

getitem ¶

iter ¶

len ¶

setitem ¶

add ¶

and ¶

invert ¶

matmul ¶

or ¶

radd ¶

rshift ¶