Graph command API

On this page

• Command builders
• Validation and application
• Command types and errors

Command builders

st_graph_workbench.component.commands.add_elements_command

add_elements_command(command_id: ElementId, *, nodes: ElementInput = None, edges: ElementInput = None) -> GraphCommand

Build a command that adds only missing nodes and edges in the browser.

Source code in st_graph_workbench/component/commands.py

def add_elements_command(
    command_id: ElementId,
    *,
    nodes: ElementInput = None,
    edges: ElementInput = None,
) -> GraphCommand:
    """Build a command that adds only missing nodes and edges in the browser."""
    command = _command_base(command_id, "add_elements")
    node_list = _as_element_list(nodes)
    edge_list = _as_element_list(edges)
    if node_list:
        command["nodes"] = node_list
    if edge_list:
        command["edges"] = edge_list
    validate_graph_commands([command])
    return command

st_graph_workbench.component.commands.upsert_elements_command

upsert_elements_command(command_id: ElementId, *, nodes: ElementInput = None, edges: ElementInput = None, replace: bool = True) -> GraphCommand

Build a command that adds or updates nodes and edges by data.id.

Source code in st_graph_workbench/component/commands.py

def upsert_elements_command(
    command_id: ElementId,
    *,
    nodes: ElementInput = None,
    edges: ElementInput = None,
    replace: bool = True,
) -> GraphCommand:
    """Build a command that adds or updates nodes and edges by `data.id`."""
    command = _command_base(command_id, "upsert_elements")
    node_list = _as_element_list(nodes)
    edge_list = _as_element_list(edges)
    if node_list:
        command["nodes"] = node_list
    if edge_list:
        command["edges"] = edge_list
    command["replace"] = replace
    validate_graph_commands([command])
    return command

st_graph_workbench.component.commands.update_data_command

update_data_command(command_id: ElementId, element_id: ElementId, data: Dict[str, Any], *, merge: bool = True) -> GraphCommand

Build a command that updates one existing node or edge data dict.

Source code in st_graph_workbench/component/commands.py

def update_data_command(
    command_id: ElementId,
    element_id: ElementId,
    data: Dict[str, Any],
    *,
    merge: bool = True,
) -> GraphCommand:
    """Build a command that updates one existing node or edge data dict."""
    command = _command_base(command_id, "update_data")
    command["element_id"] = str(element_id)
    command["data"] = copy.deepcopy(data)
    command["merge"] = merge
    validate_graph_commands([command])
    return command

st_graph_workbench.component.commands.delete_elements_command

delete_elements_command(command_id: ElementId, *, node_ids: ElementIdInput = None, edge_ids: ElementIdInput = None, remove_incident_edges: bool = True) -> GraphCommand

Build a command that deletes nodes, edges, and optional incident edges.

Source code in st_graph_workbench/component/commands.py

def delete_elements_command(
    command_id: ElementId,
    *,
    node_ids: ElementIdInput = None,
    edge_ids: ElementIdInput = None,
    remove_incident_edges: bool = True,
) -> GraphCommand:
    """Build a command that deletes nodes, edges, and optional incident edges."""
    command = _command_base(command_id, "delete_elements")
    command["node_ids"] = _as_id_list(node_ids)
    command["edge_ids"] = _as_id_list(edge_ids)
    command["remove_incident_edges"] = remove_incident_edges
    validate_graph_commands([command])
    return command

st_graph_workbench.component.commands.set_elements_command

set_elements_command(command_id: ElementId, elements: Elements) -> GraphCommand

Build a command that replaces the browser graph with full elements.

Source code in st_graph_workbench/component/commands.py

def set_elements_command(command_id: ElementId, elements: Elements) -> GraphCommand:
    """Build a command that replaces the browser graph with full elements."""
    graph = _copy_elements(elements)
    command = _command_base(command_id, "set_elements")
    command["elements"] = graph
    validate_graph_commands([command])
    return command

st_graph_workbench.component.commands.clear_graph_command

clear_graph_command(command_id: ElementId) -> GraphCommand

Build a command that clears every node and edge from the browser graph.

Source code in st_graph_workbench/component/commands.py

def clear_graph_command(command_id: ElementId) -> GraphCommand:
    """Build a command that clears every node and edge from the browser graph."""
    command = _command_base(command_id, "clear")
    validate_graph_commands([command])
    return command

st_graph_workbench.component.commands.viewport_command

viewport_command(command_id: ElementId, operation: ViewportCommandOperation, **options: Any) -> GraphCommand

Build a non-mutating viewport/layout command.

Source code in st_graph_workbench/component/commands.py

