windlass.core.registry

registry

The component registry — how Windlass stays open for extension.

Every replaceable part of the framework (LLMs, chunkers, retrievers, vector stores, ...) is a component kind. Implementations register a name against a kind; builders and the DI container resolve names to instances. Nothing in Windlass imports a concrete provider module directly, which is what keeps the core install tiny and the architecture pluggable.

Three ways to register

1. Decorator (in-process) — the common case for application code::

from windlass import register
from windlass.interfaces import Chunker

@register.chunker("sentence")
class SentenceChunker(Chunker):
    ...

2. Lazy path (used by Windlass itself) — records a dotted path without importing the module, so import windlass never pulls in optional deps::

REGISTRY.register_lazy("vectordb", "faiss", "windlass.providers.vectordb.faiss:FaissStore")

3. Entry points (distributable plugins) — a third-party package publishes::

[project.entry-points."windlass.chunker"]
sentence = "my_pkg.chunkers:SentenceChunker"

and it appears in windlass automatically, with no code change on either side.

Thread safety

All mutating operations take a re-entrant lock, and lazy resolution is idempotent, so the registry is safe to use from worker threads.

ComponentSpec dataclass

ComponentSpec(
    kind: str,
    name: str,
    target: Any = None,
    path: str | None = None,
    extra: str | None = None,
    aliases: tuple[str, ...] = (),
    description: str = "",
    defaults: dict[str, Any] = dict(),
    origin: str = "user",
)

A registry entry: how to obtain one implementation of a component kind.

Exactly one of target or path is set. path entries are resolved on first use, which is what makes registration free of import cost.

Attributes:

Name	Type	Description
kind	str	The component kind, e.g. "chunker".
name	str	The lookup name, e.g. "semantic".
target	Any	The resolved class or factory, once known.
path	str | None	Dotted "module:Attribute" path for deferred resolution.
extra	str | None	Extras group needed for this component, used in error messages.
aliases	tuple[str, ...]	Additional names that resolve to this spec.
description	str	One-line summary shown by windlass list.
defaults	dict[str, Any]	Keyword arguments merged under user config at construction.
origin	str	"builtin", "plugin" or "user". Purely informational.

resolve

resolve() -> Any

Import and cache the target if it was registered lazily.

Returns:

Type	Description
Any	The class or factory for this component.

Raises:

Type	Description
RegistryError	If the dotted path cannot be imported. Missing optional dependencies surface as :class:~windlass.core.exceptions.MissingDependencyError from inside the provider module instead.

Source code in src\windlass\core\registry.py

def resolve(self) -> Any:
    """Import and cache the target if it was registered lazily.

    Returns:
        The class or factory for this component.

    Raises:
        RegistryError: If the dotted path cannot be imported. Missing
            *optional dependencies* surface as
            :class:`~windlass.core.exceptions.MissingDependencyError` from
            inside the provider module instead.
    """
    if self.target is not None:
        return self.target
    if not self.path:  # pragma: no cover - guarded at registration
        raise RegistryError(f"Component {self.kind}:{self.name} has neither target nor path.")
    module_path, _, attr = self.path.partition(":")
    try:
        module = importlib.import_module(module_path)
    except Exception as exc:
        from windlass.core.exceptions import MissingDependencyError

        if isinstance(exc, MissingDependencyError):
            raise
        raise RegistryError(
            f"Failed to import {self.kind} {self.name!r} from {self.path!r}: {exc}",
            hint=(f'Try: pip install "windlass[{self.extra}]"' if self.extra else None),
            context={"kind": self.kind, "name": self.name, "path": self.path},
        ) from exc
    self.target = getattr(module, attr) if attr else module
    return self.target

Registry

Registry(*, discover: bool = False)

A thread-safe, kind-partitioned component registry.

You almost never construct this yourself — use the module-level :data:REGISTRY singleton, or the :data:register decorator namespace.

Parameters:

Name	Type	Description	Default
discover	bool	Whether the first lookup should scan installed distributions for entry-point plugins. The process-wide :data:REGISTRY does; a registry you construct yourself does not, so it stays an isolated container holding exactly what you put in it. Without that distinction, installing any third-party Windlass plugin would change the contents of every registry in the process, including the ones tests build to get a clean slate.	False

Example

reg = Registry() _ = reg.register("chunker", "shouty", lambda: None, description="demo") reg.has("chunker", "shouty") True reg.names("chunker") ['shouty']

Opt in explicitly when you do want the installed plugins:

discovered = Registry(discover=True) isinstance(discovered.load_plugins(), list) True

Source code in src\windlass\core\registry.py

def __init__(self, *, discover: bool = False) -> None:
    self._specs: dict[str, dict[str, ComponentSpec]] = {k: {} for k in COMPONENT_KINDS}
    self._aliases: dict[str, dict[str, str]] = {k: {} for k in COMPONENT_KINDS}
    self._lock = threading.RLock()
    self._discover = discover
    self._plugins_loaded = not discover

