Data and element helper API

On this page

• Record conversion
• Element helpers
• Validation
• Element types

Record conversion

st_graph_workbench.component.records.dataframe_to_records

dataframe_to_records(dataframe: Any, *, drop_missing: bool = False) -> list[Record]

Convert a dataframe-like object to JSON-friendly record dictionaries.

Supports pandas dataframes, Polars dataframes, and Arrow-like tables that expose one of the common record-export methods. Scalar missing values become None so records can be passed safely into graph_workbench(...). Nested tuples, sets, array-like values, datetimes, and decimals are normalized into deterministic JSON-friendly values.

Source code in st_graph_workbench/component/records.py

def dataframe_to_records(dataframe: Any, *, drop_missing: bool = False) -> list[Record]:
    """Convert a dataframe-like object to JSON-friendly record dictionaries.

    Supports pandas dataframes, Polars dataframes, and Arrow-like tables that
    expose one of the common record-export methods. Scalar missing values become
    ``None`` so records can be passed safely into ``graph_workbench(...)``.
    Nested tuples, sets, array-like values, datetimes, and decimals are
    normalized into deterministic JSON-friendly values.
    """

    if hasattr(dataframe, "to_dicts"):
        raw_records = dataframe.to_dicts()
    elif hasattr(dataframe, "to_pylist"):
        raw_records = dataframe.to_pylist()
    elif hasattr(dataframe, "to_dict"):
        try:
            raw_records = dataframe.to_dict(orient="records")
        except TypeError:
            raw_records = dataframe.to_dict("records")
    else:
        raise TypeError(
            "`dataframe` must provide `to_dicts()`, `to_pylist()`, or "
            "`to_dict(orient='records')`."
        )

    if not _is_sequence(raw_records):
        raise TypeError("Dataframe record export must return a sequence of records.")

    pandas = _optional_pandas()
    records: list[Record] = []
    for index, record in enumerate(raw_records):
        if not isinstance(record, Mapping):
            raise TypeError(f"Record {index} is not a dictionary-like object.")
        records.append(_clean_record(record, drop_missing=drop_missing, pandas=pandas))
    return records

st_graph_workbench.component.records.records_to_dataframe

records_to_dataframe(records: Mapping[str, Any] | Sequence[Mapping[str, Any]], *, group_key: str = 'record_group', flatten_data: bool = True) -> Any

Convert record dictionaries to a pandas dataframe.

records may be a normal list of dictionaries, a dictionary of named record groups such as {"sightings": [...]}, a graph element dictionary such as {"nodes": [...], "edges": [...]}, or the element records in a selection/search event. Graph and event element data fields are flattened into columns by default so returned records are immediately useful in Streamlit and notebooks.

Source code in st_graph_workbench/component/records.py

def records_to_dataframe(
    records: Mapping[str, Any] | Sequence[Mapping[str, Any]],
    *,
    group_key: str = "record_group",
    flatten_data: bool = True,
) -> Any:
    """Convert record dictionaries to a pandas dataframe.

    ``records`` may be a normal list of dictionaries, a dictionary of named
    record groups such as ``{"sightings": [...]}``, a graph element dictionary
    such as ``{"nodes": [...], "edges": [...]}``, or the element records in a
    selection/search event. Graph and event element ``data`` fields are
    flattened into columns by default so returned records are immediately
    useful in Streamlit and notebooks.
    """

    pandas = _require_pandas()

    if isinstance(records, Mapping):
        if "nodes" in records or "edges" in records:
            rows = _records_from_graph_elements(
                records,
                group_key=group_key,
                flatten_data=flatten_data,
            )
        else:
            rows = _records_from_grouped_mapping(records, group_key=group_key)
    elif _is_sequence(records):
        rows = _records_from_sequence(
            records,
            group_key=group_key,
            flatten_data=flatten_data,
        )
    else:
        raise TypeError(
            "`records` must be a sequence of dictionaries or a dictionary of "
            "record groups."
        )

    return pandas.DataFrame(rows)

