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 (REGISTRY.register, found through an operonx.resources entry point when nothing imported the package)

Example

hub = ResourceHub.from_yaml("configs/resources.yaml") llm = hub.get("llm:gpt-4o") index = hub.get("vector_store:docs")

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
    #: In-memory ``alias -> real key`` map. See :meth:`alias`.
    self._aliases: Dict[str, str] = {}

Attributes¶

source_path `property` ¶

source_path: Optional[Path]

Absolute path the hub was loaded from, or None.

Methods:¶

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

alias ¶

alias(alias_key: str, target_key: str) -> None

Point alias_key at an existing resource, in memory only.

For naming a role at the call site while an operator still chooses the resource that fills it::

hub.alias("llm:scanner", f"llm:{os.environ['LLM_RESOURCE_KEY']}")
# graph stays literal: LLMOp.of(resource="scanner", ...)

That keeps the wiring readable to tooling — which can only see a literal — without freezing the model choice into the graph.

Not :meth:register: that persists through to storage and rewrites resources.yaml, losing its comments. This touches nothing on disk and lasts for the life of the hub.

Resolution is one hop, applied on every lookup, so re-aliasing later repoints existing call sites. A cached instance under the old target is unaffected — it is keyed by the real name.

Parameters:

Name	Type	Description	Default
`alias_key`	`str`	The name call sites use, e.g. `"llm:scanner"`.	required
`target_key`	`str`	An existing resource key, e.g. `"llm:db-gemini-3-flash"`.	required

Raises:

Type	Description
`ValueError`	The alias would point at itself, or target_key is itself an alias — one hop only, so a chain that could silently become a cycle is refused at declaration time rather than hanging at first use.

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

def alias(self, alias_key: str, target_key: str) -> None:
    """Point ``alias_key`` at an existing resource, in memory only.

    For naming a *role* at the call site while an operator still
    chooses the resource that fills it::

        hub.alias("llm:scanner", f"llm:{os.environ['LLM_RESOURCE_KEY']}")
        # graph stays literal: LLMOp.of(resource="scanner", ...)

    That keeps the wiring readable to tooling — which can only see a
    literal — without freezing the model choice into the graph.

    **Not** :meth:`register`: that persists through to storage and
    rewrites ``resources.yaml``, losing its comments. This touches
    nothing on disk and lasts for the life of the hub.

    Resolution is one hop, applied on every lookup, so re-aliasing
    later repoints existing call sites. A cached *instance* under the
    old target is unaffected — it is keyed by the real name.

    Args:
        alias_key: The name call sites use, e.g. ``"llm:scanner"``.
        target_key: An existing resource key, e.g. ``"llm:db-gemini-3-flash"``.

    Raises:
        ValueError: The alias would point at itself, or *target_key*
            is itself an alias — one hop only, so a chain that could
            silently become a cycle is refused at declaration time
            rather than hanging at first use.
    """
    if alias_key == target_key:
        raise ValueError(f"alias {alias_key!r} cannot point at itself")
    if target_key in self._aliases:
        raise ValueError(
            f"alias target {target_key!r} is itself an alias "
            f"(-> {self._aliases[target_key]!r}); point {alias_key!r} at the "
            "real key instead — aliases resolve one hop only"
        )
    self._aliases[alias_key] = target_key
    LOGGER.debug("Aliased: %s -> %s", alias_key, target_key)

aliases ¶

aliases() -> Dict[str, str]

A copy of the alias -> real key map.

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

def aliases(self) -> Dict[str, str]:
    """A copy of the ``alias -> real key`` map."""
    return dict(self._aliases)

unalias ¶

unalias(alias_key: str) -> bool

Drop an alias. Returns True if one was removed.

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

def unalias(self, alias_key: str) -> bool:
    """Drop an alias. Returns True if one was removed."""
    return self._aliases.pop(alias_key, None) is not None

keys ¶

keys() -> List[str]

Every declared key: those in storage, then any registered in memory only. Parses nothing — :meth:get reports a key that does not build, and :meth:health_check reports all of them.

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

def keys(self) -> List[str]:
    """Every declared key: those in storage, then any registered in
    memory only. Parses nothing — :meth:`get` reports a key that does
    not build, and :meth:`health_check` reports all of them."""
    declared = list(self._storage.load_all())
    return declared + [k for k in self._cache if k not in declared]

