47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from ..graph_export import edge_retrieval_query, graph_from_sparql_bindings
|
|
from ..models import GraphResponse
|
|
from ..sparql_engine import SparqlEngine
|
|
from ..settings import Settings
|
|
from .layout_spiral import spiral_positions
|
|
|
|
|
|
async def fetch_graph_snapshot(
|
|
sparql: SparqlEngine,
|
|
*,
|
|
settings: Settings,
|
|
node_limit: int,
|
|
edge_limit: int,
|
|
) -> GraphResponse:
|
|
"""
|
|
Fetch a graph snapshot (nodes + edges) via SPARQL, independent of whether the
|
|
underlying engine is RDFLib or AnzoGraph.
|
|
"""
|
|
edges_q = edge_retrieval_query(edge_limit=edge_limit, include_bnodes=settings.include_bnodes)
|
|
res = await sparql.query_json(edges_q)
|
|
bindings = (((res.get("results") or {}).get("bindings")) or [])
|
|
nodes, edges = graph_from_sparql_bindings(
|
|
bindings,
|
|
node_limit=node_limit,
|
|
include_bnodes=settings.include_bnodes,
|
|
)
|
|
|
|
# Add positions so the frontend doesn't need to run a layout.
|
|
xs, ys = spiral_positions(len(nodes))
|
|
for i, node in enumerate(nodes):
|
|
node["x"] = float(xs[i])
|
|
node["y"] = float(ys[i])
|
|
|
|
meta = GraphResponse.Meta(
|
|
backend=sparql.name,
|
|
ttl_path=settings.ttl_path if settings.graph_backend == "rdflib" else None,
|
|
sparql_endpoint=settings.effective_sparql_endpoint() if settings.graph_backend == "anzograph" else None,
|
|
include_bnodes=settings.include_bnodes,
|
|
node_limit=node_limit,
|
|
edge_limit=edge_limit,
|
|
nodes=len(nodes),
|
|
edges=len(edges),
|
|
)
|
|
return GraphResponse(nodes=nodes, edges=edges, meta=meta)
|