def viewport_command(
    command_id: ElementId,
    operation: ViewportCommandOperation,
    **options: Any,
) -> GraphCommand:
    """Build a non-mutating viewport/layout command."""
    command = _command_base(command_id, operation)
    command.update(copy.deepcopy(options))
    validate_graph_commands([command])
    return command

Validation and application

st_graph_workbench.component.commands.validate_graph_commands

validate_graph_commands(commands: Optional[Iterable[GraphCommand]], *, elements: Optional[Elements] = None) -> None

Validate incremental graph commands before they are sent to the browser.

Source code in st_graph_workbench/component/commands.py

def validate_graph_commands(
    commands: Optional[Iterable[GraphCommand]],
    *,
    elements: Optional[Elements] = None,
) -> None:
    """Validate incremental graph commands before they are sent to the browser."""
    if commands is None:
        return
    if isinstance(commands, (str, bytes, dict)):
        _fail("`graph_commands` must be a list or iterable of command dicts.")

    seen_ids: Set[str] = set()
    known_node_ids = _known_node_ids(elements) if elements is not None else None
    known_edge_endpoints = (
        _known_edge_endpoints(elements) if elements is not None else None
    )
    for command in commands:
        operation = _validate_command(
            command,
            known_node_ids=known_node_ids,
            known_edge_endpoints=known_edge_endpoints,
        )
        command_id = _normalize_command_id(command.get("command_id"))
        if command_id in seen_ids:
            _fail(f"duplicate command_id `{command_id}`.")
        seen_ids.add(command_id)
        known_node_ids = _update_known_node_ids_for_command(
            known_node_ids,
            command,
            operation,
        )
        known_edge_endpoints = _update_known_edge_endpoints_for_command(
            known_edge_endpoints,
            command,
            operation,
        )

st_graph_workbench.component.commands.apply_graph_command

apply_graph_command(elements: Elements, command: GraphCommand) -> Elements

Return a new graph after applying one command to Python-owned elements.

This helper is useful in Streamlit apps that keep st.session_state authoritative while also sending the same command to the browser for an incremental Cytoscape update.

Source code in st_graph_workbench/component/commands.py

def apply_graph_command(elements: Elements, command: GraphCommand) -> Elements:
    """
    Return a new graph after applying one command to Python-owned elements.

    This helper is useful in Streamlit apps that keep `st.session_state`
    authoritative while also sending the same command to the browser for an
    incremental Cytoscape update.
    """
    validate_graph_commands([command], elements=elements)
    operation = _normalize_operation(command.get("operation"))

    if operation == "add_elements":
        for element in [*_command_nodes(command), *_command_edges(command)]:
            element_id = element["data"]["id"]
            if get_element(elements, element_id) is not None:
                raise KeyError(f"Element `{element_id}` already exists.")
        return upsert_elements(
            elements,
            nodes=_command_nodes(command),
            edges=_command_edges(command),
            replace=True,
        )

    if operation == "upsert_elements":
        return upsert_elements(
            elements,
            nodes=_command_nodes(command),
            edges=_command_edges(command),
            replace=bool(command.get("replace", True)),
        )

    if operation == "update_data":
        return update_element_data(
            elements,
            command["element_id"],
            cast(Dict[str, Any], command["data"]),
            merge=bool(command.get("merge", True)),
        )

    if operation == "delete_elements":
        return delete_elements(
            elements,
            node_ids=cast(ElementIdInput, command.get("node_ids")),
            edge_ids=cast(ElementIdInput, command.get("edge_ids")),
            remove_incident_edges=bool(command.get("remove_incident_edges", True)),
        )

    if operation == "clear":
        return {"nodes": [], "edges": []}

    if operation == "set_elements":
        return _command_elements(command)

    return _copy_elements(elements)

st_graph_workbench.component.commands.apply_graph_commands

apply_graph_commands(elements: Elements, commands: Optional[Iterable[GraphCommand]]) -> Elements

Return a new graph after applying commands in order.

Source code in st_graph_workbench/component/commands.py

def apply_graph_commands(
    elements: Elements,
    commands: Optional[Iterable[GraphCommand]],
) -> Elements:
    """Return a new graph after applying commands in order."""
    graph = _copy_elements(elements)
    if commands is None:
        return graph
    for command in commands:
        graph = apply_graph_command(graph, command)
    return graph

Command types and errors

st_graph_workbench.component.types.GraphCommand module-attribute

GraphCommand: TypeAlias = dict[str, Any]

One incremental command sent to the browser graph.

st_graph_workbench.component.commands.GraphCommandValidationError

Bases: ValueError

Raised when incremental graph commands cannot be applied safely.