ontocast.tool.sparql

SPARQL tool for incremental graph updates.

This module provides functionality for executing SPARQL operations on RDF graphs, enabling incremental updates instead of full graph replacement.

SPARQLTool

Tool for executing SPARQL operations on RDF graphs.

Source code in ontocast/tool/sparql.py

class SPARQLTool:
    """Tool for executing SPARQL operations on RDF graphs."""

    def __init__(self, triple_store_manager: TripleStoreManager | None = None):
        """Initialize SPARQL tool.

        Args:
            triple_store_manager: Optional triple store manager for persistent storage.
        """
        self.triple_store_manager = triple_store_manager
        self.operation_history = []
        self.last_finalize_metrics: dict[str, int] = {}

    def execute_operations(
        self, graph: RDFGraph, operations: list[SPARQLOperationModel]
    ) -> RDFGraph:
        """Execute a list of SPARQL operations on a graph.

        Args:
            graph: The RDF graph to operate on.
            operations: List of SPARQL operations to execute.

        Returns:
            RDFGraph: Updated graph after applying operations.
        """
        logger.info(f"Executing {len(operations)} SPARQL operations")

        for operation in operations:
            try:
                self._execute_single_operation(graph, operation)
                self.operation_history.append(operation)
                logger.debug(
                    f"Executed {operation.operation_type} operation: {operation.description}"
                )
            except Exception as e:
                logger.error(
                    f"Failed to execute {operation.operation_type} operation: {str(e)}"
                )
                raise

        return graph

    def execute_operation(self, operation: SPARQLOperationModel) -> None:
        """Execute a single SPARQL operation.

        Args:
            operation: The SPARQL operation to execute.
        """
        # For now, we'll use a simple approach - in a real implementation,
        # you might want to track which graph this operation should be applied to
        logger.info(
            f"Executing {operation.operation_type} operation: {operation.description}"
        )
        # This is a placeholder - in practice, you'd need to specify which graph to operate on
        # or maintain a default graph in the tool

    def _execute_single_operation(
        self, graph: RDFGraph, operation: SPARQLOperationModel
    ):
        """Execute a single SPARQL operation.

        Args:
            graph: The RDF graph to operate on.
            operation: The SPARQL operation to execute.
        """
        if operation.operation_type == SPARQLOperationType.INSERT:
            self._execute_insert(graph, operation)
        elif operation.operation_type == SPARQLOperationType.DELETE:
            self._execute_delete(graph, operation)
        elif operation.operation_type == SPARQLOperationType.UPDATE:
            self._execute_update(graph, operation)
        else:
            raise ValueError(f"Unknown operation type: {operation.operation_type}")

    def _execute_insert(self, graph: RDFGraph, operation: SPARQLOperationModel):
        """Execute INSERT operation.

        Args:
            graph: The RDF graph to operate on.
            operation: The INSERT operation to execute.
        """
        # Parse the INSERT query
        query = prepareQuery(operation.query)

        # For INSERT DATA, we need to parse the triples and add them to the graph
        if "INSERT DATA" in operation.query.upper():
            # Extract triples from INSERT DATA query
            triples = self._parse_insert_data_triples(operation.query)
            for triple in triples:
                graph.add(triple)
        else:
            # For other INSERT queries, execute against the graph
            graph.query(query)
            # INSERT queries typically don't return results, but we execute them

    def _execute_delete(self, graph: RDFGraph, operation: SPARQLOperationModel):
        """Execute DELETE operation.

        Args:
            graph: The RDF graph to operate on.
            operation: The DELETE operation to execute.
        """
        # Parse the DELETE query
        query = prepareQuery(operation.query)

        # For DELETE DATA, we need to parse the triples and remove them from the graph
        if "DELETE DATA" in operation.query.upper():
            # Extract triples from DELETE DATA query
            triples = self._parse_delete_data_triples(operation.query)
            for triple in triples:
                graph.remove(triple)
        else:
            # For other DELETE queries, execute against the graph
            graph.query(query)
            # DELETE queries typically don't return results, but we execute them

    def _execute_update(self, graph: RDFGraph, operation: SPARQLOperationModel):
        """Execute UPDATE operation.

        Args:
            graph: The RDF graph to operate on.
            operation: The UPDATE operation to execute.
        """
        # Parse the UPDATE query
        query = prepareQuery(operation.query)

        # Execute the UPDATE query
        graph.query(query)
        # UPDATE queries typically don't return results, but we execute them

    def _parse_insert_data_triples(self, query: str) -> list[tuple]:
        """Parse triples from INSERT DATA query.

        Args:
            query: The INSERT DATA query string.

        Returns:
            List of triples to insert.
        """
        # This is a simplified parser - in practice, you'd want a more robust parser
        triples = []

        # Extract the content between INSERT DATA { ... }
        start = query.upper().find("INSERT DATA {")
        if start == -1:
            return triples

        start += len("INSERT DATA {")
        end = query.rfind("}")

        if end == -1:
            return triples

        data_content = query[start:end].strip()

        # Split by lines and parse each triple
        lines = [line.strip() for line in data_content.split("\n") if line.strip()]

        for line in lines:
            if line.endswith("."):
                line = line[:-1]  # Remove trailing period

            # Parse the triple (simplified - assumes standard N3 format)
            parts = line.split()
            if len(parts) >= 3:
                subject = self._parse_term(parts[0])
                predicate = self._parse_term(parts[1])
                object_part = self._parse_term(" ".join(parts[2:]))

                if subject and predicate and object_part:
                    triples.append((subject, predicate, object_part))

        return triples

    def _parse_delete_data_triples(self, query: str) -> list[tuple]:
        """Parse triples from DELETE DATA query.

        Args:
            query: The DELETE DATA query string.

        Returns:
            List of triples to delete.
        """
        # Similar to INSERT DATA parsing
        return self._parse_insert_data_triples(
            query.replace("DELETE DATA", "INSERT DATA")
        )

    def _parse_term(self, term: str):
        """Parse a SPARQL term (subject, predicate, or object).

        Args:
            term: The term string to parse.

        Returns:
            Parsed RDF term (URIRef, Literal, or BNode).
        """
        term = term.strip()

        if term.startswith("<") and term.endswith(">"):
            # URI
            return URIRef(term[1:-1])
        elif term.startswith('"') and term.endswith('"'):
            # Literal
            return Literal(term[1:-1])
        elif term.startswith("_:"):
            # Blank node
            return BNode(term[2:])
        elif term.startswith('"') and '"^^' in term:
            # Typed literal
            value, datatype = term.split('"^^')
            return Literal(value[1:], datatype=URIRef(datatype))
        else:
            # Assume it's a URI without angle brackets
            return URIRef(term)

    def validate_operation(self, operation: SPARQLOperationModel) -> bool:
        """Validate a SPARQL operation.

        Args:
            operation: The operation to validate.

        Returns:
            bool: True if valid, False otherwise.
        """
        try:
            prepareQuery(operation.query)
            return True
        except Exception as e:
            logger.error(f"Invalid SPARQL operation: {str(e)}")
            return False

    def get_operation_history(self) -> list[SPARQLOperationModel]:
        """Get the history of executed operations.

        Returns:
            List of executed operations.
        """
        return self.operation_history.copy()

    def clear_history(self):
        """Clear the operation history."""
        self.operation_history.clear()

    @staticmethod
    def _build_induced_subgraph(
        ontologies: list[Ontology],
        entity_uris: list[str],
        entity_relevance: dict[str, float] | None,
        ontology_iris: list[str] | None,
        depth: int,
        max_total_triples: int,
        estimated_triples_per_query: int,
        ontology_version_filters: dict[str, set[str]] | None,
        ontology_hash_filters: dict[str, set[str]] | None,
        entity_roles: Mapping[str, str | None] | None = None,
        hub_seed_count: int = 8,
        ancestor_closure_depth: int = 3,
    ) -> tuple[RDFGraph, dict[str, int]]:
        """Merge filtered graphs; schema shell, hub BFS, and connectivity repair."""

        ontology_filter = set(ontology_iris or [])
        relevant_graphs: list[RDFGraph] = []
        for ontology in ontologies:
            if ontology_filter and ontology.iri not in ontology_filter:
                continue
            if ontology_version_filters and ontology.iri in ontology_version_filters:
                ontology_version = (
                    str(ontology.version) if ontology.version is not None else None
                )
                if ontology_version not in ontology_version_filters[ontology.iri]:
                    continue
            if ontology_hash_filters and ontology.iri in ontology_hash_filters:
                if ontology.hash not in ontology_hash_filters[ontology.iri]:
                    continue
            relevant_graphs.append(ontology.graph)
        if not relevant_graphs:
            return RDFGraph(), {}

        all_ns_map: dict[str, str] = {}
        for graph in relevant_graphs:
            for prefix, namespace in graph.namespaces():
                if prefix:
                    all_ns_map[prefix] = str(namespace)
        filtered_ns = _filter_overbroad_namespace_map(all_ns_map)

        merged_graph = RDFGraph()
        for prefix, uri in filtered_ns.items():
            merged_graph.bind(prefix, Namespace(uri))
        for graph in relevant_graphs:
            merged_graph += graph

        ontology_subjects: frozenset[str] = frozenset(
            str(s) for s, _, _ in merged_graph.triples((None, RDF.type, OWL.Ontology))
        )

        def should_include_expansion_triple(
            subj: object,
            pred: object,
            obj: object,
        ) -> bool:
            if not isinstance(pred, URIRef):
                return False
            if pred in _NOISY_EXPANSION_PREDICATES:
                return False
            if isinstance(subj, BNode) and isinstance(obj, BNode):
                return False
            if isinstance(subj, URIRef) and str(subj) in ontology_subjects:
                return False
            return True

        if not entity_uris:
            return RDFGraph(), {}
        seed_uris_ranked = list(dict.fromkeys(uri for uri in entity_uris if uri))
        if not seed_uris_ranked:
            return RDFGraph(), {}
        result = RDFGraph()
        for prefix, uri in filtered_ns.items():
            result.bind(prefix, Namespace(uri))

        if max_total_triples <= 0 or estimated_triples_per_query <= 0:
            return result, {}

        relevance = entity_relevance or {}
        roles = entity_roles or {}
        concept_seeds, property_seeds = _classify_and_promote_seeds(
            seed_uris_ranked, merged_graph, roles, ontology_subjects
        )
        property_seeds = _crosslink_property_seeds(
            merged_graph, concept_seeds, property_seeds, ontology_subjects
        )

        property_triple_budget = min(
            max(32, max_total_triples // 6),
            max_total_triples // 4,
        )
        property_triples_start = len(result)
        for prop_uri in property_seeds:
            if len(result) >= max_total_triples:
                break
            if len(result) - property_triples_start >= property_triple_budget:
                break
            for triple in merged_graph.triples((URIRef(prop_uri), None, None)):
                subj, pred, obj = triple
                if pred not in _PROPERTY_DEFINITION_PREDICATES:
                    continue
                if not should_include_expansion_triple(subj, pred, obj):
                    continue
                if triple in result:
                    continue
                result.add(triple)

        protected_uris = _protected_uris_for_snapshot(
            seed_uris_ranked, concept_seeds, property_seeds, merged_graph
        )

        if not concept_seeds:
            concept_for_finalize = list(
                dict.fromkeys(
                    str(obj)
                    for prop_uri in property_seeds
                    for _, pred, obj in merged_graph.triples(
                        (URIRef(prop_uri), None, None)
                    )
                    if pred in (RDFS.domain, RDFS.range) and isinstance(obj, URIRef)
                )
            )
            metrics = _finalize_induced_subgraph_snapshot(
                merged_graph,
                result,
                concept_for_finalize,
                concept_for_finalize,
                property_seeds,
                protected_uris,
                max_total_triples=max_total_triples,
                should_include=should_include_expansion_triple,
            )
            return result, metrics

        concept_relevance = _build_concept_relevance(
            seed_uris_ranked, merged_graph, relevance, ontology_subjects
        )
        sorted_seed_uris = sorted(
            concept_seeds,
            key=lambda uri: (-float(concept_relevance.get(uri, 0.0)), uri),
        )

        _schema_shell_for_concept_seeds(
            merged_graph,
            sorted_seed_uris,
            result,
            max_total_triples=max_total_triples,
            should_include=should_include_expansion_triple,
            ancestor_closure_depth=ancestor_closure_depth,
        )

        score_by_seed: dict[str, float] = {
            uri: float(concept_relevance.get(uri, 0.0)) for uri in sorted_seed_uris
        }
        score_total = sum(max(score, 0.0) for score in score_by_seed.values())
        if score_total <= 0.0:
            score_by_seed = {uri: 1.0 for uri in sorted_seed_uris}
            score_total = float(len(sorted_seed_uris))

        remaining = max_total_triples - len(result)
        per_entity_cap = max(1, estimated_triples_per_query)
        hub_count = (
            len(sorted_seed_uris)
            if hub_seed_count <= 0
            else min(hub_seed_count, len(sorted_seed_uris))
        )
        hub_seeds = sorted_seed_uris[:hub_count]
        tail_seeds = sorted_seed_uris[hub_count:]

        hub_budget = int(remaining * 0.65) if remaining > 0 else 0
        tail_budget = remaining - hub_budget

        if hub_seeds and hub_budget > 0:
            hub_quota_base = max(1, hub_budget // len(hub_seeds))
            for seed_uri in hub_seeds:
                if len(result) >= max_total_triples:
                    break
                quota = min(per_entity_cap, hub_quota_base)
                _bfs_expand_from_seed(
                    merged_graph,
                    seed_uri,
                    result,
                    max_total_triples=max_total_triples,
                    should_include=should_include_expansion_triple,
                    depth=depth,
                    quota=quota,
                )

        if tail_seeds and tail_budget > 0:
            tail_quota_total = tail_budget
            for seed_uri in tail_seeds:
                if tail_quota_total <= 0 or len(result) >= max_total_triples:
                    break
                weight = max(score_by_seed.get(seed_uri, 0.0), 0.0) / score_total
                quota = max(1, int(tail_quota_total * weight))
                quota = min(quota, per_entity_cap)
                _bfs_expand_from_seed(
                    merged_graph,
                    seed_uri,
                    result,
                    max_total_triples=max_total_triples,
                    should_include=should_include_expansion_triple,
                    depth=max(0, depth - 1),
                    quota=quota,
                )
                tail_quota_total -= quota

        metrics = _finalize_induced_subgraph_snapshot(
            merged_graph,
            result,
            sorted_seed_uris,
            concept_seeds,
            property_seeds,
            protected_uris,
            max_total_triples=max_total_triples,
            should_include=should_include_expansion_triple,
        )
        return result, metrics

    def get_induced_subgraph(
        self,
        entity_uris: list[str],
        entity_relevance: dict[str, float] | None = None,
        entity_roles: Mapping[str, str | None] | None = None,
        ontology_iris: list[str] | None = None,
        depth: int = 1,
        max_total_triples: int = 300,
        estimated_triples_per_query: int = 24,
        ontology_version_filters: dict[str, set[str]] | None = None,
        ontology_hash_filters: dict[str, set[str]] | None = None,
        hub_seed_count: int = 8,
        ancestor_closure_depth: int = 3,
    ) -> RDFGraph:
        """Fetch a deterministic induced subgraph around selected entities."""
        if self.triple_store_manager is None:
            return RDFGraph()
        if depth < 0:
            raise ValueError("depth must be >= 0")
        if max_total_triples <= 0:
            return RDFGraph()
        if estimated_triples_per_query <= 0:
            return RDFGraph()

        ontologies = self.triple_store_manager.fetch_ontologies()
        result, metrics = SPARQLTool._build_induced_subgraph(
            ontologies,
            entity_uris,
            entity_relevance,
            ontology_iris,
            depth,
            max_total_triples,
            estimated_triples_per_query,
            ontology_version_filters,
            ontology_hash_filters,
            entity_roles,
            hub_seed_count,
            ancestor_closure_depth,
        )
        self.last_finalize_metrics = metrics
        return result

    async def aget_induced_subgraph(
        self,
        entity_uris: list[str],
        entity_relevance: dict[str, float] | None = None,
        entity_roles: Mapping[str, str | None] | None = None,
        ontology_iris: list[str] | None = None,
        depth: int = 1,
        max_total_triples: int = 300,
        estimated_triples_per_query: int = 24,
        ontology_version_filters: dict[str, set[str]] | None = None,
        ontology_hash_filters: dict[str, set[str]] | None = None,
        hub_seed_count: int = 8,
        ancestor_closure_depth: int = 3,
    ) -> RDFGraph:
        """Like ``get_induced_subgraph`` but uses ``afetch_ontologies`` for I/O."""
        if self.triple_store_manager is None:
            return self.get_induced_subgraph(
                entity_uris=entity_uris,
                entity_relevance=entity_relevance,
                entity_roles=entity_roles,
                ontology_iris=ontology_iris,
                depth=depth,
                max_total_triples=max_total_triples,
                estimated_triples_per_query=estimated_triples_per_query,
                ontology_version_filters=ontology_version_filters,
                ontology_hash_filters=ontology_hash_filters,
                hub_seed_count=hub_seed_count,
                ancestor_closure_depth=ancestor_closure_depth,
            )
        if depth < 0:
            raise ValueError("depth must be >= 0")
        if max_total_triples <= 0:
            return RDFGraph()
        if estimated_triples_per_query <= 0:
            return RDFGraph()

        ontologies = await self.triple_store_manager.afetch_ontologies()
        result, metrics = await asyncio.to_thread(
            SPARQLTool._build_induced_subgraph,
            ontologies,
            entity_uris,
            entity_relevance,
            ontology_iris,
            depth,
            max_total_triples,
            estimated_triples_per_query,
            ontology_version_filters,
            ontology_hash_filters,
            entity_roles,
            hub_seed_count,
            ancestor_closure_depth,
        )
        self.last_finalize_metrics = metrics
        return result

    def create_insert_operation(
        self, query: str, description: str = ""
    ) -> SPARQLOperationModel:
        """Create an INSERT operation.

        Args:
            query: The SPARQL INSERT query.
            description: Optional description of the operation.

        Returns:
            SPARQLOperationModel: The created operation.
        """
        return SPARQLOperationModel(
            operation_type=SPARQLOperationType.INSERT,
            query=query,
            description=description,
        )

    def create_delete_operation(
        self, query: str, description: str = ""
    ) -> SPARQLOperationModel:
        """Create a DELETE operation.

        Args:
            query: The SPARQL DELETE query.
            description: Optional description of the operation.

        Returns:
            SPARQLOperationModel: The created operation.
        """
        return SPARQLOperationModel(
            operation_type=SPARQLOperationType.DELETE,
            query=query,
            description=description,
        )

    def create_update_operation(
        self, query: str, description: str = ""
    ) -> SPARQLOperationModel:
        """Create an UPDATE operation.

        Args:
            query: The SPARQL UPDATE query.
            description: Optional description of the operation.

        Returns:
            SPARQLOperationModel: The created operation.
        """
        return SPARQLOperationModel(
            operation_type=SPARQLOperationType.UPDATE,
            query=query,
            description=description,
        )

__init__(triple_store_manager=None)

Initialize SPARQL tool.

Parameters:

Name	Type	Description	Default
triple_store_manager	TripleStoreManager | None	Optional triple store manager for persistent storage.	None

Source code in ontocast/tool/sparql.py

def __init__(self, triple_store_manager: TripleStoreManager | None = None):
    """Initialize SPARQL tool.

    Args:
        triple_store_manager: Optional triple store manager for persistent storage.
    """
    self.triple_store_manager = triple_store_manager
    self.operation_history = []
    self.last_finalize_metrics: dict[str, int] = {}

aget_induced_subgraph(entity_uris, entity_relevance=None, entity_roles=None, ontology_iris=None, depth=1, max_total_triples=300, estimated_triples_per_query=24, ontology_version_filters=None, ontology_hash_filters=None, hub_seed_count=8, ancestor_closure_depth=3) async

Like get_induced_subgraph but uses afetch_ontologies for I/O.

Source code in ontocast/tool/sparql.py

async def aget_induced_subgraph(
    self,
    entity_uris: list[str],
    entity_relevance: dict[str, float] | None = None,
    entity_roles: Mapping[str, str | None] | None = None,
    ontology_iris: list[str] | None = None,
    depth: int = 1,
    max_total_triples: int = 300,
    estimated_triples_per_query: int = 24,
    ontology_version_filters: dict[str, set[str]] | None = None,
    ontology_hash_filters: dict[str, set[str]] | None = None,
    hub_seed_count: int = 8,
    ancestor_closure_depth: int = 3,
) -> RDFGraph:
    """Like ``get_induced_subgraph`` but uses ``afetch_ontologies`` for I/O."""
    if self.triple_store_manager is None:
        return self.get_induced_subgraph(
            entity_uris=entity_uris,
            entity_relevance=entity_relevance,
            entity_roles=entity_roles,
            ontology_iris=ontology_iris,
            depth=depth,
            max_total_triples=max_total_triples,
            estimated_triples_per_query=estimated_triples_per_query,
            ontology_version_filters=ontology_version_filters,
            ontology_hash_filters=ontology_hash_filters,
            hub_seed_count=hub_seed_count,
            ancestor_closure_depth=ancestor_closure_depth,
        )
    if depth < 0:
        raise ValueError("depth must be >= 0")
    if max_total_triples <= 0:
        return RDFGraph()
    if estimated_triples_per_query <= 0:
        return RDFGraph()

    ontologies = await self.triple_store_manager.afetch_ontologies()
    result, metrics = await asyncio.to_thread(
        SPARQLTool._build_induced_subgraph,
        ontologies,
        entity_uris,
        entity_relevance,
        ontology_iris,
        depth,
        max_total_triples,
        estimated_triples_per_query,
        ontology_version_filters,
        ontology_hash_filters,
        entity_roles,
        hub_seed_count,
        ancestor_closure_depth,
    )
    self.last_finalize_metrics = metrics
    return result

clear_history()

Clear the operation history.

Source code in ontocast/tool/sparql.py

def clear_history(self):
    """Clear the operation history."""
    self.operation_history.clear()

create_delete_operation(query, description='')

Create a DELETE operation.

Parameters:

Name	Type	Description	Default
query	str	The SPARQL DELETE query.	required
description	str	Optional description of the operation.	''

Returns:

Name	Type	Description
SPARQLOperationModel	SPARQLOperationModel	The created operation.

Source code in ontocast/tool/sparql.py

def create_delete_operation(
    self, query: str, description: str = ""
) -> SPARQLOperationModel:
    """Create a DELETE operation.

    Args:
        query: The SPARQL DELETE query.
        description: Optional description of the operation.

    Returns:
        SPARQLOperationModel: The created operation.
    """
    return SPARQLOperationModel(
        operation_type=SPARQLOperationType.DELETE,
        query=query,
        description=description,
    )

create_insert_operation(query, description='')

Create an INSERT operation.

Parameters:

Name	Type	Description	Default
query	str	The SPARQL INSERT query.	required
description	str	Optional description of the operation.	''

Returns:

Name	Type	Description
SPARQLOperationModel	SPARQLOperationModel	The created operation.

Source code in ontocast/tool/sparql.py

def create_insert_operation(
    self, query: str, description: str = ""
) -> SPARQLOperationModel:
    """Create an INSERT operation.

    Args:
        query: The SPARQL INSERT query.
        description: Optional description of the operation.

    Returns:
        SPARQLOperationModel: The created operation.
    """
    return SPARQLOperationModel(
        operation_type=SPARQLOperationType.INSERT,
        query=query,
        description=description,
    )

create_update_operation(query, description='')

Create an UPDATE operation.

Parameters:

Name	Type	Description	Default
query	str	The SPARQL UPDATE query.	required
description	str	Optional description of the operation.	''

Returns:

Name	Type	Description
SPARQLOperationModel	SPARQLOperationModel	The created operation.

Source code in ontocast/tool/sparql.py

def create_update_operation(
    self, query: str, description: str = ""
) -> SPARQLOperationModel:
    """Create an UPDATE operation.

    Args:
        query: The SPARQL UPDATE query.
        description: Optional description of the operation.

    Returns:
        SPARQLOperationModel: The created operation.
    """
    return SPARQLOperationModel(
        operation_type=SPARQLOperationType.UPDATE,
        query=query,
        description=description,
    )

execute_operation(operation)

Execute a single SPARQL operation.

Parameters:

Name	Type	Description	Default
operation	SPARQLOperationModel	The SPARQL operation to execute.	required

Source code in ontocast/tool/sparql.py

def execute_operation(self, operation: SPARQLOperationModel) -> None:
    """Execute a single SPARQL operation.

    Args:
        operation: The SPARQL operation to execute.
    """
    # For now, we'll use a simple approach - in a real implementation,
    # you might want to track which graph this operation should be applied to
    logger.info(
        f"Executing {operation.operation_type} operation: {operation.description}"
    )

execute_operations(graph, operations)

Execute a list of SPARQL operations on a graph.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	The RDF graph to operate on.	required
operations	list[SPARQLOperationModel]	List of SPARQL operations to execute.	required

Returns:

Name	Type	Description
RDFGraph	RDFGraph	Updated graph after applying operations.

Source code in ontocast/tool/sparql.py

def execute_operations(
    self, graph: RDFGraph, operations: list[SPARQLOperationModel]
) -> RDFGraph:
    """Execute a list of SPARQL operations on a graph.

    Args:
        graph: The RDF graph to operate on.
        operations: List of SPARQL operations to execute.

    Returns:
        RDFGraph: Updated graph after applying operations.
    """
    logger.info(f"Executing {len(operations)} SPARQL operations")

    for operation in operations:
        try:
            self._execute_single_operation(graph, operation)
            self.operation_history.append(operation)
            logger.debug(
                f"Executed {operation.operation_type} operation: {operation.description}"
            )
        except Exception as e:
            logger.error(
                f"Failed to execute {operation.operation_type} operation: {str(e)}"
            )
            raise

    return graph

get_induced_subgraph(entity_uris, entity_relevance=None, entity_roles=None, ontology_iris=None, depth=1, max_total_triples=300, estimated_triples_per_query=24, ontology_version_filters=None, ontology_hash_filters=None, hub_seed_count=8, ancestor_closure_depth=3)

Fetch a deterministic induced subgraph around selected entities.

Source code in ontocast/tool/sparql.py

def get_induced_subgraph(
    self,
    entity_uris: list[str],
    entity_relevance: dict[str, float] | None = None,
    entity_roles: Mapping[str, str | None] | None = None,
    ontology_iris: list[str] | None = None,
    depth: int = 1,
    max_total_triples: int = 300,
    estimated_triples_per_query: int = 24,
    ontology_version_filters: dict[str, set[str]] | None = None,
    ontology_hash_filters: dict[str, set[str]] | None = None,
    hub_seed_count: int = 8,
    ancestor_closure_depth: int = 3,
) -> RDFGraph:
    """Fetch a deterministic induced subgraph around selected entities."""
    if self.triple_store_manager is None:
        return RDFGraph()
    if depth < 0:
        raise ValueError("depth must be >= 0")
    if max_total_triples <= 0:
        return RDFGraph()
    if estimated_triples_per_query <= 0:
        return RDFGraph()

    ontologies = self.triple_store_manager.fetch_ontologies()
    result, metrics = SPARQLTool._build_induced_subgraph(
        ontologies,
        entity_uris,
        entity_relevance,
        ontology_iris,
        depth,
        max_total_triples,
        estimated_triples_per_query,
        ontology_version_filters,
        ontology_hash_filters,
        entity_roles,
        hub_seed_count,
        ancestor_closure_depth,
    )
    self.last_finalize_metrics = metrics
    return result

get_operation_history()

Get the history of executed operations.

Returns:

Type	Description
list[SPARQLOperationModel]	List of executed operations.

Source code in ontocast/tool/sparql.py

def get_operation_history(self) -> list[SPARQLOperationModel]:
    """Get the history of executed operations.

    Returns:
        List of executed operations.
    """
    return self.operation_history.copy()

validate_operation(operation)

Validate a SPARQL operation.

Parameters:

Name	Type	Description	Default
operation	SPARQLOperationModel	The operation to validate.	required

Returns:

Name	Type	Description
bool	bool	True if valid, False otherwise.

Source code in ontocast/tool/sparql.py

def validate_operation(self, operation: SPARQLOperationModel) -> bool:
    """Validate a SPARQL operation.

    Args:
        operation: The operation to validate.

    Returns:
        bool: True if valid, False otherwise.
    """
    try:
        prepareQuery(operation.query)
        return True
    except Exception as e:
        logger.error(f"Invalid SPARQL operation: {str(e)}")
        return False