add_kind

add_kind(kind: str) -> None

Declare a brand new component kind at runtime.

Parameters:

Name	Type	Description	Default
kind	str	The new kind's name.	required

Source code in src\windlass\core\registry.py

def add_kind(self, kind: str) -> None:
    """Declare a brand new component kind at runtime.

    Args:
        kind: The new kind's name.
    """
    with self._lock:
        self._specs.setdefault(kind, {})
        self._aliases.setdefault(kind, {})

kinds

kinds() -> list[str]

Return every known component kind, sorted.

Source code in src\windlass\core\registry.py

def kinds(self) -> list[str]:
    """Return every known component kind, sorted."""
    return sorted(self._specs)

register

register(
    kind: str,
    name: str,
    target: Any,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "user",
    override: bool = False
) -> ComponentSpec

Register a concrete class or factory.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Lookup name; case-insensitive.	required
target	Any	A class or a zero-or-more-argument factory callable.	required
aliases	tuple[str, ...] | str	Extra names resolving to the same component.	()
description	str	One-line summary for CLI listings and docs.	''
extra	str | None	Extras group required, used in error hints.	None
defaults	dict[str, Any] | None	Config defaults merged beneath user-supplied kwargs.	None
origin	str	builtin / plugin / user.	'user'
override	bool	Allow replacing an existing registration.	False

Returns:

Type	Description
ComponentSpec	The stored :class:ComponentSpec.

Raises:

Type	Description
DuplicateComponentError	If name is taken and override is False.

Source code in src\windlass\core\registry.py

def register(
    self,
    kind: str,
    name: str,
    target: Any,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "user",
    override: bool = False,
) -> ComponentSpec:
    """Register a concrete class or factory.

    Args:
        kind: Component kind.
        name: Lookup name; case-insensitive.
        target: A class or a zero-or-more-argument factory callable.
        aliases: Extra names resolving to the same component.
        description: One-line summary for CLI listings and docs.
        extra: Extras group required, used in error hints.
        defaults: Config defaults merged beneath user-supplied kwargs.
        origin: ``builtin`` / ``plugin`` / ``user``.
        override: Allow replacing an existing registration.

    Returns:
        The stored :class:`ComponentSpec`.

    Raises:
        DuplicateComponentError: If ``name`` is taken and ``override`` is False.
    """
    return self._store(
        ComponentSpec(
            kind=kind,
            name=name.lower(),
            target=target,
            aliases=(aliases,) if isinstance(aliases, str) else tuple(aliases),
            description=description,
            extra=extra,
            defaults=dict(defaults or {}),
            origin=origin,
        ),
        override=override,
    )

register_lazy

register_lazy(
    kind: str,
    name: str,
    path: str,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "builtin",
    override: bool = False
) -> ComponentSpec

Register a component by dotted path without importing it.

This is how every built-in provider is wired up: the module is only imported the first time somebody actually asks for that component.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Lookup name.	required
path	str	"package.module:Attribute".	required
aliases	tuple[str, ...] | str	Extra names resolving here.	()
description	str	One-line summary.	''
extra	str | None	Extras group required by the target module.	None
defaults	dict[str, Any] | None	Config defaults.	None
origin	str	Provenance label.	'builtin'
override	bool	Allow replacing an existing registration.	False

Returns:

Type	Description
ComponentSpec	The stored :class:ComponentSpec.

Source code in src\windlass\core\registry.py

def register_lazy(
    self,
    kind: str,
    name: str,
    path: str,
    *,
    aliases: tuple[str, ...] | str = (),
    description: str = "",
    extra: str | None = None,
    defaults: dict[str, Any] | None = None,
    origin: str = "builtin",
    override: bool = False,
) -> ComponentSpec:
    """Register a component by dotted path without importing it.

    This is how every built-in provider is wired up: the module is only
    imported the first time somebody actually asks for that component.

    Args:
        kind: Component kind.
        name: Lookup name.
        path: ``"package.module:Attribute"``.
        aliases: Extra names resolving here.
        description: One-line summary.
        extra: Extras group required by the target module.
        defaults: Config defaults.
        origin: Provenance label.
        override: Allow replacing an existing registration.

    Returns:
        The stored :class:`ComponentSpec`.
    """
    return self._store(
        ComponentSpec(
            kind=kind,
            name=name.lower(),
            path=path,
            aliases=(aliases,) if isinstance(aliases, str) else tuple(aliases),
            description=description,
            extra=extra,
            defaults=dict(defaults or {}),
            origin=origin,
        ),
        override=override,
    )

unregister

unregister(kind: str, name: str) -> None

