operonx.core.registry

Resource hub, configuration storage, and op registry.

ResourceHub

ResourceHub

ResourceHub(storage: ConfigStorage, source_path: Optional[Path] = None)

Centralized registry for managing application resources.

Features: - Lazy loading: resources are initialized on first access - Pluggable storage: YAML, JSON, or custom backend - Extensible: external packages register their configs and factories

Example

hub = ResourceHub.from_yaml("configs/resources.yaml") llm = hub.get("llm:gpt-4o") stt = hub.get("triton:stt")

Or use global hub

from operonx.core.registry import ResourceHub llm = ResourceHub.instance().get("llm:gpt-4o")

Initialize hub with storage backend.

Parameters:

Name	Type	Description	Default
storage	ConfigStorage	Storage backend for configs	required
source_path	Optional[Path]	Absolute path of the file the storage was loaded from (set by from_yaml / from_json). None for in-memory or test-injected storage. Used in error messages to tell the user where to fix things.	None

Source code in operonx/core/registry/resource_hub.py

def __init__(self, storage: ConfigStorage, source_path: Optional[Path] = None):
    """Initialize hub with storage backend.

    Args:
        storage: Storage backend for configs
        source_path: Absolute path of the file the storage was loaded
            from (set by ``from_yaml`` / ``from_json``). ``None`` for
            in-memory or test-injected storage. Used in error messages
            to tell the user where to fix things.
    """
    self._storage = storage
    self._cache: Dict[str, CacheEntry] = {}
    self._source_path: Optional[Path] = source_path

Attributes

source_path property

source_path: Optional[Path]

Absolute path the hub was loaded from, or None.

Functions

from_yaml classmethod

from_yaml(path: str | Path) -> ResourceHub

Create hub with YAML file storage.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to YAML config file	required

Returns:

Type	Description
ResourceHub	ResourceHub instance

Source code in operonx/core/registry/resource_hub.py

@classmethod
def from_yaml(cls, path: str | Path) -> "ResourceHub":
    """Create hub with YAML file storage.

    Args:
        path: Path to YAML config file

    Returns:
        ResourceHub instance
    """
    abs_path = Path(path).resolve()
    storage = YamlConfigStorage(abs_path)
    return cls(storage, source_path=abs_path)

from_json classmethod

from_json(path: str | Path) -> ResourceHub

Create hub with JSON file storage.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to JSON config file	required

Returns:

Type	Description
ResourceHub	ResourceHub instance

Source code in operonx/core/registry/resource_hub.py

@classmethod
def from_json(cls, path: str | Path) -> "ResourceHub":
    """Create hub with JSON file storage.

    Args:
        path: Path to JSON config file

    Returns:
        ResourceHub instance
    """
    from .storage import JsonConfigStorage

    abs_path = Path(path).resolve()
    storage = JsonConfigStorage(abs_path)
    return cls(storage, source_path=abs_path)

auto classmethod

auto() -> Optional[ResourceHub]

Try to install ResourceHub from ./resources.yaml in CWD.

Behavior: - If a hub is already installed, return it unchanged (idempotent). - If ./resources.yaml exists, load it, install the singleton, and return the new hub. - If not found, emit a :class:ResourceHubWarning naming the path checked and return None. No singleton is installed.

Never raises. The warning is the early signal that setup is incomplete; silent miss would defer the problem to first resource resolution.

Source code in operonx/core/registry/resource_hub.py

@classmethod
def auto(cls) -> Optional["ResourceHub"]:
    """Try to install ResourceHub from ``./resources.yaml`` in CWD.

    Behavior:
    - If a hub is already installed, return it unchanged (idempotent).
    - If ``./resources.yaml`` exists, load it, install the singleton,
      and return the new hub.
    - If not found, emit a :class:`ResourceHubWarning` naming the path
      checked and return ``None``. No singleton is installed.

    Never raises. The warning is the early signal that setup is
    incomplete; silent miss would defer the problem to first
    resource resolution.
    """
    if cls._instance is not None:
        return cls._instance

    candidate = (Path.cwd() / "resources.yaml").resolve()
    if not candidate.exists():
        warnings.warn(
            f"No resources.yaml found at {candidate}. "
            "ResourceHub not installed; provider ops will fail at "
            "resolution. Call ResourceHub.from_yaml(<path>) with an "
            "explicit path if your file lives elsewhere.",
            ResourceHubWarning,
            stacklevel=2,
        )
        return None

    hub = cls.from_yaml(candidate)
    cls.set_instance(hub)
    return hub