Element helpers

st_graph_workbench.component.elements.get_element

get_element(elements: Elements, element_id: ElementId) -> Element | None

Return a copy of a node or edge by ID, or None when it is not present.

Source code in st_graph_workbench/component/elements.py

def get_element(elements: Elements, element_id: ElementId) -> Element | None:
    """Return a copy of a node or edge by ID, or None when it is not present."""
    normalized_id = str(element_id)
    graph = _copy_elements(elements)
    for element in [*graph["nodes"], *graph["edges"]]:
        if _element_id(element) == normalized_id:
            return cast(Element, copy.deepcopy(element))
    return None

st_graph_workbench.component.elements.upsert_elements

upsert_elements(elements: Elements, *, nodes: ElementInput = None, edges: ElementInput = None, replace: bool = True) -> Elements

Return a graph with nodes and edges added or updated by data.id.

Source code in st_graph_workbench/component/elements.py

def upsert_elements(
    elements: Elements,
    *,
    nodes: ElementInput = None,
    edges: ElementInput = None,
    replace: bool = True,
) -> Elements:
    """Return a graph with nodes and edges added or updated by data.id."""
    graph = _copy_elements(elements)
    graph["nodes"] = _upsert_group(
        graph["nodes"], _as_element_list(nodes), group="nodes", replace=replace
    )
    graph["edges"] = _upsert_group(
        graph["edges"], _as_element_list(edges), group="edges", replace=replace
    )
    validate_elements(graph)
    return graph

st_graph_workbench.component.elements.update_element_data

update_element_data(elements: Elements, element_id: ElementId, data: dict[str, Any], *, merge: bool = True) -> Elements

Return a graph with one node or edge data dict updated.

Source code in st_graph_workbench/component/elements.py

def update_element_data(
    elements: Elements,
    element_id: ElementId,
    data: dict[str, Any],
    *,
    merge: bool = True,
) -> Elements:
    """Return a graph with one node or edge data dict updated."""
    if not isinstance(data, dict):
        raise TypeError("`data` must be a dict.")

    normalized_id = str(element_id)
    _validate_update_identity(normalized_id, data, merge=merge)
    graph = _copy_elements(elements)

    for group in ("nodes", "edges"):
        for element in graph[group]:
            if _element_id(element) != normalized_id:
                continue
            element["data"] = (
                {**copy.deepcopy(element.get("data", {})), **copy.deepcopy(data)}
                if merge
                else copy.deepcopy(data)
            )
            validate_elements(graph)
            return graph

    raise KeyError(f"Element `{normalized_id}` was not found.")

st_graph_workbench.component.elements.delete_elements

delete_elements(elements: Elements, *, node_ids: ElementIdInput = None, edge_ids: ElementIdInput = None, remove_incident_edges: bool = True) -> Elements

Return a graph with selected nodes, edges, and optional incident edges removed.

Source code in st_graph_workbench/component/elements.py

def delete_elements(
    elements: Elements,
    *,
    node_ids: ElementIdInput = None,
    edge_ids: ElementIdInput = None,
    remove_incident_edges: bool = True,
) -> Elements:
    """Return a graph with selected nodes, edges, and optional incident edges removed."""
    nodes_to_remove = _normalized_ids(node_ids)
    edges_to_remove = _normalized_ids(edge_ids)
    graph = _copy_elements(elements)

    graph["nodes"] = [
        node for node in graph["nodes"] if _element_id(node) not in nodes_to_remove
    ]

    def keep_edge(edge: Element) -> bool:
        data = edge["data"]
        if _element_id(edge) in edges_to_remove:
            return False
        if remove_incident_edges and (
            str(data.get("source")) in nodes_to_remove
            or str(data.get("target")) in nodes_to_remove
        ):
            return False
        return True

    graph["edges"] = [edge for edge in graph["edges"] if keep_edge(edge)]
    validate_elements(graph)
    return graph