Remove a component and its aliases. No-op when absent.

Source code in src\windlass\core\registry.py

def unregister(self, kind: str, name: str) -> None:
    """Remove a component and its aliases. No-op when absent."""
    with self._lock:
        bucket = self._bucket(kind)
        resolved = self._aliases[kind].pop(name.lower(), name.lower())
        spec = bucket.pop(resolved, None)
        if spec is not None:
            for alias in spec.aliases:
                self._aliases[kind].pop(alias.lower(), None)

get

get(kind: str, name: str) -> ComponentSpec

Return the spec registered under name.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Name or alias, case-insensitive.	required

Returns:

Type	Description
ComponentSpec	The matching spec.

Raises:

Type	Description
ComponentNotFoundError	If nothing matches. The error lists all available names for that kind.

Source code in src\windlass\core\registry.py

def get(self, kind: str, name: str) -> ComponentSpec:
    """Return the spec registered under ``name``.

    Args:
        kind: Component kind.
        name: Name or alias, case-insensitive.

    Returns:
        The matching spec.

    Raises:
        ComponentNotFoundError: If nothing matches. The error lists all
            available names for that kind.
    """
    self.ensure_plugins_loaded()
    with self._lock:
        bucket = self._bucket(kind)
        key = name.lower()
        key = self._aliases[kind].get(key, key)
        spec = bucket.get(key)
        if spec is None:
            raise ComponentNotFoundError(kind, name, list(bucket))
        return spec

has

has(kind: str, name: str) -> bool

Return whether name resolves for kind.

Source code in src\windlass\core\registry.py

def has(self, kind: str, name: str) -> bool:
    """Return whether ``name`` resolves for ``kind``."""
    try:
        self.get(kind, name)
    except (ComponentNotFoundError, RegistryError):
        return False
    return True

names

names(kind: str) -> list[str]

Return every registered name for kind, sorted.

Source code in src\windlass\core\registry.py

def names(self, kind: str) -> list[str]:
    """Return every registered name for ``kind``, sorted."""
    self.ensure_plugins_loaded()
    with self._lock:
        return sorted(self._bucket(kind))

specs

specs(kind: str) -> list[ComponentSpec]

Return every spec for kind, sorted by name.

Source code in src\windlass\core\registry.py

def specs(self, kind: str) -> list[ComponentSpec]:
    """Return every spec for ``kind``, sorted by name."""
    self.ensure_plugins_loaded()
    with self._lock:
        return [self._bucket(kind)[n] for n in sorted(self._bucket(kind))]

catalog

catalog() -> dict[str, list[ComponentSpec]]

Return every spec grouped by kind — powers windlass list.

Source code in src\windlass\core\registry.py

def catalog(self) -> dict[str, list[ComponentSpec]]:
    """Return every spec grouped by kind — powers ``windlass list``."""
    return {kind: self.specs(kind) for kind in self.kinds()}

create

create(kind: str, name: str, /, **config: Any) -> Any

Instantiate a registered component.

Registered defaults are merged underneath config, so caller arguments always win.

Parameters:

Name	Type	Description	Default
kind	str	Component kind.	required
name	str	Name or alias.	required
**config	Any	Constructor keyword arguments.	{}

Returns:

Type	Description
Any	The constructed component.

Raises:

Type	Description
ComponentNotFoundError	If the name is not registered.
ConfigurationError	If the constructor rejects the arguments.

Example

from windlass.core.registry import REGISTRY llm = REGISTRY.create("llm", "fake", responses=["hi"]) llm.complete("anything").content 'hi'

Source code in src\windlass\core\registry.py

def create(self, kind: str, name: str, /, **config: Any) -> Any:
    """Instantiate a registered component.

    Registered defaults are merged *underneath* ``config``, so caller
    arguments always win.

    Args:
        kind: Component kind.
        name: Name or alias.
        **config: Constructor keyword arguments.

    Returns:
        The constructed component.

    Raises:
        ComponentNotFoundError: If the name is not registered.
        ConfigurationError: If the constructor rejects the arguments.

    Example:
        >>> from windlass.core.registry import REGISTRY
        >>> llm = REGISTRY.create("llm", "fake", responses=["hi"])
        >>> llm.complete("anything").content
        'hi'
    """
    spec = self.get(kind, name)
    factory = spec.resolve()
    merged = {**spec.defaults, **config}
    try:
        return factory(**merged)
    except TypeError as exc:
        from windlass.core.exceptions import ConfigurationError

        raise ConfigurationError(
            f"Could not construct {kind} {name!r}: {exc}",
            hint=f"Check the accepted options for {getattr(factory, '__name__', factory)}.",
            context={"kind": kind, "name": name, "config": sorted(merged)},
        ) from exc

ensure_plugins_loaded

ensure_plugins_loaded() -> None

