climate_ref.provider_registry #

@frozen
class ProviderRegistry:
    """
    Registry for the currently active providers

    In some cases we can't directly import a provider and it's diagnostics,
    in this case we need to proxy the diagnostics.
    """

    providers: list[DiagnosticProvider] = field(factory=list)

    def get(self, slug: str) -> DiagnosticProvider:
        """
        Retrieve a provider by name

        Parameters
        ----------
        slug
            Slug of the provider of interest

        Raises
        ------
        KeyError
            A provider with the matching slug has not been registered

        Returns
        -------
            The requested provider
        """
        for p in self.providers:
            if p.slug == slug:
                return p

        raise KeyError(f"No provider with slug matching: {slug}")

    def get_metric(self, provider_slug: str, diagnostic_slug: str) -> "Diagnostic":
        """
        Retrieve a diagnostic by name

        This is a convenience method to retrieve a diagnostic from a provider

        Parameters
        ----------
        provider_slug :
            Slug of the provider of interest
        diagnostic_slug
            Slug of the diagnostic of interest

        Raises
        ------
        KeyError
            If the provider/diagnostic with the given slugs is not found.

        Returns
        -------
            The requested diagnostic.
        """
        return self.get(provider_slug).get(diagnostic_slug)

    @staticmethod
    def build_from_config(config: Config, db: Database) -> "ProviderRegistry":
        """
        Create a ProviderRegistry instance using information from the database

        Parameters
        ----------
        config
            Configuration object
        db
            Database instance

        Returns
        -------
        :
            A new ProviderRegistry instance
        """
        providers = []
        for provider_info in config.diagnostic_providers:
            provider = import_provider(provider_info.provider)
            provider.configure(config)
            providers.append(provider)

        with db.session.begin():
            for provider in providers:
                _register_provider(db, provider)

        return ProviderRegistry(providers=providers)