Create compute_provider_manager.py

compute_provider_manager.py

@@ -0,0 +1,115 @@
+# =====================================================
+# 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
+# =====================================================