Examples¶
Complete programs that use Signified. Each one runs on its own; the command to run it from a clone of the repository is at the top of the file.
Effects¶
Three ways to create an effect¶
Create an Effect object, decorate a function with @effect, or call
.rx.effect() on a reactive value.
examples/effects/basics.py
"""Three ways to create an effect.
Run with: uv run examples/effects/basics.py
"""
from signified import Effect, Signal, effect
@effect
def log_count(value: int, label: str) -> None:
# A Signal argument arrives as its current value on each run. Other
# arguments, like label, pass through unchanged.
print(f"{label}: {value}")
def example() -> None:
count = Signal(0)
print("1: Create an Effect object directly:")
observer = Effect(lambda: print(f"Effect: {count.value}"))
try:
count.value = 1
count.value = 1 # Unchanged scalar values do not trigger another run.
finally:
observer.dispose()
print("\n2: Use the @effect decorator:")
logger = log_count(count, "@effect")
try:
count.value = 2
finally:
logger.dispose()
print("\n3: Use .rx.effect():")
display = count.rx.effect(lambda value: print(f"rx.effect: {value}"))
try:
count.value = 3
finally:
display.dispose()
if __name__ == "__main__":
example()
Save settings when they change¶
An effect writes settings to a file, and batch() combines related edits into
one save.
examples/effects/autosave.py
"""Save application settings when they change, combining related edits.
Run with: uv run examples/effects/autosave.py
"""
import json
from pathlib import Path
from tempfile import TemporaryDirectory
from signified import Signal, batch, effect
@effect
def save_settings(settings_path: Path, current_theme: str, current_font_size: int) -> None:
settings = {"theme": current_theme, "font_size": current_font_size}
# write_text opens and closes the file on each run.
settings_path.write_text(json.dumps(settings, indent=2) + "\n", encoding="utf-8")
print(f"Saved {settings}")
def example() -> None:
theme = Signal("light")
font_size = Signal(14)
with TemporaryDirectory(prefix="signified-settings-") as temporary:
settings_path = Path(temporary) / "settings.json"
saver = save_settings(settings_path, theme, font_size) # Saves the initial settings.
try:
# Both writes produce a single save with the final settings.
with batch():
theme.value = "dark"
font_size.value = 18
font_size.value = 20 # Saves synchronously before this line returns.
finally:
saver.dispose()
theme.value = "light" # No save after disposal.
print("\nLast saved settings:")
print(settings_path.read_text(encoding="utf-8"), end="")
# The temporary directory is removed when this example exits.
if __name__ == "__main__":
example()
Follow the selected sensor¶
@tracked_fields makes a dataclass's attributes reactive, so a display follows
whichever sensor is selected.
examples/effects/dashboard.py
"""Batch updates and follow only the selected sensor's readings.
`Sensors` is an ordinary dataclass. `@tracked_fields` makes its attributes
reactive: a calculation or effect that reads `sensors.primary` runs again when
`sensors.primary` is reassigned. Writes are plain attribute assignments.
- `select_sensors` is a calculation that reads whichever sensor is in use. It
takes the whole `sensors` object and reads its fields inside, so it follows
them. Passing `sensors.primary` instead would pass today's number, which
never changes.
- `render` is an effect that prints the selected reading with its label. It
runs again whenever either one changes.
- `batch()` groups several writes into one update.
Expected output:
Temperature: 20 C
Batch writes into one display update:
Temperature: 22 C
Rename the display:
Office: 22 C
Switch to the backup sensor:
Office: 16 C
The primary sensor is no longer a dependency:
Office: 19 C
Run with: uv run examples/effects/dashboard.py
"""
from dataclasses import dataclass
from signified import Effect, batch, computed, tracked_fields
@tracked_fields
@dataclass(slots=True)
class Sensors:
primary: int = 20
backup: int = 18
use_backup: bool = False
label: str = "Temperature"
@computed
def select_sensors(sensors: Sensors) -> int:
return sensors.backup if sensors.use_backup else sensors.primary
def example() -> None:
sensors = Sensors()
selected = select_sensors(sensors)
def render() -> None:
print(f"{sensors.label}: {selected.value} C")
# Note: we hold a reference to effect so that it doesn't get
# immediately garbage collected.
display = Effect(render) # Temperature: 20 C # noqa
print("\nBatch writes into one display update:")
with batch():
sensors.primary = 22
sensors.backup = 16
print("\nRename the display:")
sensors.label = "Office" # Office: 22 C
print("\nSwitch to the backup sensor:")
sensors.use_backup = True # Office: 16 C
print("The primary sensor is no longer a dependency:")
sensors.primary = 99 # No display update.
sensors.backup = 19 # Office: 19 C
if __name__ == "__main__":
example()
Rebuild an index when files change¶
A file watcher updates a signal, and an effect rewrites index.md only when
the notes' titles change. This one needs watchdog.
examples/effects/notes_index/example.py
"""Rebuild an index of Markdown notes whenever the notes folder changes.
`helper.py` creates a temporary notes folder and plays the part of you editing
it. This file only watches the folder and reacts:
- `notes` is a signal holding each note's text. Whenever anything in the
folder changes, the watcher re-reads the folder into `notes`.
- `table_of_contents` is a calculation that lists each note's title.
- `write_index` is an effect that writes the table of contents to `index.md`.
It is the only code that changes anything outside the program.
Editing a note's body without changing its title produces the same table of
contents, so `write_index` does not run and `index.md` is left alone.
The watcher runs on its own thread, so `notes` changes and `write_index` runs
there. That is fine because nothing else uses these values while the edits
happen: Signified needs each set of connected values used by one thread at a
time. If other code also used `notes` while watching, you would pass the
watcher's events to that code's thread, for example through a queue.
Expected output:
Wrote index.md:
- Groceries (groceries.md)
- Ideas (ideas.md)
Adding travel.md
Wrote index.md:
- Groceries (groceries.md)
- Ideas (ideas.md)
- Travel plans (travel.md)
Adding eggs to groceries.md
Renaming Ideas to Project ideas
Wrote index.md:
- Groceries (groceries.md)
- Project ideas (ideas.md)
- Travel plans (travel.md)
Run with: uv run --with watchdog examples/effects/notes_index/example.py
"""
from pathlib import Path
from helper import edit_notes, notes_folder
from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers.polling import PollingObserver
from signified import Signal, computed, effect
def read_notes(folder: Path) -> dict[str, str]:
return {path.name: path.read_text() for path in sorted(folder.glob("*.md"))}
@computed
def table_of_contents(notes: dict[str, str]) -> str:
titles = {name: text.splitlines()[0].removeprefix("# ") for name, text in notes.items()}
return "\n".join(f"- {title} ({name})" for name, title in titles.items())
@effect
def write_index(toc: str, path: Path) -> None:
path.write_text(toc + "\n")
print(f"Wrote {path.name}:\n{toc}")
class UpdateNotes(FileSystemEventHandler):
"""Re-read the folder into `notes` whenever anything in it changes."""
def __init__(self, folder: Path, notes: Signal[dict[str, str]]) -> None:
self.folder = folder
self.notes = notes
def on_any_event(self, event: FileSystemEvent) -> None:
self.notes.value = read_notes(self.folder)
def example() -> None:
with notes_folder() as folder:
notes = Signal(read_notes(folder))
writer = write_index(table_of_contents(notes), folder.parent / "index.md")
# Polling works on every OS; native macOS notifications can go missing.
observer = PollingObserver(timeout=0.1)
observer.schedule(UpdateNotes(folder, notes), str(folder))
observer.start()
try:
edit_notes(folder)
finally:
observer.stop()
observer.join()
writer.dispose()
if __name__ == "__main__":
example()
examples/effects/notes_index/helper.py
"""Stand in for a person editing notes, so the example runs unattended.
Nothing here uses Signified.
"""
import time
from collections.abc import Generator
from contextlib import contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
@contextmanager
def notes_folder() -> Generator[Path]:
"""Create a temporary folder with two notes. It is deleted on exit."""
with TemporaryDirectory() as temporary:
folder = Path(temporary, "notes")
folder.mkdir()
(folder / "groceries.md").write_text("# Groceries\n- milk\n")
(folder / "ideas.md").write_text("# Ideas\n- a reactive to-do app\n")
yield folder
def edit_notes(folder: Path) -> None:
"""Add a note, change one's body, and retitle another.
Pauses half a second before each edit and after the last, so a watcher has
time to notice each one.
"""
edits = [
("Adding travel.md", "travel.md", "# Travel plans\n- book flights\n"),
("Adding eggs to groceries.md", "groceries.md", "# Groceries\n- milk\n- eggs\n"),
("Renaming Ideas to Project ideas", "ideas.md", "# Project ideas\n- a reactive to-do app\n"),
]
for message, name, text in edits:
time.sleep(0.5)
print(f"\n{message}")
(folder / name).write_text(text)
time.sleep(0.5)
Plugins¶
Count reads and writes¶
A plugin that records how often each reactive value is read and changed.
examples/plugins/access_tracker.py
"""Count reads and writes of each reactive value with a plugin.
Run with: uv run examples/plugins/access_tracker.py
"""
import datetime
import time
from collections import defaultdict
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Any
from weakref import WeakValueDictionary
from signified import Signal, Variable
from signified.plugins import hookimpl, plugin_manager
class EventType(Enum):
READ = auto()
WRITE = auto()
CREATE = auto()
@dataclass
class AccessEvent:
"""Represents a single access to a reactive value."""
timestamp: datetime.datetime
event_type: EventType
value: Any
@dataclass
class AccessStats:
read_count: int = 0
write_count: int = 0
creation_time: datetime.datetime = field(default_factory=datetime.datetime.now)
last_access_time: datetime.datetime = field(default_factory=datetime.datetime.now)
value_history: list[AccessEvent] = field(default_factory=list, repr=False)
def add_event(self, event_type: EventType, value: Any = None) -> None:
now = datetime.datetime.now()
if event_type == EventType.READ:
self.read_count += 1
self.last_access_time = now
elif event_type in (EventType.WRITE, EventType.CREATE):
if event_type == EventType.WRITE:
self.write_count += 1
elif event_type == EventType.CREATE:
self.creation_time = now
self.last_access_time = now
self.value_history.append(AccessEvent(now, event_type, value))
@property
def total_accesses(self) -> int:
"""Total number of reads and modifications."""
return self.read_count + self.write_count
@property
def last_value(self) -> Any:
"""Most recently recorded value."""
return self.value_history[-1].value if self.value_history else None
class AccessTracker:
"""Plugin for tracking reactive value access patterns.
History records the internal cached value without invoking read hooks or
evaluating lazy computations. A Computed's creation value is its initial
None placeholder; its evaluated value is recorded by the updated hook.
"""
def __init__(self):
self.stats: dict[int, AccessStats] = defaultdict(AccessStats)
self._variables: WeakValueDictionary[int, Variable[Any]] = WeakValueDictionary()
def cleanup(self) -> None:
"""Clear stats."""
self.stats.clear()
self._variables.clear()
@hookimpl
def created(self, value: Variable[Any]) -> None:
"""Track creation of new reactive values."""
self.stats[id(value)].add_event(EventType.CREATE, value._value)
self._variables[id(value)] = value
@hookimpl
def updated(self, value: Variable[Any]) -> None:
"""Track modifications to reactive values."""
self.stats[id(value)].add_event(EventType.WRITE, value._value)
@hookimpl
def read(self, value: Variable[Any]) -> None:
"""Record a read access to a reactive value."""
self.stats[id(value)].add_event(EventType.READ)
def get_stats(self, value: Variable[Any]) -> AccessStats:
"""Get access statistics for a specific reactive value."""
return self.stats[id(value)]
def print_summary(self) -> None:
"""Print summary statistics for all tracked values."""
print("\nAccess Pattern Summary:")
print("-" * 50)
for value_id, stats in self.stats.items():
value = self._variables.get(value_id)
print(f"Value {value:n}:")
print(f"\tReads: {stats.read_count}")
print(f"\tModifications: {stats.write_count}")
print(f"\tTotal accesses: {stats.total_accesses}")
if stats.creation_time and stats.last_access_time:
lifetime = stats.last_access_time - stats.creation_time
print(f"\tLifetime: {lifetime.total_seconds():.2f} seconds")
print()
tracker = AccessTracker()
plugin_manager.register(tracker)
if __name__ == "__main__":
x = Signal(1).with_name("x")
y = Signal(2).with_name("y")
time.sleep(0.1)
x.value = 10
y.value = 20
x.value = 30
z = (x + y).with_name("z")
_ = z.value # Reads z, which reads x and y as it calculates.
time.sleep(0.1)
tracker.print_summary()
print(f"Value history for x: {[v.value for v in tracker.get_stats(x).value_history]}")
print(f"Value history for y: {[v.value for v in tracker.get_stats(y).value_history]}")
print(f"Value history for z: {[v.value for v in tracker.get_stats(z).value_history]}")
plugin_manager.unregister(tracker)
tracker.cleanup()