compute_provider_executor.py

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

"""
Compute Provider Executor.

Version 1.0
"""


class ComputeProviderExecutor:

    def __init__(self, manager):

        if manager is None:

            raise ValueError(
                "Manager is required"
            )

        self.manager = manager


    # =================================================
    # Execute Task
    # =================================================

    def execute(self, task):

        if task is None:

            return {
                "task_id": None,
                "status": "error",
                "result": None,
                "error": "Task is required",
            }

        task_type = task.get(
            "task_type"
        )

        if not task_type:

            return {
                "task_id": task.get(
                    "task_id"
                ),
                "status": "error",
                "result": None,
                "error": "Missing task_type",
            }

        provider = (
            self.manager.select_provider(
                task_type
            )
        )

        if provider is None:

            return {
                "task_id": task.get(
                    "task_id"
                ),
                "status": "error",
                "result": None,
                "error": (
                    "No provider available "
                    f"for task type: {task_type}"
                ),
            }

        return provider.execute(
            task
        )


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