dispatch_node

@dataclass
class AbstractDispatchTask(abc.ABC):
    """A graph execution split node, not to be executed."""

    name: str
    label: Optional[str] = None
    class_name: Optional[str] = None
    node_names: List[str] = field(default_factory=list)
    task_chunk: Optional[TaskChunkParameters] = None

    ui: Optional[ui_utils.UiNodeMetadata] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    @property
    def node_type(self) -> str:
        return self.__class__.__name__

    def __str__(self) -> str:
        return f"<{self.node_type} {self.label}>"

    @staticmethod
    def get_associated_manager() -> str:
        """The manager associated to the dispatch task."""
        raise NotImplementedError("abstract method")

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "AbstractDispatchTask":
        if data.get("task_chunk") and not isinstance(
            data["task_chunk"], TaskChunkParameters
        ):
            data["task_chunk"] = TaskChunkParameters(**data["task_chunk"])

        # Backward compatibility, dispatch task used to
        # contain all nodes are entire object instead of just names.
        if data.get("nodes") and not data.get("node_names"):
            data["node_names"] = [
                node.get("name")
                for node in data.pop("nodes")
            ]

        return cls(
            name=data.pop("name"),
            label=data.pop("label", None),
            **data,
        )

@dataclass
class DeadlineThinkbox(AbstractDispatchTask):
    """A split node associated to Deadline Thinkbox."""

    # TODO: this should be moved to ayon-deadline at some point
    job_name: Optional[str] = None
    priority: Optional[int] = None
    pool: Optional[str] = None
    group: Optional[str] = None
    limit_groups: Optional[List[str]] = field(default_factory=list)
    override_task_failure: Optional[bool] = False
    task_failure_detection: Optional[int] = 0  # no detection
    username: Optional[str] = None
    comment: Optional[str] = None

    @staticmethod
    def get_associated_manager() -> str:
        return "Deadline Thinkbox"

class GenericDispatchTask(AbstractDispatchTask):
    """A generic split node."""

    @staticmethod
    def get_associated_manager() -> str:
        return "AYON"

@dataclass
class RoyalRender(AbstractDispatchTask):
    """A split node associated to RoyalRender."""

    # TODO: this should be moved to ayon-royalrender at some point
    job_name: Optional[str] = None
    priority: Optional[int] = None
    username: Optional[str] = None
    required_license: Optional[str] = None
    required_plugin: Optional[str] = None

    @staticmethod
    def get_associated_manager() -> str:
        return "RoyalRender"

@dataclass
class TaskChunkParameters:
    """A graph execution split node, not to be executed."""
    node_name: str
    node_input_name: str
    chunk_size: int = 1