workflow

@dataclass
class Workflow(AbstractSerializable):
    """A container of graphs"""

    execution_graph: Graph = field(default_factory=Graph)
    dispatch_graphs: List[DispatchGraph] = field(default_factory=list)

    metadata: Dict[str, Any] = field(default_factory=dict)

    @property
    def dispatch_graph_names(self) -> Tuple[str]:
        return tuple(graph.name for graph in self.dispatch_graphs)

    def create_dispatch_graph(self, name: str) -> DispatchGraph:
        dispatch_graph = DispatchGraph(name=name)
        self.dispatch_graphs.append(dispatch_graph)
        return dispatch_graph

    def validate_dispatch_graph(self, dispatch_graph: DispatchGraph):
        if dispatch_graph not in self.dispatch_graphs:
            raise ValueError(
                f"Invalid dispatch graph {dispatch_graph.name} "
                "not part of current workflow."
            )

        unordered_slices_nodes = {
            dispatch_node.name: dispatch_node.node_names
            for dispatch_node in dispatch_graph.all_nodes
        }

        if not unordered_slices_nodes:
            raise ValueError(
                f"Invalid empty dispatch graph {dispatch_graph}."
            )

        duplicates = set()
        nodes_within_slices = set()
        for node_names in unordered_slices_nodes.values():
            node_names = set(node_names)
            duplicates |= nodes_within_slices & node_names
            nodes_within_slices |= node_names

        if duplicates:
            raise ValueError(
                f"Invalid dispatch graph {dispatch_graph}. "
                f"Execution node(s) {duplicates} are associated "
                "to multiple dispatch tasks."
            )

        # Detect any execution node not part of any slice.
        all_execution_node_names = {
            node.name for node in self.execution_graph.all_nodes
        }
        ambiguous_nodes = all_execution_node_names - nodes_within_slices
        if ambiguous_nodes:
            raise ValueError(
                f"Invalid dispatch graph {dispatch_graph}. "
                f"No dispatch task associated with {ambiguous_nodes}."
            )

        # For each dispatch task, validate chunk parameters if defined.
        for dispatch_node in dispatch_graph.all_nodes:
            task_chunk = dispatch_node.task_chunk
            if task_chunk is None:  # no chunking
                continue

            if task_chunk.chunk_size < 1:
                raise ValueError(
                    f"Invalid chunk size: {task_chunk.chunk_size}."
                )
            if task_chunk.node_name not in dispatch_node.node_names:
                raise ValueError(
                    f"Invalid chunk node: {task_chunk.node_name}: "
                    "chunk node is not part of the dispatch task."
                )
            if task_chunk.node_name not in all_execution_node_names:
                raise ValueError(
                    f"Invalid chunk node: {task_chunk.node_name}: "
                    "chunk node is not part of the execution graph."
                )

            chunk_node = self.execution_graph.get_node_by_name(
                task_chunk.node_name
            )
            if task_chunk.node_input_name not in chunk_node.inputs:
                raise ValueError(
                    f"Invalid chunk input: {task_chunk.node_input_name}. "
                    f"Chunk node {chunk_node} does not define this input."
                )


    def validate(self):
        """Ensure the workflow is valid, raise if it is not."""
        # Validate all dispatch graphs.
        for dispatch_graph in self.dispatch_graphs:
            self.validate_dispatch_graph(dispatch_graph)

    def to_json(self, sorted_keys: Optional[bool] = False) -> str:
        self.validate()
        return super().to_json(sorted_keys=sorted_keys)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Workflow":
        data = serialization.deserialize_from_primitive(data)

        # Backward compatible with previous serialization method.
        # TODO remove legacy serialization format support in a future version.
        if isinstance(data, dict):
            logger.warning(
                "DispatchGraph.from_dict: using legacy serialization format"
            )
            dispatch_graphs = []
            for graph_data in data.get("dispatch_graphs", []):
                try:
                    dispatch_graphs.append(DispatchGraph.from_dict(graph_data))
                except TypeError as error:
                    raise ValueError(
                        f"Unsupported Graph data: {graph_data}."
                    ) from error

                # backward compatibility: empty default_task_cls and class_name
                this_dispatch_graph = dispatch_graphs[-1]
                if not this_dispatch_graph.default_task_cls:
                    graph_task_types = set()
                    # check all task nodes for a valid class_name
                    for node in this_dispatch_graph.all_nodes:
                        if not node.class_name:
                            # infer task type based on name, assuming we have a
                            # ClassName[0-9]+
                            task_class = re.sub(r"\d+$", "", node.name)
                            node.class_name = task_class
                            graph_task_types.add(task_class)
                        else:
                            graph_task_types.add(node.class_name)

                    if len(graph_task_types) == 1:
                        this_dispatch_graph.default_task_cls = (
                            graph_task_types.pop()
                        )
                    else:
                        raise ValueError(
                            f"Invalid dispatch graph {this_dispatch_graph}. "
                            f"Multiple task types found: {graph_task_types}."
                        )

            return cls(
                name=data.get("name"),
                description=data.get("description"),
                # default to data to allow reading pre-execution_graph files
                execution_graph=Graph.from_dict(
                    data.get("execution_graph", data)
                ),
                dispatch_graphs=dispatch_graphs,
                metadata=data.get("metadata"),
            )

        return data