declares ¶

declares(key: str) -> bool

Whether key is configured, without loading it.

Unlike :meth:has this parses nothing and caches nothing: a check made before a project has imported the module that registers a custom category must not leave that key cached as a raw dict that :meth:get then refuses.

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

def declares(self, key: str) -> bool:
    """Whether *key* is configured, without loading it.

    Unlike :meth:`has` this parses nothing and caches nothing: a check
    made before a project has imported the module that registers a
    custom category must not leave that key cached as a raw dict that
    :meth:`get` then refuses."""
    key = self._resolve_alias(key)
    return key in self._cache or self._storage.load_one(key) is not None

has ¶

has(key: str) -> bool

Whether key is configured. The same question as :meth:declares: it parses nothing, so it neither fails on a category not registered yet nor caches anything.

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

def has(self, key: str) -> bool:
    """Whether *key* is configured. The same question as
    :meth:`declares`: it parses nothing, so it neither fails on a
    category not registered yet nor caches anything."""
    return self.declares(key)

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, its config does not parse, or the resource failed to initialize. The message says which, with `source_path` (and the available keys when the key is missing) so the user can tell which fix to apply.
`ResourceCategoryError`	A `KeyError` for a key whose category no package registers.
`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, its config does not parse, or the
            resource failed to initialize. The message says which,
            with ``source_path`` (and the available keys when the key
            is missing) so the user can tell which fix to apply.
        ResourceCategoryError: A ``KeyError`` for a key whose category
            no package registers.
        EnvVarUnsetError: If the resource references a ``${VAR}``
            whose env var is unset. Subclass of ``RuntimeError`` for
            backwards compatibility.
    """
    key = self._resolve_alias(key)

    # Return cached instance if available
    if key in self._cache and self._cache[key].instance is not None:
        instance = self._cache[key].instance
        # Refresh the bearer token if this resource carries a provider
        self._refresh_token(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 config is None:
        raise KeyError(self._not_found_message(key))

    # Resolve a token reference if configured
    # (api_key: "keycloak:xxx" or "oauth2:xxx")
    resolved_config = self._resolve_token_ref(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 the provider so a later cache hit can refresh the token.
    ref = _split_token_ref(getattr(config, "api_key", None))
    if resolved_config is not None and ref is not None:
        provider = self.get(f"{ref[0]}:{ref[1]}")
        instance._token_provider = provider
        # Back-compat alias: this attribute predates oauth2 and is
        # asserted on by name, so both point at the same object.
        instance._keycloak_provider = provider
        instance._original_config = config

    limit = self._cache[key].rate_limit
    if limit is not None:
        from .rate_limit import limit_instance

        limit_instance(instance, limit, key=key)
    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 config is None:
        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 not self.declares(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}")

oauth2 ¶

oauth2(key: str) -> OAuth2TokenProvider

Get an OAuth2TokenProvider by key.

Parameters:

Name	Type	Description	Default
`key`	`str`	OAuth2 config identifier (e.g. `'databricks'`).	required

Returns:

Type	Description
`OAuth2TokenProvider`	OAuth2TokenProvider instance with a `get_token()` method.

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

def oauth2(self, key: str) -> "OAuth2TokenProvider":  # noqa: F821
    """Get an OAuth2TokenProvider by key.

    Args:
        key: OAuth2 config identifier (e.g. ``'databricks'``).

    Returns:
        OAuth2TokenProvider instance with a ``get_token()`` method.
    """
    return self.get(f"oauth2:{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,
    *,
    probe: bool = False,
    timeout: float = 2.0,
) -> HealthCheckResult

Check all or specified resources: does each build, and answer?

Parameters:

Name	Type	Description	Default
`keys`	`Optional[List[str]]`	Keys to check. None checks every configured resource.	`None`
`probe`	`bool`	Also open a TCP connection to whatever address the config points at. Building a client opens no socket, so without this a hub full of corporate endpoints reports healthy on a laptop with the VPN down — and the real symptom arrives later as every call timing out one at a time. Resources with no address (in-memory, filesystem) are built and not probed. Off by default: this method has always meant "does it construct", and flipping that would quietly change the answer for every existing caller. `require_reachable` turns it on.	`False`
`timeout`	`float`	Seconds to wait for the connection. Short on purpose: the point is to fail at startup rather than across a batch, so a slow answer is as useful to us as no answer.	`2.0`

Returns:

Type	Description
`HealthCheckResult`	HealthCheckResult. Never raises for an unhealthy resource — see
`HealthCheckResult`	`require_reachable` for the version that stops the flow.

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,
    *,
    probe: bool = False,
    timeout: float = 2.0,
) -> HealthCheckResult:
    """Check all or specified resources: does each build, and answer?

    Args:
        keys: Keys to check. None checks every configured resource.
        probe: Also open a TCP connection to whatever address the
            config points at. Building a client opens no socket, so
            without this a hub full of corporate endpoints reports
            healthy on a laptop with the VPN down — and the real
            symptom arrives later as every call timing out one at a
            time. Resources with no address (in-memory, filesystem)
            are built and not probed.

            Off by default: this method has always meant "does it
            construct", and flipping that would quietly change the
            answer for every existing caller. ``require_reachable``
            turns it on.
        timeout: Seconds to wait for the connection. Short on purpose:
            the point is to fail at startup rather than across a batch,
            so a slow answer is as useful to us as no answer.

    Returns:
        HealthCheckResult. Never raises for an unhealthy resource — see
        ``require_reachable`` for the version that stops the flow.

    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:
            instance = self.get(key)
        except Exception as e:
            results[key] = False
            errors[key] = str(e)
            LOGGER.warning("Health check failed for '%s': %s", key, e)
            continue

        if not probe:
            results[key] = True
            continue

        reason = self._probe_endpoint(key, instance, timeout)
        results[key] = reason is None
        if reason is not None:
            errors[key] = reason
            LOGGER.warning("Resource '%s' unreachable: %s", key, reason)

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

require_reachable ¶

require_reachable(*keys: str, timeout: float = 2.0) -> HealthCheckResult

Check the named resources and raise if any cannot be reached.

No arguments checks every configured resource::

hub.require_reachable()                       # all of them
hub.require_reachable("llm:scorer", "embedding:corpus")

Every failure is collected before raising, so one run names everything that is missing rather than the first thing.

Raises:

Type	Description
`ResourceUnreachable`	when any checked resource failed.

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

def require_reachable(
    self,
    *keys: str,
    timeout: float = 2.0,
) -> HealthCheckResult:
    """Check the named resources and **raise** if any cannot be reached.

    No arguments checks every configured resource::

        hub.require_reachable()                       # all of them
        hub.require_reachable("llm:scorer", "embedding:corpus")

    Every failure is collected before raising, so one run names
    everything that is missing rather than the first thing.

    Raises:
        ResourceUnreachable: when any checked resource failed.
    """
    result = self.health_check(list(keys) or None, probe=True, timeout=timeout)
    if not result.healthy:
        raise ResourceUnreachable({k: result.errors.get(k, "unhealthy") for k in result.failed})
    return result

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)

ResourceCategoryError ¶

Bases: KeyError

A resource key's category has no config class to parse it with.

Either no package registers the category, or the package that should (an operonx.resources entry point) failed to. Subclasses KeyError because :meth:ResourceHub.get has always raised that for a resource it cannot build, and callers catch it.

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.

operonx.core.registry¶

ResourceHub¶

ResourceHub ¶

Or use global hub¶

Attributes¶

source_path property ¶

Methods:¶

from_yaml classmethod ¶

from_json classmethod ¶

auto classmethod ¶

instance classmethod ¶

set_instance classmethod ¶

reset_instance classmethod ¶

alias ¶

aliases ¶

unalias ¶

keys ¶

declares ¶

has ¶

get ¶

get_config ¶

register ¶

remove ¶

clear ¶

close ¶

keycloak ¶

oauth2 ¶

warmup async ¶

health_check ¶

require_reachable ¶

Errors and warnings¶

EnvVarUnsetError ¶

ResourceHubWarning ¶

ResourceCategoryError ¶

Bootstrap state¶

source_path `property` ¶

from_yaml `classmethod` ¶

from_json `classmethod` ¶

auto `classmethod` ¶

instance `classmethod` ¶

set_instance `classmethod` ¶

reset_instance `classmethod` ¶

warmup `async` ¶