compute_provider_manager.py

# =====================================================
# Apckeyl Framework
# Version 1.0
# compute_provider_manager.py
# =====================================================

"""
Compute Provider Manager.

Version 1.0
"""

from compute_provider_factory import (
    ComputeProviderFactory,
)

from compute_provider_selector import (
    ComputeProviderSelector,
)


# =====================================================
# Manager
# =====================================================

class ComputeProviderManager:

    def __init__(self, registry):

        if registry is None:

            raise ValueError(
                "Registry is required"
            )

        self.factory = (
            ComputeProviderFactory(
                registry
            )
        )

        self.selector = (
            ComputeProviderSelector(
                registry
            )
        )


    # =================================================
    # Register Provider
    # =================================================

    def register_provider(
        self,
        provider,
    ):

        return (
            self.factory.create_provider(
                provider
            )
        )


    # =================================================
    # Get Provider
    # =================================================

    def get_provider(
        self,
        provider_id: str,
    ):

        return (
            self.factory.get_provider(
                provider_id
            )
        )


    # =================================================
    # Select Provider
    # =================================================

    def select_provider(
        self,
        task_type: str,
    ):

        return (
            self.selector.select(
                task_type
            )
        )


    # =================================================
    # Check Availability
    # =================================================

    def is_available(
        self,
        task_type: str,
    ) -> bool:

        return (
            self.selector.is_available(
                task_type
            )
        )


# =====================================================
# End of File
# =====================================================