Function workflow_dag

preswald/interfaces/components.py:961–1023

Render the workflow's DAG visualization. Args: workflow: The workflow object to visualize title: Optional title for the visualization

(
        workflow: Workflow,
        title: str = "Workflow Dependency Graph",
        component_id: str | None = None,
        **kwargs
)

@with_render_tracking("workflow_dag")
def workflow_dag(
        workflow: Workflow,
        title: str = "Workflow Dependency Graph",
        component_id: str | None = None,
        **kwargs
) -> ComponentReturn:
    """
    Render the workflow's DAG visualization.

    Args:
        workflow: The workflow object to visualize
        title: Optional title for the visualization
    """
    try:
        from .workflow import WorkflowAnalyzer

        analyzer = WorkflowAnalyzer(workflow)
        analyzer.build_graph()  # Ensure graph is built

        # Get node data
        nodes_data = []
        for node, data in analyzer.graph.nodes(data=True):
            nodes_data.append(
                {
                    "name": node,
                    "status": data["status"],
                    "execution_time": data["execution_time"],
                    "attempts": data["attempts"],
                    "error": data["error"],
                    "dependencies": data["dependencies"],
                    "force_recompute": data["force_recompute"],
                }
            )

        # Create the component with the correct type and data structure
        component = {
            "type": "dag",  # Changed from "plot" to "dag"
            "id": component_id,
            "data": {
                "data": [
                    {
                        "type": "scatter",
                        "customdata": nodes_data,
                        "node": {"positions": []},  # Will be calculated by react-flow
                    }
                ],
                "layout": {"title": {"text": title}, "showlegend": True},
            },
        }

        logger.debug(f"[WORKFLOW_DAG] Created DAG component with id {component_id}")
        return ComponentReturn(component, component)

    except Exception as e:
        logger.error(
            f"[WORKFLOW_DAG] Error creating DAG visualization: {e!s}", exc_info=True
        )
        error_component = {