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. 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"

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.