Effect API

reaktiv.Effect

A reactive effect that automatically tracks signal dependencies and re-runs when they change.

Effect creates a side effect function that runs immediately and re-runs whenever any signal it depends on changes. For synchronous effects outside a batch, every successful dependency update that changes a value triggers one effect execution. Updates inside batch() are intentionally coalesced and run once after the outermost batch completes. It supports optional cleanup logic.

Lifecycle and Memory Management

Effects are weakly referenced to prevent memory leaks. This means:

• Without a stored reference: The Effect will be garbage collected immediately and stop responding to signal changes. Cleanup functions will run automatically during garbage collection.

• With a stored reference: The Effect persists and continues reacting to changes. Call dispose() explicitly when done to ensure immediate cleanup.

Best practices: - Store Effect references: self.effect = Effect(...) - Call dispose() explicitly: self.effect.dispose() - Cleanup functions run automatically on GC, but prefer explicit dispose()

Cleanup Functions

Effects can register cleanup logic to run when: 1. The effect reruns (cleanup from previous run) 2. dispose() is called explicitly 3. The Effect is garbage collected (automatic)

Register cleanup by returning a function or using the on_cleanup parameter.

Note

Async functions are supported but still experimental and may not behave as expected in all scenarios.

Parameters:

Name	Type	Description	Default
func	EffectFn	The effect function to run. Can be sync or async (experimental). May optionally accept an on_cleanup callback parameter for registering cleanup logic, or return a cleanup function.	required

Examples:

Basic effect:

from reaktiv import Signal, Effect

counter = Signal(0)

# Effect runs immediately and on every change
# Must retain reference to prevent garbage collection
effect = Effect(lambda: print(f"Counter: {counter()}"))
# Prints: "Counter: 0"

counter.set(1)
# Prints: "Counter: 1"

counter.set(2)
# Prints: "Counter: 2"

Effect with cleanup:

from reaktiv import Signal, Effect

user_id = Signal(1)

def subscribe_to_user():
    uid = user_id()
    print(f"Subscribing to user {uid}")

    # Return cleanup function
    def cleanup():
        print(f"Unsubscribing from user {uid}")
    return cleanup

effect = Effect(subscribe_to_user)
# Prints: "Subscribing to user 1"

user_id.set(2)
# Prints: "Unsubscribing from user 1"
# Prints: "Subscribing to user 2"

effect.dispose()
# Prints: "Unsubscribing from user 2"

Effect with on_cleanup parameter:

from reaktiv import Signal, Effect

enabled = Signal(True)

def my_effect(on_cleanup):
    if enabled():
        print("Starting...")
        on_cleanup(lambda: print("Stopping..."))

effect = Effect(my_effect)
# Prints: "Starting..."

enabled.set(False)
# Prints: "Stopping..."

Manual disposal:

from reaktiv import Signal, Effect

count = Signal(0)

effect = Effect(lambda: print(count()))
# Prints: 0

count.set(1)
# Prints: 1

effect.dispose()

count.set(2)
# No print - effect is disposed

dispose()

Stop the effect and prevent it from running again.

This method: - Marks the effect as disposed - Unsubscribes from all signal dependencies - Runs any pending cleanup functions - Cancels any in-progress async tasks

After calling dispose(), the effect will no longer react to signal changes.

Examples:

from reaktiv import Signal, Effect

counter = Signal(0)

effect = Effect(lambda: print(f"Count: {counter()}"))
# Prints: "Count: 0"

counter.set(1)
# Prints: "Count: 1"

effect.dispose()

counter.set(2)
# No output - effect is disposed

__del__()

Run cleanup function when Effect is garbage collected.

This ensures that cleanup functions (e.g., closing connections, canceling subscriptions) are executed even if dispose() is not called explicitly.

Note: This is called automatically by Python's garbage collector when the Effect has no more references. For deterministic cleanup, prefer calling dispose() explicitly.