instance classmethod

instance() -> ResourceHub

Get the global ResourceHub singleton.

Raises:

Type	Description
RuntimeError	If no hub has been loaded. The message points at operonx.bootstrap() and ResourceHub.from_yaml(...) as the two ways to install one.

Source code in operonx/core/registry/resource_hub.py

@classmethod
def instance(cls) -> "ResourceHub":
    """Get the global ResourceHub singleton.

    Raises:
        RuntimeError: If no hub has been loaded. The message points at
            ``operonx.bootstrap()`` and ``ResourceHub.from_yaml(...)``
            as the two ways to install one.
    """
    if cls._instance is None:
        raise RuntimeError(
            "ResourceHub not initialized. Install one before resolving resources:\n"
            "  import operonx\n"
            "  operonx.bootstrap()                            # auto-discover ./resources.yaml + .env\n"
            "Or with an explicit path:\n"
            "  from operonx.core.registry import ResourceHub\n"
            "  ResourceHub.set_instance(ResourceHub.from_yaml('path/to/resources.yaml'))"
        )
    return cls._instance

set_instance classmethod

set_instance(hub: ResourceHub) -> None

Install hub as the global singleton (used by Operon(...)).

Source code in operonx/core/registry/resource_hub.py

@classmethod
def set_instance(cls, hub: "ResourceHub") -> None:
    """Install *hub* as the global singleton (used by ``Operon(...)``)."""
    cls._instance = hub

reset_instance classmethod

reset_instance() -> None

Clear the global singleton. For tests only.

Source code in operonx/core/registry/resource_hub.py

@classmethod
def reset_instance(cls) -> None:
    """Clear the global singleton. For tests only."""
    cls._instance = None

keys

keys() -> List[str]

Return all registered keys (loads all configs from storage).

Source code in operonx/core/registry/resource_hub.py

def keys(self) -> List[str]:
    """Return all registered keys (loads all configs from storage)."""
    all_configs = self._storage.load_all()

    for key, config_data in all_configs.items():
        if key not in self._cache:
            # Use _load_config which handles category resolution
            self._load_config(key)

    return list(self._cache.keys())

has

has(key: str) -> bool

Check if resource exists in registry.

Source code in operonx/core/registry/resource_hub.py

def has(self, key: str) -> bool:
    """Check if resource exists in registry."""
    if key in self._cache:
        return True
    # Try loading from storage
    return self._load_config(key) is not None

get

get(key: str) -> Any

Get resource instance by key (lazy load on first access).

Parameters:

Name	Type	Description	Default
key	str	Registry key of resource	required

Returns:

Type	Description
Any	Initialized resource instance

Raises:

Type	Description
KeyError	If key not found or resource failed to initialize. The message includes source_path and available keys so the user can tell which fix to apply.
EnvVarUnsetError	If the resource references a ${VAR} whose env var is unset. Subclass of RuntimeError for backwards compatibility.

Source code in operonx/core/registry/resource_hub.py

def get(self, key: str) -> Any:
    """Get resource instance by key (lazy load on first access).

    Args:
        key: Registry key of resource

    Returns:
        Initialized resource instance

    Raises:
        KeyError: If key not found or resource failed to initialize.
            The message includes ``source_path`` and available keys
            so the user can tell which fix to apply.
        EnvVarUnsetError: If the resource references a ``${VAR}``
            whose env var is unset. Subclass of ``RuntimeError`` for
            backwards compatibility.
    """
    # Return cached instance if available
    if key in self._cache and self._cache[key].instance is not None:
        instance = self._cache[key].instance
        # Refresh keycloak token if applicable (LLM resources)
        self._refresh_keycloak(instance)
        return instance

    # Load config from storage. ``EnvVarUnsetError`` from missing
    # ``${VAR}`` interpolation propagates as-is (branch 4) — distinct
    # from the "key not found" path below.
    config = self._load_config(key)
    if not config:
        raise KeyError(self._not_found_message(key))

    # Resolve keycloak token if configured (api_key: "keycloak:xxx")
    resolved_config = self._resolve_keycloak(config)
    create_config = resolved_config or config

    # Lazy initialize resource
    try:
        instance = REGISTRY.create(create_config)
    except Exception as e:
        LOGGER.warning("Failed to create resource '%s': %s", key, e)
        raise KeyError(f"Resource '{key}' failed to initialize: {e}") from e

    if instance is None:
        raise KeyError(f"Cannot create resource for '{key}': factory returned None")

    # Attach keycloak provider for future token refresh
    if resolved_config is not None:
        keycloak_name = config.api_key[9:]
        instance._keycloak_provider = self.keycloak(keycloak_name)
        instance._original_config = config

    self._cache[key].instance = instance
    LOGGER.debug("Lazy loaded resource: %s", key)

    return self._cache[key].instance

