Real-World Examples

This section contains practical examples of using reaktiv in real-world scenarios. These examples demonstrate how reactive programming can simplify complex state management challenges.

Table of Contents

• Real-World Examples
• Table of Contents
• Configuration Management
• Data Processing Pipeline
• Cache Invalidation
• Form Validation
• API Data Fetching
• Status Monitoring

Configuration Management

Managing configuration from multiple sources (defaults, user settings, environment) with automatic priority resolution:

Editor (session: default) Run

from reaktiv import signal, computed, effect

# Different configuration sources
default_config = signal({
    "timeout": 30,
    "retries": 3,
    "debug": False,
    "log_level": "INFO"
})

user_config = signal({})
env_config = signal({})

@computed
def effective_config():
    return {
        **default_config(),
        **user_config(),
        **env_config(),  # Environment overrides everything
    }

# Derived settings that automatically update when any config changes
@computed
def connection_settings():
    return {
        "connect_timeout": effective_config()["timeout"],
        "max_attempts": effective_config()["retries"],
        "verbose": effective_config()["debug"],
    }

@computed
def logger_settings():
    return {
        "level": effective_config()["log_level"],
        "debug_mode": effective_config()["debug"],
    }

# Effect to log when settings change
def print_config():
    print(f"Config updated: {effective_config()}")

config_monitor = effect(print_config)

# Update a specific config source
user_config.set({"timeout": 60, "log_level": "DEBUG"})
# Automatically updates effective_config, connection_settings, logger_settings
# and triggers the config_monitor effect

# Later, update from environment
env_config.set({"retries": 5})
# Everything dependent on retries updates automatically

print(connection_settings())
print(logger_settings())
config_monitor.dispose()

Output Clear

Data Processing Pipeline

Building a multi-stage data processing pipeline where each step depends on the previous one:

Editor (session: default) Run

from reaktiv import signal, computed
import json

# Raw data source
raw_data = signal('{"values": [1, 2, 3, -4, 5, -6]}')

# Parsing stage
@computed
def parsed_data():
    return json.loads(raw_data())

# Extraction stage
@computed
def values():
    return parsed_data()["values"]

# Filtering stage
@computed
def positive_values():
    return [x for x in values() if x > 0]

# Transformation stage
@computed
def squared_values():
    return [x * x for x in positive_values()]

# Aggregation stage
@computed
def stats():
    values = squared_values()
    return {
        "count": len(values),
        "sum": sum(values),
        "average": sum(values) / len(values) if values else 0,
    }

print(stats())
# {'count': 4, 'sum': 39, 'average': 9.75}

# Update the raw data - all stages recompute automatically
raw_data.set('{"values": [10, 20, 30, 40]}')
print(stats())
# {'count': 4, 'sum': 3000, 'average': 750.0}

Output Clear

Cache Invalidation

Smart cache invalidation system that automatically refreshes cached data when dependencies change:

Editor (session: default) Run

from reaktiv import signal, computed, effect
import time

# Simulated database
db = {"user1": {"name": "Alice"}, "user2": {"name": "Bob"}}

# Cache version signal - incremented when database changes
cache_version = signal(1)

# Active user ID
user_id = signal("user1")

# This computed value acts as our cache with automatic invalidation
@computed
def user_data():
    return {
        "id": user_id(),
        "data": db[user_id()],  # In real code, this would be a database query
        "cached_at": time.time(),
        "version": cache_version(),  # Including version causes cache refresh when version changes
    }

# Cache monitor
def print_cache_entry():
    print(
        f"Cache data for user {user_data()['id']}: {user_data()['data']} "
        f"(version: {user_data()['version']})"
    )

cache_logger = effect(print_cache_entry)

# Change user - cache recomputes automatically
user_id.set("user2")

# Simulate database update and cache invalidation
db["user2"] = {"name": "Robert"}
cache_version.update(lambda v: v + 1)  # Increment version to invalidate cache
cache_logger.dispose()

Output Clear

Form Validation

Complex form validation with interdependent fields:

Editor (session: default) Run

from reaktiv import signal, computed, effect

# Form fields
username = signal("")
password = signal("")
password_confirm = signal("")
email = signal("")
terms_accepted = signal(False)

# Individual field validations
@computed
def username_error():
    if not username():
        return "Username is required"
    if len(username()) < 3:
        return "Username must be at least 3 characters"
    return None

@computed
def password_error():
    if not password():
        return "Password is required"
    if len(password()) < 8:
        return "Password must be at least 8 characters"
    return None

@computed
def password_match_error():
    return "Passwords don't match" if password() != password_confirm() else None

@computed
def email_error():
    if not email():
        return "Email is required"
    if "@" not in email():
        return "Invalid email format"
    return None

@computed
def terms_error():
    return "You must accept the terms" if not terms_accepted() else None

# Combined form validation status
@computed
def form_errors():
    return {
        "username": username_error(),
        "password": password_error(),
        "password_confirm": password_match_error(),
        "email": email_error(),
        "terms": terms_error(),
    }

@computed
def has_errors():
    return any(error is not None for error in form_errors().values())

# Form submission state
@computed
def can_submit():
    return not has_errors() and terms_accepted()

# Monitor submission state
def print_submission_state():
    print(f"Form can be submitted: {can_submit()}")