Validation

st_graph_workbench.component.validation.validate_elements

validate_elements(elements: Any, *, strict: bool = True) -> None

Validate Streamlit graph elements before they are sent to Cytoscape.

Cytoscape treats IDs as strings and requires every edge source/target to reference an existing node. This helper catches common data-shape mistakes on the Python side so Streamlit apps get actionable errors instead of a browser-side rendering failure.

Source code in st_graph_workbench/component/validation.py

def validate_elements(elements: Any, *, strict: bool = True) -> None:
    """
    Validate Streamlit graph elements before they are sent to Cytoscape.

    Cytoscape treats IDs as strings and requires every edge source/target to
    reference an existing node. This helper catches common data-shape mistakes
    on the Python side so Streamlit apps get actionable errors instead of a
    browser-side rendering failure.
    """
    if not isinstance(elements, dict):
        _fail("`elements` must be a dict with optional `nodes` and `edges` lists.")

    nodes = _as_list(elements.get("nodes", []), "nodes")
    edges = _as_list(elements.get("edges", []), "edges")

    node_ids: set[str] = set()
    all_ids: list[str] = []
    parent_by_node: dict[str, str] = {}

    for index, node in enumerate(nodes):
        data = _get_data(node, "nodes", index)
        node_id = _normalize_id(data.get("id"), f"nodes[{index}].data.id")
        node_ids.add(node_id)
        all_ids.append(node_id)
        _validate_expansion(data, node_id)

        parent = data.get("parent")
        if parent is not None and parent != "":
            parent_id = _normalize_id(parent, f"node `{node_id}` data.parent")
            if parent_id == node_id:
                _fail(f"node `{node_id}` cannot be its own compound parent.")
            parent_by_node[node_id] = parent_id

    edge_refs: list[tuple[str, str, str]] = []
    for index, edge in enumerate(edges):
        data = _get_data(edge, "edges", index)
        edge_id = _normalize_id(data.get("id"), f"edges[{index}].data.id")
        all_ids.append(edge_id)

        source = _normalize_id(data.get("source"), f"edge `{edge_id}` source")
        target = _normalize_id(data.get("target"), f"edge `{edge_id}` target")
        edge_refs.append((edge_id, source, target))

    _validate_unique_ids(all_ids)

    if not strict:
        return

    for edge_id, source, target in edge_refs:
        if source not in node_ids:
            _fail(f"edge `{edge_id}` source `{source}` does not match any node id.")
        if target not in node_ids:
            _fail(f"edge `{edge_id}` target `{target}` does not match any node id.")

    for node_id, parent_id in parent_by_node.items():
        if parent_id not in node_ids:
            _fail(
                f"node `{node_id}` compound parent `{parent_id}` "
                "does not match any node id."
            )

    _validate_parent_cycles(parent_by_node)

st_graph_workbench.component.validation.ElementValidationError

Bases: ValueError

Raised when graph elements cannot be safely rendered by Cytoscape.

Element types

st_graph_workbench.component.types.Element module-attribute

Element: TypeAlias = dict[str, Any]

One Cytoscape node or edge dictionary.

st_graph_workbench.component.types.ElementId module-attribute

ElementId: TypeAlias = str | int | float | bool

One node, edge, command, source, target, or parent identifier.

st_graph_workbench.component.types.ElementIdInput module-attribute

ElementIdInput: TypeAlias = ElementId | Iterable[ElementId] | None

One identifier, an iterable of identifiers, or no identifiers.

st_graph_workbench.component.types.Elements module-attribute

Elements: TypeAlias = dict[str, Any]

A graph dictionary containing nodes and edges lists.

st_graph_workbench.component.types.Record module-attribute

Record: TypeAlias = dict[str, Any]

One table-like or JSON-ready record dictionary.

st_graph_workbench.component.types.Layout module-attribute

Layout: TypeAlias = str | dict[str, Any]

A named layout or a Cytoscape layout options dictionary.