get_config

get_config(key: str) -> YamlModel

Get config object of resource.

Parameters:

Name	Type	Description	Default
key	str	Registry key	required

Returns:

Type	Description
YamlModel	Resource config

Raises:

Type	Description
KeyError	If key not found

Source code in operonx/core/registry/resource_hub.py

def get_config(self, key: str) -> YamlModel:
    """Get config object of resource.

    Args:
        key: Registry key

    Returns:
        Resource config

    Raises:
        KeyError: If key not found
    """
    config = self._load_config(key)
    if not config:
        raise KeyError(self._not_found_message(key))
    return config

register

register(config: YamlModel, registry_key: Optional[str] = None) -> str

Register new resource config.

Parameters:

Name	Type	Description	Default
config	YamlModel	Resource config object	required
registry_key	Optional[str]	Custom key (auto-generated if not provided)	None

Returns:

Type	Description
str	Registry key used

Source code in operonx/core/registry/resource_hub.py

def register(self, config: YamlModel, registry_key: Optional[str] = None) -> str:
    """Register new resource config.

    Args:
        config: Resource config object
        registry_key: Custom key (auto-generated if not provided)

    Returns:
        Registry key used
    """
    if not registry_key:
        registry_key = self._key_of(config)

    # Create instance immediately
    instance = REGISTRY.create(config)
    self._cache[registry_key] = CacheEntry(config=config, instance=instance)

    # Persist to storage (no type field needed - category is in the key)
    config_dict = json.loads(config.model_dump_json(exclude_none=True))
    self._storage.save(registry_key, config_dict)

    LOGGER.debug("Registered: %s", registry_key)
    return registry_key

remove

remove(key: str) -> bool

Remove resource from registry.

Parameters:

Name	Type	Description	Default
key	str	Registry key to remove	required

Returns:

Type	Description
bool	True if removed, False if not found

Source code in operonx/core/registry/resource_hub.py

def remove(self, key: str) -> bool:
    """Remove resource from registry.

    Args:
        key: Registry key to remove

    Returns:
        True if removed, False if not found
    """
    if key not in self._cache:
        # Try loading first
        if not self._load_config(key):
            return False

    if key in self._cache:
        del self._cache[key]

    self._storage.remove(key)
    LOGGER.debug("Removed: %s", key)
    return True

clear

clear()

Clear all resources from registry and storage.

Source code in operonx/core/registry/resource_hub.py

def clear(self):
    """Clear all resources from registry and storage."""
    keys = list(self._cache.keys())
    self._cache.clear()
    for key in keys:
        self._storage.remove(key)
    LOGGER.debug("Cleared all resources")

close

close()

Close storage connection and cleanup.

Source code in operonx/core/registry/resource_hub.py

def close(self):
    """Close storage connection and cleanup."""
    if self._storage:
        self._storage.close()

keycloak

keycloak(key: str) -> KeycloakTokenProvider

Get KeycloakTokenProvider by key.

Parameters:

Name	Type	Description	Default
key	str	Keycloak config identifier (e.g., 'myapp')	required

Returns:

Type	Description
KeycloakTokenProvider	KeycloakTokenProvider instance with get_token() method

Source code in operonx/core/registry/resource_hub.py