submission_monitor = effect(print_submission_state)

# User interaction simulation
username.set("bob")
email.set("bob@example.com")
password.set("password123")
password_confirm.set("password123")
terms_accepted.set(True)
submission_monitor.dispose()

Output Clear

API Data Fetching

Keep request parameters, loading state, errors, and automatic refresh together in a reactive model:

Editor (session: default) Run

import asyncio

from reaktiv import (
    ReactiveModel,
    ResourceLoaderParams,
    ResourceStatus,
    effect,
    field,
    resource,
)


class UserStore(ReactiveModel):
    user_id = field(2)

    def __init__(self):
        self.finished = asyncio.Event()
        super().__init__()

    @resource[int, dict[str, object]]
    def user(self):
        return self.user_id()

    @user.loader
    async def load_user(self, request: ResourceLoaderParams[int]):
        await asyncio.sleep(0.1)  # Replace with an HTTP request.
        user_id = request.params
        if user_id % 2:
            raise RuntimeError(f"User {user_id} was not found")
        return {"id": user_id, "name": f"User {user_id}", "active": True}

    @effect
    def show_status(self):
        status = self.user.status()

        if status in {ResourceStatus.LOADING, ResourceStatus.RELOADING}:
            print(f"Loading user {self.user_id()}...")
        elif status == ResourceStatus.ERROR:
            print(f"Error: {self.user.error()}")
            self.finished.set()
        elif self.user.has_value():
            print(f"Loaded: {self.user.value()}")
            self.finished.set()


async def wait_for_load(store):
    await store.finished.wait()
    store.finished.clear()


store = UserStore()
await wait_for_load(store)

print("\nSwitching users reloads automatically:")
store.user_id.set(4)
await wait_for_load(store)

print("\nRefreshing the current user:")
store.user.reload()
await wait_for_load(store)

print("\nErrors are reactive state too:")
store.user_id.set(3)
await wait_for_load(store)

store.dispose()

Output Clear

Status Monitoring

Building a reactive system monitoring dashboard:

Editor (session: default) Run

from reaktiv import signal, computed, effect

# System metrics (in a real app, these would be updated from actual monitoring)
cpu_usage = signal(25.0)  # percentage
memory_usage = signal(40.0)  # percentage
disk_usage = signal(60.0)  # percentage
error_count = signal(0)
request_count = signal(1000)

# Derived metrics
@computed
def error_rate():
    return (error_count() / request_count() * 100) if request_count() > 0 else 0

# Status thresholds
@computed
def cpu_status():
    if cpu_usage() > 90:
        return "critical"
    if cpu_usage() > 70:
        return "warning"
    return "normal"

@computed
def memory_status():
    if memory_usage() > 90:
        return "critical"
    if memory_usage() > 70:
        return "warning"
    return "normal"

@computed
def disk_status():
    if disk_usage() > 90:
        return "critical"
    if disk_usage() > 80:
        return "warning"
    return "normal"

@computed
def error_status():
    if error_rate() > 5:
        return "critical"
    if error_rate() > 1:
        return "warning"
    return "normal"

# Overall system status (worst of any individual status)
@computed
def system_status():
    statuses = (cpu_status, memory_status, disk_status, error_status)
    if any(status() == "critical" for status in statuses):
        return "critical"
    if any(status() == "warning" for status in statuses):
        return "warning"
    return "normal"

# Alert system
def alert_system():
    status = system_status()
    components = []

    if cpu_status() != "normal":
        components.append(f"CPU: {cpu_usage():.1f}%")
    if memory_status() != "normal":
        components.append(f"Memory: {memory_usage():.1f}%")
    if disk_status() != "normal":
        components.append(f"Disk: {disk_usage():.1f}%")
    if error_status() != "normal":
        components.append(f"Error rate: {error_rate():.2f}%")

    if status != "normal":
        print(f"❗ System status: {status.upper()}")
        print(f"   Affected components: {', '.join(components)}")
    else:
        print("✓ System status: NORMAL - All systems operational")

# Monitor status changes
status_monitor = effect(alert_system)

# Initial output: "✓ System status: NORMAL - All systems operational"

# Simulate memory spike
memory_usage.set(75.0)
# Output: "❗ System status: WARNING
#          Affected components: Memory: 75.0%"

# Simulate error increase
error_count.set(100)
request_count.set(1000)
# Output: "❗ System status: WARNING
#          Affected components: Memory: 75.0%, Error rate: 10.00%"

# Simulate critical CPU spike
cpu_usage.set(95.0)
# Output: "❗ System status: CRITICAL
#          Affected components: CPU: 95.0%, Memory: 75.0%, Error rate: 10.00%"

# Simulate recovery
memory_usage.set(50.0)
cpu_usage.set(30.0)
error_count.set(5)
# Output: "❗ System status: WARNING
#          Affected components: Error rate: 0.50%"

# Full recovery
error_count.set(0)
# Output: "✓ System status: NORMAL - All systems operational"
status_monitor.dispose()

Output Clear

Each of these examples demonstrates how reaktiv simplifies complex state management scenarios by automatically handling dependencies and updates. You can build on these patterns to create more complex reactive systems tailored to your specific needs.

For more detailed examples or to contribute your own, visit our GitHub repository.