windlass.interfaces.base

base

The base every Windlass component shares.

:class:Component gives all thirteen component kinds the same three things:

1. Identity — a name used in traces, errors and registry lookups.
2. Configuration — a validated config mapping, echoed in repr and traces so a misbehaving pipeline is debuggable from its logs alone.
3. Native access (Level 3) — :meth:Component.native hands back the underlying SDK object. Windlass simplifies libraries; it never hides them.

Subclasses implement async methods. The sync variants are derived once, here and in each interface, so the two APIs can never drift.

SupportsNative

Bases: ABC

Mixin declaring the Level 3 escape hatch.

Any adapter wrapping a third-party object should return it from :meth:native so advanced users can reach past Windlass without forking it.

native abstractmethod

native() -> Any

Return the underlying provider object.

Returns:

Type	Description
Any	The wrapped SDK client, index, graph or handle. Adapters with no
Any	third-party object behind them return self.

Example

from windlass.providers.llm.fake import FakeLLM FakeLLM().native() is not None True

Source code in src\windlass\interfaces\base.py

@abc.abstractmethod
def native(self) -> Any:
    """Return the underlying provider object.

    Returns:
        The wrapped SDK client, index, graph or handle. Adapters with no
        third-party object behind them return ``self``.

    Example:
        >>> from windlass.providers.llm.fake import FakeLLM
        >>> FakeLLM().native() is not None
        True
    """

Component

Component(*, name: str | None = None, **config: Any)

Bases: SupportsNative

Abstract base for every registrable Windlass component.

Parameters:

Name	Type	Description	Default
name	str | None	Identifier used in traces and error messages. Defaults to the class's registered name, or its class name.	None
**config	Any	Arbitrary component configuration, kept on :attr:config.	{}

Attributes:

Name	Type	Description
name		The component's identifier.
config	dict[str, Any]	The configuration it was constructed with.

Example

class Upper(Component): ... def apply(self, text: str) -> str: ... return text.upper() Upper(name="upper").name 'upper'

Source code in src\windlass\interfaces\base.py

def __init__(self, *, name: str | None = None, **config: Any) -> None:
    self.name = name or getattr(type(self), "provider_name", None) or type(self).__name__
    self.config: dict[str, Any] = dict(config)
    self._log = get_logger(f"{self.kind}.{self.name}")

option

option(key: str, default: Any = None) -> Any

Read a configuration value.

Parameters:

Name	Type	Description	Default
key	str	Configuration key.	required
default	Any	Returned when the key is absent.	None

Returns:

Type	Description
Any	The configured value or default.

Source code in src\windlass\interfaces\base.py

def option(self, key: str, default: Any = None) -> Any:
    """Read a configuration value.

    Args:
        key: Configuration key.
        default: Returned when the key is absent.

    Returns:
        The configured value or ``default``.
    """
    return self.config.get(key, default)

require_option

require_option(key: str) -> Any

Read a configuration value that must be present.

Parameters:

Name	Type	Description	Default
key	str	Configuration key.	required

Returns:

Type	Description
Any	The configured value.

Raises:

Type	Description
ConfigurationError	When the key is missing or None.

Source code in src\windlass\interfaces\base.py

def require_option(self, key: str) -> Any:
    """Read a configuration value that must be present.

    Args:
        key: Configuration key.

    Returns:
        The configured value.

    Raises:
        ConfigurationError: When the key is missing or ``None``.
    """
    value = self.config.get(key)
    if value is None:
        raise ConfigurationError(
            f"{type(self).__name__} requires the {key!r} option.",
            hint=f"Pass {key}=... when constructing the {self.kind}.",
            context={"component": self.name, "kind": self.kind, "option": key},
        )
    return value

with_config

with_config(**overrides: Any) -> Component

Return a new component of the same type with merged configuration.

Components are treated as immutable once built; reconfiguring produces a copy so a shared component cannot be mutated out from under another pipeline.

Parameters:

Name	Type	Description	Default
**overrides	Any	Configuration to merge over the current values.	{}

Returns:

Type	Description
Component	A new instance of the same class.

Source code in src\windlass\interfaces\base.py

def with_config(self, **overrides: Any) -> Component:
    """Return a new component of the same type with merged configuration.

    Components are treated as immutable once built; reconfiguring produces a
    copy so a shared component cannot be mutated out from under another
    pipeline.

    Args:
        **overrides: Configuration to merge over the current values.

    Returns:
        A new instance of the same class.
    """
    merged = {**self.config, **overrides}
    return type(self)(name=self.name, **merged)

aclose async

aclose() -> None

Release resources held by the component.

The default is a no-op. Adapters holding sockets, file handles or thread pools should override it. Safe to call more than once.

Source code in src\windlass\interfaces\base.py

async def aclose(self) -> None:
    """Release resources held by the component.

    The default is a no-op. Adapters holding sockets, file handles or thread
    pools should override it. Safe to call more than once.
    """

close

close() -> None

Blocking counterpart of :meth:aclose.

Source code in src\windlass\interfaces\base.py

def close(self) -> None:
    """Blocking counterpart of :meth:`aclose`."""
    from windlass.core.concurrency import run_sync

    run_sync(self.aclose())

native

native() -> Any

Return the wrapped provider object.

The default returns self, which is correct for pure-Python components that wrap nothing.

Source code in src\windlass\interfaces\base.py

def native(self) -> Any:
    """Return the wrapped provider object.

    The default returns ``self``, which is correct for pure-Python
    components that wrap nothing.
    """
    return self

describe

describe() -> dict[str, Any]

Return a JSON-safe summary of this component.

Used by tracing, by Pipeline.describe() and by the CLI.

Returns:

Type	Description
dict[str, Any]	A dict with kind, name, class and config.

Source code in src\windlass\interfaces\base.py

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe summary of this component.

    Used by tracing, by ``Pipeline.describe()`` and by the CLI.

    Returns:
        A dict with ``kind``, ``name``, ``class`` and ``config``.
    """
    return {
        "kind": self.kind,
        "name": self.name,
        "class": f"{type(self).__module__}.{type(self).__qualname__}",
        "config": {k: _safe(k, v) for k, v in self.config.items()},
    }