feat: workflow status

qualibrate_runner/api/routes/__init__.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter
+
+from .get_runnables import get_runnables_router
+from .last_run import last_run_router
+from .others import others_router
+from .submit import submit_router
+
+base_router = APIRouter()
+
+base_router.include_router(submit_router)
+base_router.include_router(get_runnables_router)
+base_router.include_router(last_run_router)
+base_router.include_router(others_router)

qualibrate_runner/api/routes/get_runnables.py

@@ -0,0 +1,83 @@
+from typing import Annotated, Any, Mapping, Sequence, cast
+
+from fastapi import APIRouter, Depends
+from qualibrate.qualibration_graph import QualibrationGraph
+from qualibrate.qualibration_node import QualibrationNode
+
+from qualibrate_runner.api.dependencies import (
+    cache_clear,
+    get_library,
+)
+from qualibrate_runner.api.dependencies import (
+    get_graph as get_qgraph,
+)
+from qualibrate_runner.api.dependencies import (
+    get_graphs as get_qgraphs,
+)
+from qualibrate_runner.api.dependencies import (
+    get_node as get_qnode,
+)
+from qualibrate_runner.api.dependencies import (
+    get_nodes as get_qnodes,
+)
+from qualibrate_runner.config import (
+    QualibrateRunnerSettings,
+    get_settings,
+)
+
+get_runnables_router = APIRouter()
+
+
+@get_runnables_router.get("/get_nodes")
+def get_nodes(
+    nodes: Annotated[Mapping[str, QualibrationNode], Depends(get_qnodes)],
+    settings: Annotated[QualibrateRunnerSettings, Depends(get_settings)],
+    rescan: bool = False,
+) -> Mapping[str, Any]:
+    if rescan:
+        cache_clear()
+        library = get_library(settings)
+        nodes = get_qnodes(library)
+    return {node_name: node.serialize() for node_name, node in nodes.items()}
+
+
+@get_runnables_router.get("/get_graphs")
+def get_graphs(
+    graphs: Annotated[Mapping[str, QualibrationNode], Depends(get_qgraphs)],
+    settings: Annotated[QualibrateRunnerSettings, Depends(get_settings)],
+    rescan: bool = False,
+    cytoscape: bool = False,
+) -> Mapping[str, Any]:
+    if rescan:
+        cache_clear()
+        library = get_library(settings)
+        graphs = get_qgraphs(library)
+    return {
+        graph_name: graph.serialize(cytoscape=cytoscape)
+        for graph_name, graph in graphs.items()
+    }
+
+
+@get_runnables_router.get("/get_node")
+def get_node(
+    node: Annotated[QualibrationNode, Depends(get_qnode)],
+) -> Mapping[str, Any]:
+    return cast(Mapping[str, Any], node.serialize())
+
+
+@get_runnables_router.get("/get_graph")
+def get_graph(
+    graph: Annotated[QualibrationGraph, Depends(get_qgraph)],
+    cytoscape: bool = False,
+) -> Mapping[str, Any]:
+    return cast(Mapping[str, Any], graph.serialize(cytoscape=cytoscape))
+
+
+@get_runnables_router.get("/get_graph/cytoscape")
+def get_graph_cytoscape(
+    graph: Annotated[QualibrationGraph, Depends(get_qgraph)],
+) -> Sequence[Mapping[str, Any]]:
+    return cast(
+        Sequence[Mapping[str, Any]],
+        graph.cytoscape_representation(graph.serialize()),
+    )

qualibrate_runner/api/routes/last_run.py

@@ -0,0 +1,55 @@
+from typing import Annotated, Any, Mapping, Optional, cast
+
+from fastapi import APIRouter, Depends
+from qualibrate.orchestration.execution_history import ExecutionHistory
+from qualibrate.qualibration_graph import QualibrationGraph
+
+from qualibrate_runner.api.dependencies import get_state
+from qualibrate_runner.config import State
+from qualibrate_runner.core.models.last_run import LastRun
+from qualibrate_runner.core.models.workflow import WorkflowStatus
+
+last_run_router = APIRouter(prefix="/last_run")
+
+
+@last_run_router.get("/")
+def get_last_run(
+    state: Annotated[State, Depends(get_state)],
+) -> Optional[LastRun]:
+    return state.last_run
+
+
+@last_run_router.get("/workflow/status")
+def get_workflow_status(
+    state: Annotated[State, Depends(get_state)],
+) -> Optional[WorkflowStatus]:
+    if not isinstance(state.run_item, QualibrationGraph):
+        return None
+    graph: QualibrationGraph = state.run_item
+    run_duration = float(
+        state.last_run.run_duration  # type: ignore
+        if state.last_run
+        else 0.0
+    )
+    return WorkflowStatus(
+        active=state.is_running,
+        nodes_completed=graph.completed_count(),
+        run_duration=run_duration,
+    )
+
+
+@last_run_router.get("/workflow/execution_history")
+def get_execution_history(
+    state: Annotated[State, Depends(get_state)],
+) -> Optional[Mapping[str, Any]]:
+    if not isinstance(state.run_item, QualibrationGraph):
+        return None
+    graph: QualibrationGraph = state.run_item
+    orch = graph._orchestrator
+    if orch is None:
+        raise RuntimeError("No graph orchestrator")
+    history: ExecutionHistory = orch.get_execution_history()
+    return cast(
+        Mapping[str, Any],
+        history.model_dump(mode="json", serialize_as_any=True),
+    )