reaktiv

A ReactiveModel is a class that groups related reactive state and behavior into a single unit. Each instance of a ReactiveModel owns its own independent graph of writable fields, computed values, linked state, effects, and resources.

Use ReactiveModel to represent application concepts like forms, stores, view models, workflows, or services.

Note on Naming Convention: When defining new models, use the lowercase primitive API: field(...), @computed, @linked, @effect, and @resource. Avoid using the uppercase class names (like Signal or Effect) in your model definitions.

from reaktiv import ReactiveModel, computed, effect, field


class ShoppingCart(ReactiveModel):
    unit_price = field(12.50)
    quantity = field(1)
    discount = field(0.0)

    @computed
    def subtotal(self) -> float: 
        return self.unit_price() * self.quantity()

    @computed
    def total(self) -> float: 
        return self.subtotal() * (1 - self.discount())

    @effect
    def show_total(self) -> None:
        print(f"{self.quantity()} item(s): ${self.total():.2f}")


cart = ShoppingCart()
cart.quantity.set(3)
cart.dispose()

To prevent unnecessary network calls, use @computed to derive the parameters for your resource. Reaktiv only triggers the loader when the value returned by the params function actually changes. If a dependency signal changes but the resulting parameter object remains identical, the request is deduplicated and not re-run.

import asyncio
from reaktiv import resource, signal, computed

async def fetch_data(params):
    return {"data": f"Result for {params.params['key']}"}

async def main():
    signal_a = signal(1)
    signal_b = signal(2)
    
    @computed
    def combined_key():
        return f"{signal_a()}-{signal_b()}"
    
    data_resource = resource(
        params=lambda: {"key": combined_key()},
        loader=fetch_data
    )
    
    await asyncio.sleep(0.2)
    
    # Changing signal_a to 1 (no change in combined_key) won't trigger a new request
    signal_a.set(1)
    await asyncio.sleep(0.2)
    
    # Changing signal_a to 3 will trigger a new request
    signal_a.set(3)
    await asyncio.sleep(0.2)
    data_resource.destroy()

await main()

reaktiv is built on three core primitives that form a reactive dependency graph:

1. signal: Stores changing state. You read the value by calling the signal as a function (e.g., my_signal()) and update it using .set(value) or .update(fn).
2. computed: Describes derived state. Dependencies are discovered automatically when the function is executed. Results are lazy (computed only when needed) and memoized (cached until dependencies change).
3. effect: Reacts to changes. An effect is a side effect that automatically re-runs whenever any signal or computed value read during its execution changes.

By using these, you can change a single base value and let the reactive graph handle all downstream updates automatically.

from reaktiv import computed, effect, signal

query = signal("")

@computed
def results():
    return search(query())

@computed
def summary():
    return f"{len(results())} matches"

def display_results():
    render(summary(), results())

display = effect(display_results)

query.set("python")

By default, reaktiv uses identity comparison (is) to detect changes. This means:

• Primitives (numbers, strings, booleans) work as expected.
• Mutable objects (lists, dicts) will not trigger updates if modified in-place. They only trigger updates if a new object instance is provided via .set().

To detect changes based on content rather than identity, provide a custom equality function to the equal parameter when creating a signal.

Implementation Strategies for Mutable Objects

1. Custom Equality: Pass a function to signal(..., equal=my_func) that compares values.
2. Immutable Updates: Instead of modifying an object in-place, always provide a new instance (e.g., using current + [item] for lists or {**current, key: val} for dicts).

from reaktiv import signal

# 1. Custom equality for lists
def list_equal(a, b):
    if len(a) != len(b):
        return False
    return all(a_item == b_item for a_item, b_item in zip(a, b))

items = signal([1, 2, 3], equal=list_equal)
items.set([1, 2, 3])  # Won't trigger updates (same values)
items.set([1, 2, 3, 4]) # Will trigger updates

# 2. Custom equality for dictionaries
def dict_equal(a, b):
    return a.keys() == b.keys() and all(a[k] == b[k] for k in a.keys())

config = signal({"theme": "dark"}, equal=dict_equal)
config.set({"theme": "dark"}) # Won't trigger updates

# 3. Immutable approach (Recommended)
items_immut = signal([1, 2, 3])
items_immut.update(lambda current: current + [4]) # Triggers update by creating new list

For complex state management, inherit from ReactiveModel. This allows you to define state using field and derived state using @computed within a class structure.

• field(factory=...): Defines a reactive field within the model. Use .set() or .update() to modify it.
• @computed: Defines a reactive property that automatically tracks dependencies within the model.

This approach centralizes state and its dependencies, ensuring that updating a single source field triggers a chain of updates across all related computed properties.

from reaktiv import ReactiveModel, computed, field

class ReactiveShoppingCart(ReactiveModel):
    # Define a reactive field with a factory
    items = field(factory=list)

    @computed
    def subtotal(self):
        return sum(item["price"] for item in self.items())

    @computed
    def tax(self):
        return self.subtotal() * 0.1

    @computed
    def total(self):
        return self.subtotal() + self.tax()

    def add_item(self, item):
        # Use .update() to modify the field's content reactively
        self.items.update(lambda items: items + [item])

# Usage
cart = ReactiveShoppingCart(items=[{"name": "Notebook", "price": 12.50}])
print(cart.total())  # 13.75

cart.add_item({"name": "Pen", "price": 2.50})
print(cart.total())  # 16.5

A Resource is a reactive abstraction for managing asynchronous data fetching (like API calls or database queries) within a synchronous signal graph.

Think of it as a spreadsheet cell that fetches data whenever its dependencies change:

• Signal: A cell containing a value (e.g., user_id).
• Resource: A cell that fetches data based on that value (e.g., user_data).
• Computed: A cell that displays a transformation of the fetched data (e.g., user_name).

Key benefits include:

• Async Data, Sync Interface: You can read resource values synchronously in computed signals and effects.
• Automatic Request Management: Automatically cancels outdated requests and manages loading/error states when parameters change.
• Status-Driven UI: Exposes signals like is_loading and error to drive UI state without manual management.

# Reactive (declarative) approach
user_id = signal("user123")

user_resource = resource(
    params=lambda: {"id": user_id()},
    loader=fetch_user
)

# Loading, cancellation, and state management are automatic.
# Access with: user_resource.value(), user_resource.is_loading(), etc.