Discover entry-point plugins exactly once, on first lookup.

A no-op unless this registry was constructed with discover=True. :meth:load_plugins remains available on any registry for callers that want discovery on demand.

Source code in src\windlass\core\registry.py

def ensure_plugins_loaded(self) -> None:
    """Discover entry-point plugins exactly once, on first lookup.

    A no-op unless this registry was constructed with ``discover=True``.
    :meth:`load_plugins` remains available on any registry for callers that
    want discovery on demand.
    """
    if self._plugins_loaded:
        return
    with self._lock:
        if self._plugins_loaded:
            return
        self._plugins_loaded = True  # set first: discovery must not recurse
    self.load_plugins()

load_plugins

load_plugins(*, strict: bool = False) -> list[str]

Scan installed distributions for Windlass plugins.

Two entry-point conventions are supported:

• windlass.plugins — the value is a callable taking the registry; use this to register many components or to do custom setup.
• windlass.<kind> (e.g. windlass.chunker) — the value is the component class itself, registered under the entry-point name.

Parameters:

Name	Type	Description	Default
strict	bool	Raise on the first plugin that fails instead of warning.	False

Returns:

Type	Description
list[str]	Names of the plugins that loaded successfully.

Raises:

Type	Description
PluginError	Only when strict is True and a plugin fails.

Source code in src\windlass\core\registry.py

def load_plugins(self, *, strict: bool = False) -> list[str]:
    """Scan installed distributions for Windlass plugins.

    Two entry-point conventions are supported:

    * ``windlass.plugins`` — the value is a callable taking the registry;
      use this to register many components or to do custom setup.
    * ``windlass.<kind>`` (e.g. ``windlass.chunker``) — the value is the
      component class itself, registered under the entry-point name.

    Args:
        strict: Raise on the first plugin that fails instead of warning.

    Returns:
        Names of the plugins that loaded successfully.

    Raises:
        PluginError: Only when ``strict`` is True and a plugin fails.
    """
    from importlib.metadata import entry_points

    loaded: list[str] = []

    def _handle(err: Exception, label: str) -> None:
        if strict:
            raise PluginError(
                f"Plugin {label!r} failed to load: {err}", context={"plugin": label}
            ) from err
        _log.warning("Skipping Windlass plugin %r: %s", label, err)

    for ep in entry_points(group=PLUGIN_GROUP):
        try:
            hook = ep.load()
            hook(self)
            loaded.append(ep.name)
        except Exception as exc:
            _handle(exc, ep.name)

    for kind in self.kinds():
        for ep in entry_points(group=f"windlass.{kind}"):
            try:
                self.register_lazy(
                    kind,
                    ep.name,
                    ep.value.replace(":", ":", 1),
                    description=f"Plugin from {ep.value}",
                    origin="plugin",
                    override=True,
                )
                loaded.append(f"{kind}:{ep.name}")
            except Exception as exc:
                _handle(exc, f"{kind}:{ep.name}")

    return loaded

snapshot

snapshot() -> dict[str, dict[str, ComponentSpec]]

Return a shallow copy of the registry state (for tests).

Source code in src\windlass\core\registry.py

def snapshot(self) -> dict[str, dict[str, ComponentSpec]]:
    """Return a shallow copy of the registry state (for tests)."""
    with self._lock:
        return {kind: dict(bucket) for kind, bucket in self._specs.items()}

restore

restore(snapshot: dict[str, dict[str, ComponentSpec]]) -> None

Restore a previously captured :meth:snapshot (for tests).

Source code in src\windlass\core\registry.py

def restore(self, snapshot: dict[str, dict[str, ComponentSpec]]) -> None:
    """Restore a previously captured :meth:`snapshot` (for tests)."""
    with self._lock:
        self._specs = {kind: dict(bucket) for kind, bucket in snapshot.items()}

create

create(kind: str, name: str, /, **config: Any) -> Any

Instantiate a component from the global registry.

Thin module-level shortcut for :meth:Registry.create.

Source code in src\windlass\core\registry.py

def create(kind: str, name: str, /, **config: Any) -> Any:
    """Instantiate a component from the global registry.

    Thin module-level shortcut for :meth:`Registry.create`.
    """
    return REGISTRY.create(kind, name, **config)

available

available(kind: str) -> list[str]

List registered names for kind in the global registry.

Source code in src\windlass\core\registry.py

def available(kind: str) -> list[str]:
    """List registered names for ``kind`` in the global registry."""
    return REGISTRY.names(kind)

load_plugins

load_plugins(*, strict: bool = False) -> list[str]

Force plugin discovery on the global registry.

Source code in src\windlass\core\registry.py

def load_plugins(*, strict: bool = False) -> list[str]:
    """Force plugin discovery on the global registry."""
    return REGISTRY.load_plugins(strict=strict)