xdol.registry

A typed dict-backed plugin registry.

The Registry is the dict-pattern that recurs across packages (falaw.registry, lacing.processors, lookbook.registry, …): a dict, a register(key, value) call, a get(key) lookup, an iteration protocol, and a small number of conveniences that each package keeps re-implementing — aliases, lazy loading, decorator-form registration, and conflict policies. Registry collects these into one builtin-only, MutableMapping-compatible class.

Basic use:

>>> from xdol.registry import Registry
>>> tools = Registry[str](name="tools")
>>> tools.register("hello", "say hi")
'say hi'
>>> tools["hello"]
'say hi'
>>> sorted(tools)
['hello']

Decorator form:

>>> handlers = Registry()
>>> @handlers.register_decorator("greet")
... def greet(name): return f"hi {name}"
>>> handlers["greet"]("ada")
'hi ada'

Aliases:

>>> models = Registry()
>>> models.register("fal-ai/flux/dev", {"family": "flux"})
{'family': 'flux'}
>>> models.alias("flux", "fal-ai/flux/dev")
>>> models["flux"]
{'family': 'flux'}
>>> "flux" in models
True

Conflict policies — error (default), replace, keep:

>>> r = Registry(on_conflict="error")
>>> r.register("k", 1)
1
>>> r.register("k", 2)
Traceback (most recent call last):
    ...
xdol.registry.RegistryConflict: 'k' is already registered in <Registry>

>>> r2 = Registry(on_conflict="keep")
>>> r2.register("k", 1)
1
>>> r2.register("k", 2)  # silently kept
1
>>> r2["k"]
1

Plain assignment replaces; opt in to strict_setitem to make it honour the policy:

>>> loose = Registry()
>>> loose["k"] = 1
>>> loose["k"] = 2  # explicit assignment replaces
>>> loose["k"]
2
>>> strict = Registry(strict_setitem=True)
>>> strict["k"] = 1
>>> strict["k"] = 2
Traceback (most recent call last):
    ...
xdol.registry.RegistryConflict: 'k' is already registered in <Registry>

Lazy loading — value is built on first access:

>>> calls = []
>>> def make_thing():
...     calls.append(1)
...     return {"built": True}
>>> r3 = Registry()
>>> r3.register_lazy("thing", make_thing)
>>> calls
[]
>>> r3["thing"]
{'built': True}
>>> r3["thing"]  # cached after first build
{'built': True}
>>> calls
[1]

Subscription — observe registrations as they happen:

>>> r4 = Registry()
>>> seen = []
>>> sub = r4.subscribe(lambda key, value: seen.append((key, value)))
>>> r4.register("a", 1)
1
>>> r4.register("b", 2)
2
>>> seen
[('a', 1), ('b', 2)]
>>> sub.unsubscribe()
>>> r4.register("c", 3)
3
>>> seen  # no longer notified
[('a', 1), ('b', 2)]

Classes

Registry(*[, name, on_conflict, strict_setitem])	A typed dict-backed plugin registry.
Subscription(registry, callback)	Handle returned by Registry.subscribe().

Exceptions

RegistryConflict	Raised when registering a key that is already taken under on_conflict='error'.
RegistryError	Base for registry errors.
RegistryMissing	Raised by Registry.get() when a key is missing.

class xdol.registry.Registry(*, name='', on_conflict='error', strict_setitem=False)

Bases: MutableMapping, Generic[V]

A typed dict-backed plugin registry.

Parameters:

• name (str) – Optional name; appears in error messages and repr.

• on_conflict (Literal['error', 'replace', 'keep']) – 'error' (default) raises RegistryConflict when registering a key that already exists. 'replace' silently overwrites. 'keep' silently keeps the original.

• strict_setitem (bool) – Opt-in (default False). By default reg[k] = v is an explicit assignment that replaces whatever is there, bypassing on_conflict. With strict_setitem=True it follows the on_conflict policy exactly like register() (so MutableMapping.update and setdefault do too; note update is not atomic: entries before the first conflict stay).

• (reading (Keys that are aliases resolve to their canonical key everywhere)

:param : :param reg[k] = v: :param register(): :param register_lazy()): :param so the on_conflict: :param policy always applies to the canonical entry.: :param Implements MutableMapping so anything that takes a mapping (dict(...): :param : :param iteration: :param length checks: :param in lookups: :param .items()) just works.:

alias(alias, target)

Make alias resolve to target (which must already be registered).

Return type:

None

get(key, default=<object object>)

Like dict.get but raises RegistryMissing if no default given.

keys_with_tag(tag)

Return canonical keys whose tag set contains tag.

Return type:

list[str]

register(key, value, *, tags=())

Register value under key. Returns value so it can be used inline.

If key is an alias, it is resolved to its canonical key first, so the on_conflict policy applies to that entry (no unreachable shadow entry).

Return type:

TypeVar(V)

register_decorator(key, *, tags=())

Return a decorator that registers the decorated object under key.

Return type:

Callable[[TypeVar(V)], TypeVar(V)]

>>> r = Registry()
>>> @r.register_decorator("hi")
... def hi(): return "hello"
>>> r["hi"]()
'hello'

register_lazy(key, loader, *, tags=())

Register a no-arg loader callable; value is built on first __getitem__.

Once built, the result replaces the loader and is cached. Alias keys resolve to their canonical key, as in register().

Return type:

None

search(*, tags=())

Return values whose tag set contains ALL of tags (empty tags = all).

Return type:

list[TypeVar(V)]

subscribe(callback)

Call callback(key, value) on every eager registration.

Lazy registrations do NOT fire the callback at registration time; only when the value is materialized via __getitem__. This keeps subscribers free of partially-built state.

Return type:

Subscription

exception xdol.registry.RegistryConflict

Bases: RegistryError

Raised when registering a key that is already taken under on_conflict='error'.

exception xdol.registry.RegistryError

Bases: Exception

Base for registry errors.

exception xdol.registry.RegistryMissing

Bases: RegistryError, KeyError

Raised by Registry.get() when a key is missing.

class xdol.registry.Subscription(registry, callback)

Bases: object

Handle returned by Registry.subscribe(). Call unsubscribe() to detach.