sequenceDiagram
participant H as hot_add(GermanDict)
participant R as Registry
participant S as Signal (rt.dicts)
participant E as @effect rebuild_index
H->>R: provide(IDictionary, german)
R->>S: Signal.set([en, fr, de])
S->>E: notify: deps changed
E->>E: re-run → rebuilds index
Note over E: Index now has EN, FR, DE
3. Dynamic Services
Services come and go. Your component reacts automatically.
The Problem
You have two dictionaries (English, French) and a spell checker. @requires(dictionary=IDictionary) injects one dictionary. But the spell checker needs all dictionaries — and it needs to handle new ones appearing and old ones disappearing at runtime.
In traditional service-oriented frameworks, you’d write @bind/@unbind callbacks — manual wiring for every service that comes and goes. In SignalPy, you use @requires(dicts=list[IDictionary]) plus a reactive method called @effect (introduced below) — a method the kernel re-runs automatically when its tracked dependencies change. The reactive engine handles propagation:
Aggregate Injection
@component("checker")
@requires(dictionaries=list[IDictionary])
class SpellChecker:
@lifecycle.activate
def activate(self):
print(f"Checker ready with {len(self.rt.dictionaries)} dictionaries")Checker ready with 2 dictionaries
Note
@requires(dictionaries=list[IDictionary]) injects a list of all services matching the contract. The list[C] type hint tells the kernel “I want all of them.”
list[X] is a subscription, not a dependency
The scalar form @requires(dictionary=IDictionary) from Tutorial 2 was a boot-blocking dependency: the spell checker waited for a dictionary to exist before activating. The aggregate form @requires(dictionaries=list[IDictionary]) looks similar in code but behaves differently — it’s a reactive subscription on the registry:
- The spell checker activates immediately, even with zero dictionaries.
self.rt.dictionariesstarts as[]and the registry listener appends/removes providers as they come and go.- No toposort edge is created, so dictionaries don’t have to boot first.
That’s why the same decorator name (@requires) carries two different boot meanings — and why aggregates pair naturally with @effect: the activate method runs once, but the effect re-runs every time the list changes. If you depend on the list being populated, put that logic in @effect, not in activate.
The same applies to @requires(..., key="prop") (reactive map) and @requires(..., optional=True) (single or None) — both non-blocking. Only the bare scalar shape blocks. The Decorators reference has the full table.
Reactive Effects
An @effect is a method that re-runs whenever its tracked dependencies change. Read self.rt.dictionaries inside an effect → the reactive engine tracks that. When the list changes, the effect re-runs automatically.
@component("checker")
@requires(dictionaries=list[IDictionary])
class SpellChecker:
@lifecycle.activate
def activate(self):
self.by_language = {}
@effect
def rebuild_index(self):
# Reads self.rt.dictionaries — tracked automatically.
# When the list changes, this method re-runs.
self.by_language = {}
for d in self.rt.dictionaries:
lang = getattr(d, "_language", "??")
self.by_language[lang] = d
print(f" Index rebuilt: {list(self.by_language.keys())}") Index rebuilt: ['EN', 'FR']
@effect replaces @bind/@unbind
Instead of separate add/remove callbacks, write one method that reads the current state. The reactive engine tracks what it reads and re-runs it when those values change.
Hot-Add at Runtime
@component("dict-de")
@provides(IDictionary)
@prop("_language", "language", "DE") # @prop(attr_name, service_property, default)
class GermanDict:
@lifecycle.activate
def activate(self):
self.words = {"hallo", "welt"}
def check_word(self, word):
return word.lower() in self.words
await kernel.hot_add(GermanDict)@prop(attr_name, service_property, default)
Three arguments: the Python attribute name on the instance (self._language), the service-registry property name other components match against ("language"), and the default value. Service properties are how the registry filters and ranks providers — see “Properties & Ranking” below.
Index rebuilt: ['EN', 'FR', 'DE']The spell checker’s @effect fires automatically. Zero changes to the spell checker code.
Hot-Remove
await kernel.hot_remove("dict-fr") Index rebuilt: ['EN', 'DE']
Zero code changes in the spell checker
It didn’t know German was coming. It doesn’t know French left. The @effect kept its index current automatically.
Computed Properties
A @computed property caches its result and recomputes only when dependencies change:
@computed
def available_languages(self):
return [getattr(d, "_language", "?") for d in self.rt.dictionaries]Read self.available_languages() — always current, cached until the dictionary list changes.
When two writes belong together: batch()
A reactive engine fires effects after every write. That’s what you want most of the time — but sometimes two writes belong together, and you don’t want anyone to observe the half-updated state in between.
The problem, made concrete
You’re rotating an API endpoint and its credential at the same time. Both live in config, written one after the other.
(IConfig is one of the kernel’s built-in contracts from signalpy.kernel.contracts — ConfigProvider ships in the kernel and implements it. You can @requires(config=IConfig) without writing a provider yourself.)
@component("client")
@requires(config=IConfig)
class APIClient:
@effect
async def reconnect(self):
url = self.rt.config.get("api.url")
token = self.rt.config.get("api.token")
await self._open(url, token) # uses BOTH on every changeIn a deploy script:
config.set("api.url", "https://api-v2.example.com")
config.set("api.token", "tok_v2_xxxxx")Without batching, the effect runs twice:
| Step | Trigger | Body sees | Outcome |
|---|---|---|---|
| 1 | set("api.url", v2_url) fires |
v2_url, v1_token |
tries to open v2 endpoint with v1 token → auth error |
| 2 | set("api.token", v2_token) fires |
v2_url, v2_token |
opens correctly |
The first run wasn’t just wasted work — it was wrong. A real auth attempt went out with mismatched creds.
The fix
Wrap the correlated writes:
from signalpy.kernel import batch
with batch():
config.set("api.url", "https://api-v2.example.com")
config.set("api.token", "tok_v2_xxxxx")
# effect fires ONCE, with both new valuesNow the body sees (v2_url, v2_token) on its single run. No half-state ever exists from the effect’s point of view.
When you don’t need it
The kernel batches automatically in two common situations, so you don’t need to reach for batch():
- Hot-add / hot-remove cascades — a single
kernel.hot_add(GermanDict)causes one cascade; one effect run. - Diamond dependencies — if
A → BandA → Cand an effect reads bothBandC, a single update toAruns the effect once, not twice.
You only need batch() for two top-level set calls that are semantically one update. The kernel cannot guess they’re related; you have to tell it.
Async effects and long-running work
What if a dependency changes while an async effect is mid-await? The engine handles this safely by default: it queues a re-run after the in-flight body finishes. You don’t need to do anything special for most cases.
For advanced scenarios (expensive work you want to bail out of, or resources the next run needs released), see Threading Model → Async Supersede which covers is_stale() and cancel_on_supersede=True.
Properties & Ranking
Properties are metadata attached to a service when it registers:
@component("dict-en")
@provides(IDictionary)
@prop("_language", "language", "EN") # property: language = "EN"
@prop("_ranking", "service.ranking", 0) # priority: 0 = highest
class EnglishDict: ...The special property service.ranking controls priority. When multiple services match a single @requires, the one with the lowest ranking wins.