orchestrata_ultimate.providers.registry

Reference › orchestrata_ultimate › providers › registry

registry stores the providers known to the manager, with their configurations and model pricing (specification 011, requirements 024 and 028). ProviderRegistry is used by provider_manager and router. Registration is dynamic: register adds a provider with its ProviderConfig and deregister removes it, returning whether it was found. Lookups are get_provider, get_config, get_pricing (for one provider and model), resolve_model (a model name to its provider and model pair) and find_by_capability, which returns the provider and model entries that support all of the requested capabilities. The properties provider_names and model_count report what is registered.

At a glance

Class diagram

Dependency diagram

API

Classes

ProviderRegistry

Decorators: traces(spec='011', story='US-001', reqs=['REQ-011-024', 'REQ-011-028']) · Line 22

Registry of provider adapters and their configurations.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/providers/registry.py

# mypy: ignore-errors
"""Provider registry.

SPEC-011 FR-005/FR-006: Stores registered providers, their configurations,
and model metadata. Supports dynamic registration and deregistration.
"""

import logging

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    ModelCapability,
    ModelPricing,
    ProviderConfig,
    ProviderProtocol,
)

logger = logging.getLogger(__name__)


@traces(spec="011", story="US-001", reqs=["REQ-011-024", "REQ-011-028"])
class ProviderRegistry:
    """Registry of provider adapters and their configurations.

    Stores provider instances, configs, and model metadata.
    Supports capability-based lookup and model aliasing.
    """

    def __init__(self):
        self._providers: dict[str, ProviderProtocol] = {}
        self._configs: dict[str, ProviderConfig] = {}
        self._model_aliases: dict[str, tuple] = {}  # alias -> (provider, model)

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-024"])
    def register(
        self,
        name: str,
        provider: ProviderProtocol,
        config: ProviderConfig,
    ) -> None:
        """Register a provider with its configuration."""
        self._providers[name] = provider
        self._configs[name] = config
        # Index models
        for model_name in config.models:
            self._model_aliases[model_name] = (name, model_name)
        logger.info("Registered provider: %s with %d models", name, len(config.models))

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-024"])
    def deregister(self, name: str) -> bool:
        """Remove a provider. Returns True if found and removed."""
        if name not in self._providers:
            return False
        config = self._configs.get(name)
        if config:
            for model_name in config.models:
                self._model_aliases.pop(model_name, None)
        del self._providers[name]
        del self._configs[name]
        logger.info("Deregistered provider: %s", name)
        return True

    def get_provider(self, name: str) -> ProviderProtocol | None:
        """Get a provider by name."""
        return self._providers.get(name)

    def get_config(self, name: str) -> ProviderConfig | None:
        """Get a provider's configuration."""
        return self._configs.get(name)

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-028"])
    def get_pricing(self, provider_name: str, model_name: str) -> ModelPricing | None:
        """Get pricing for a specific model from a provider."""
        config = self._configs.get(provider_name)
        if not config:
            return None
        return config.pricing.get(model_name)

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-007"])
    def resolve_model(self, model_name: str) -> tuple | None:
        """Resolve a model name to (provider_name, model_name)."""
        return self._model_aliases.get(model_name)

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-007"])
    def find_by_capability(
        self,
        capabilities: list[ModelCapability],
    ) -> list[tuple]:
        """Find providers+models that support all requested capabilities.

        Returns list of (provider_name, model_name, ModelPricing) tuples.
        """
        results = []
        for provider_name, config in self._configs.items():
            for model_name, pricing in config.pricing.items():
                model_caps = set(pricing.capabilities)
                required_caps = set(capabilities)
                if required_caps.issubset(model_caps):
                    results.append((provider_name, model_name, pricing))
        return results

    @property
    def provider_names(self) -> list[str]:
        """List of registered provider names."""
        return list(self._providers.keys())

    @property
    def model_count(self) -> int:
        """Total number of registered models across all providers."""
        return len(self._model_aliases)