def keycloak(self, key: str) -> "KeycloakTokenProvider":
    """Get KeycloakTokenProvider by key.

    Args:
        key: Keycloak config identifier (e.g., 'myapp')

    Returns:
        KeycloakTokenProvider instance with get_token() method
    """
    return self.get(f"keycloak:{key}")

warmup async

warmup(key: str, **kwargs: Any) -> None

Pre-warm a resource's connection (e.g. LLM prompt cache).

Forces lazy-load of the resource, then calls its warmup() method if it exists. For LLM providers this typically sends a minimal API request to establish the TCP+TLS connection and optionally seed the server-side prompt cache.

Parameters:

Name	Type	Description	Default
key	str	Registry key (e.g. "llm:default")	required
**kwargs	Any	Passed to the resource's warmup() method (e.g. system_prompt="..." for LLMs)	{}

Example::

hub = ResourceHub.instance()
await hub.warmup("llm:default", system_prompt=sys_prompt)

Source code in operonx/core/registry/resource_hub.py

async def warmup(self, key: str, **kwargs: Any) -> None:
    """Pre-warm a resource's connection (e.g. LLM prompt cache).

    Forces lazy-load of the resource, then calls its ``warmup()``
    method if it exists. For LLM providers this typically sends a
    minimal API request to establish the TCP+TLS connection and
    optionally seed the server-side prompt cache.

    Args:
        key: Registry key (e.g. ``"llm:default"``)
        **kwargs: Passed to the resource's ``warmup()`` method
                  (e.g. ``system_prompt="..."`` for LLMs)

    Example::

        hub = ResourceHub.instance()
        await hub.warmup("llm:default", system_prompt=sys_prompt)
    """
    try:
        resource = self.get(key)
        if hasattr(resource, "warmup"):
            result = resource.warmup(**kwargs)
            # Support both sync and async warmup methods
            if hasattr(result, "__await__"):
                await result
            LOGGER.debug("Warmup completed: %s", key)
        else:
            LOGGER.debug("Resource '%s' has no warmup method, skipping", key)
    except Exception as exc:
        LOGGER.warning("Warmup failed for '%s' (non-fatal): %s", key, exc)

health_check

health_check(keys: Optional[List[str]] = None) -> HealthCheckResult

Check health of all or specified resources.

Parameters:

Name	Type	Description	Default
keys	Optional[List[str]]	Optional list of keys to check. If None, checks all.	None

Returns:

Type	Description
HealthCheckResult	HealthCheckResult with status of each resource

Example

result = hub.health_check() if not result.healthy: print(f"Unhealthy resources: {result.failed}")

Source code in operonx/core/registry/resource_hub.py

def health_check(self, keys: Optional[List[str]] = None) -> HealthCheckResult:
    """Check health of all or specified resources.

    Args:
        keys: Optional list of keys to check. If None, checks all.

    Returns:
        HealthCheckResult with status of each resource

    Example:
        result = hub.health_check()
        if not result.healthy:
            print(f"Unhealthy resources: {result.failed}")
    """
    check_keys = keys if keys else self.keys()
    results: Dict[str, bool] = {}
    errors: Dict[str, str] = {}

    for key in check_keys:
        try:
            # Try to load the resource
            self.get(key)
            results[key] = True
        except Exception as e:
            results[key] = False
            errors[key] = str(e)
            LOGGER.warning("Health check failed for '%s': %s", key, e)

    return HealthCheckResult(
        results=results,
        errors=errors,
    )

Errors and warnings

EnvVarUnsetError

Bases: RuntimeError

Raised when a required ${VAR} reference cannot be resolved.

Subclasses RuntimeError so existing except RuntimeError callers continue to work — this preserves backwards-compatible behavior with the previous _raise_missing_env_vars helper.

ResourceHubWarning

Bases: UserWarning

Warning emitted by ResourceHub setup (auto / from_yaml / bootstrap).

Subclassing UserWarning lets users silence ResourceHub-specific warnings without muting unrelated UserWarning traffic::

import warnings
from operonx.core.registry import ResourceHubWarning
warnings.filterwarnings("ignore", category=ResourceHubWarning)

Bootstrap state

BOOTSTRAP_ENV_PATHS is populated by operonx.bootstrap and surfaces in the EnvVarUnsetError message so you can see exactly which .env paths were searched.