server/tasks/task_registry.py

"""Task registry for the environment."""


from typing import Dict, Type

from server.tasks.base import BaseTask
from server.tasks.task_1_build_errors import DockerfileSyntaxTask
from server.tasks.task_2_docker_runtime import DockerfileRuntimeTask
from server.tasks.task_3_workflow_syntax import WorkflowSyntaxStructureTask
from server.tasks.task_4_workflow_secrets_permissions import WorkflowSecretsPermissionsTask
from server.tasks.task_5_ci_docker_integration import CIDockerIntegrationTask
from server.tasks.task_6_multi_stage_matrix import MultiStageMatrixTask
from server.tasks.k8s_pod import K8sPodTask
from server.tasks.k8s_networking import K8sNetworkingTask
from server.tasks.pipeline_build_deploy import PipelineBuildDeployTask
from server.tasks.pipeline_full import PipelineFullTask

TASK_REGISTRY: Dict[str, Type[BaseTask]] = {
    "dockerfile_syntax": DockerfileSyntaxTask,
    "dockerfile_runtime": DockerfileRuntimeTask,
    "workflow_syntax_structure": WorkflowSyntaxStructureTask,
    "workflow_secrets_permissions": WorkflowSecretsPermissionsTask,
    "ci_docker_integration": CIDockerIntegrationTask,
    "multi_stage_pipeline_matrix": MultiStageMatrixTask,
    "k8s_pod_failures": K8sPodTask,
    "k8s_networking": K8sNetworkingTask,
    "pipeline_build_deploy": PipelineBuildDeployTask,
    "pipeline_full_stack": PipelineFullTask,
}


def get_task(task_id: str) -> BaseTask:
    if task_id not in TASK_REGISTRY:
        raise ValueError(f"Unknown task: {task_id}")
    return TASK_REGISTRY[task_id]()