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,
        merged: tuple[RDFGraph, dict[str, str]] | None = None,
        type_promotion_score_factor: float = 1.0,
        seed_order: str = "score",
        entity_groups: Mapping[str, str] | None = None,
        extra_description_predicates: Sequence[URIRef] = (),
    ) -> tuple[RDFGraph, dict[str, int]]:
        """Merge filtered graphs; schema shell, hub BFS, and connectivity repair.

        Args:
            merged: Pre-merged ``(graph, prefix_map)`` for the already-filtered
                ontologies. When supplied, ``ontologies`` and the three filter
                arguments are not consulted -- the caller has resolved them.
            type_promotion_score_factor: Fraction of a seed's retrieval score
                inherited by its promoted type IRIs.
            seed_order: ``"score"`` expands seeds in global score order;
                ``"ontology_round_robin"`` interleaves ontology groups
                (requires ``entity_groups``).
            entity_groups: Seed IRI -> ontology IRI, used by round-robin ordering.
            extra_description_predicates: Additional description predicates (e.g.
                symbol/notation annotations) admitted for seed nodes.
        """

        if merged is None:
            relevant = select_relevant_ontologies(
                ontologies,
                ontology_iris,
                ontology_version_filters,
                ontology_hash_filters,
            )
            if not relevant:
                return RDFGraph(), {}
            merged_graph, filtered_ns = merge_ontology_graphs(relevant)
        else:
            merged_graph, filtered_ns = merged
            if len(merged_graph) == 0:
                return RDFGraph(), {}

        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(), {}
        # Prefixes are bound only after the snapshot is built (_bind_used_prefixes):
        # binding every merged ontology's prefixes up front advertised namespaces
        # downstream prompts could not see a single term from.
        result = RDFGraph()

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

        description_predicates = _compose_description_predicates(
            tuple(extra_description_predicates)
        )
        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,
                description_predicates=description_predicates,
            )
            _bind_used_prefixes(result, filtered_ns, merged_graph.declared_prefix_map())
            return result, metrics

        concept_relevance, first_rank = _build_concept_relevance(
            seed_uris_ranked,
            merged_graph,
            relevance,
            ontology_subjects,
            type_promotion_score_factor=type_promotion_score_factor,
        )
        default_rank = len(seed_uris_ranked)
        sorted_seed_uris = sorted(
            concept_seeds,
            key=lambda uri: (
                -float(concept_relevance.get(uri, 0.0)),
                first_rank.get(uri, default_rank),
                uri,
            ),
        )
        if seed_order == "ontology_round_robin" and entity_groups:
            sorted_seed_uris = _interleave_by_group(
                sorted_seed_uris,
                _expand_groups_to_promoted(
                    seed_uris_ranked, merged_graph, ontology_subjects, entity_groups
                ),
            )

        _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,
            description_predicates=description_predicates,
        )

        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,
                    description_predicates=description_predicates,
                )

        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,
                    description_predicates=description_predicates,
                )
                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,
            description_predicates=description_predicates,
        )
        _bind_used_prefixes(result, filtered_ns, merged_graph.declared_prefix_map())
        return result, metrics

    def _fetch_ontologies_sync(self, ontology_iris: list[str] | None) -> list[Ontology]:
        """Read only the requested ontologies, from synchronous context.

        Raises:
            RuntimeError: If called while an event loop is running. Use
                :meth:`aget_induced_subgraph` from async code.
        """
        manager = self.triple_store_manager
        assert manager is not None
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(manager.afetch_ontologies_by_iri(ontology_iris or []))
        raise RuntimeError(
            "get_induced_subgraph() cannot fetch inside async code; "
            "use await aget_induced_subgraph()"
        )

    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,
        ontologies: list[Ontology] | None = None,
        merged: tuple[RDFGraph, dict[str, str]] | None = None,
        type_promotion_score_factor: float = 1.0,
        seed_order: str = "score",
        entity_groups: Mapping[str, str] | None = None,
        extra_description_predicates: Sequence[URIRef] = (),
    ) -> RDFGraph:
        """Fetch a deterministic induced subgraph around selected entities.

        This is a primitive: ``SPARQLTool`` holds no vector-store settings, so
        the budget arguments here are conservative literals, *not* the
        deployment's configured budget. ``OntologyPatchRetriever`` -- the
        production caller -- passes every one of them from
        ``ONTOLOGY_PATCH_INDUCED_SUBGRAPH_*``. Direct callers who want the
        configured behaviour should do the same rather than rely on these.

        Args:
            ontologies: Pre-fetched catalog to build from. When ``None`` only the
                ontologies named by ``ontology_iris`` are read from the triple
                store (the whole catalog when ``ontology_iris`` is empty).
            merged: Pre-merged ``(graph, prefix_map)``. Supplying it skips both
                the fetch and the merge entirely.
        """
        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()

        if merged is None and ontologies is None:
            ontologies = self._fetch_ontologies_sync(ontology_iris)
        result, metrics = SPARQLTool._build_induced_subgraph(
            ontologies or [],
            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,
            merged,
            type_promotion_score_factor,
            seed_order,
            entity_groups,
            extra_description_predicates,
        )
        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,
        ontologies: list[Ontology] | None = None,
        merged: tuple[RDFGraph, dict[str, str]] | None = None,
        type_promotion_score_factor: float = 1.0,
        seed_order: str = "score",
        entity_groups: Mapping[str, str] | None = None,
        extra_description_predicates: Sequence[URIRef] = (),
    ) -> RDFGraph:
        """Like ``get_induced_subgraph`` but uses ``afetch_ontologies`` for I/O.

        Args:
            ontologies: Pre-fetched catalog to build from. When ``None`` only the
                ontologies named by ``ontology_iris`` are read from the triple
                store (the whole catalog when ``ontology_iris`` is empty).
            merged: Pre-merged ``(graph, prefix_map)``. Supplying it skips both
                the fetch and the merge entirely.
        """
        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,
                ontologies=ontologies,
                merged=merged,
                type_promotion_score_factor=type_promotion_score_factor,
                seed_order=seed_order,
                entity_groups=entity_groups,
                extra_description_predicates=extra_description_predicates,
            )
        if depth < 0:
            raise ValueError("depth must be >= 0")
        if max_total_triples <= 0:
            return RDFGraph()
        if estimated_triples_per_query <= 0:
            return RDFGraph()

        if merged is None and ontologies is None:
            ontologies = await self.triple_store_manager.afetch_ontologies_by_iri(
                ontology_iris or []
            )
        result, metrics = await asyncio.to_thread(
            SPARQLTool._build_induced_subgraph,
            ontologies or [],
            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,
            merged,
            type_promotion_score_factor,
            seed_order,
            entity_groups,
            extra_description_predicates,
        )
        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, ontologies=None, merged=None, type_promotion_score_factor=1.0, seed_order='score', entity_groups=None, extra_description_predicates=()) async

Like get_induced_subgraph but uses afetch_ontologies for I/O.

Parameters:

Name	Type	Description	Default
ontologies	list[Ontology] | None	Pre-fetched catalog to build from. When None only the ontologies named by ontology_iris are read from the triple store (the whole catalog when ontology_iris is empty).	None
merged	tuple[RDFGraph, dict[str, str]] | None	Pre-merged (graph, prefix_map). Supplying it skips both the fetch and the merge entirely.	None

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,
    ontologies: list[Ontology] | None = None,
    merged: tuple[RDFGraph, dict[str, str]] | None = None,
    type_promotion_score_factor: float = 1.0,
    seed_order: str = "score",
    entity_groups: Mapping[str, str] | None = None,
    extra_description_predicates: Sequence[URIRef] = (),
) -> RDFGraph:
    """Like ``get_induced_subgraph`` but uses ``afetch_ontologies`` for I/O.

    Args:
        ontologies: Pre-fetched catalog to build from. When ``None`` only the
            ontologies named by ``ontology_iris`` are read from the triple
            store (the whole catalog when ``ontology_iris`` is empty).
        merged: Pre-merged ``(graph, prefix_map)``. Supplying it skips both
            the fetch and the merge entirely.
    """
    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,
            ontologies=ontologies,
            merged=merged,
            type_promotion_score_factor=type_promotion_score_factor,
            seed_order=seed_order,
            entity_groups=entity_groups,
            extra_description_predicates=extra_description_predicates,
        )
    if depth < 0:
        raise ValueError("depth must be >= 0")
    if max_total_triples <= 0:
        return RDFGraph()
    if estimated_triples_per_query <= 0:
        return RDFGraph()

    if merged is None and ontologies is None:
        ontologies = await self.triple_store_manager.afetch_ontologies_by_iri(
            ontology_iris or []
        )
    result, metrics = await asyncio.to_thread(
        SPARQLTool._build_induced_subgraph,
        ontologies or [],
        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,
        merged,
        type_promotion_score_factor,
        seed_order,
        entity_groups,
        extra_description_predicates,
    )
    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, ontologies=None, merged=None, type_promotion_score_factor=1.0, seed_order='score', entity_groups=None, extra_description_predicates=())

Fetch a deterministic induced subgraph around selected entities.

This is a primitive: SPARQLTool holds no vector-store settings, so the budget arguments here are conservative literals, not the deployment's configured budget. OntologyPatchRetriever -- the production caller -- passes every one of them from ONTOLOGY_PATCH_INDUCED_SUBGRAPH_*. Direct callers who want the configured behaviour should do the same rather than rely on these.

Parameters:

Name	Type	Description	Default
ontologies	list[Ontology] | None	Pre-fetched catalog to build from. When None only the ontologies named by ontology_iris are read from the triple store (the whole catalog when ontology_iris is empty).	None
merged	tuple[RDFGraph, dict[str, str]] | None	Pre-merged (graph, prefix_map). Supplying it skips both the fetch and the merge entirely.	None

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,
    ontologies: list[Ontology] | None = None,
    merged: tuple[RDFGraph, dict[str, str]] | None = None,
    type_promotion_score_factor: float = 1.0,
    seed_order: str = "score",
    entity_groups: Mapping[str, str] | None = None,
    extra_description_predicates: Sequence[URIRef] = (),
) -> RDFGraph:
    """Fetch a deterministic induced subgraph around selected entities.

    This is a primitive: ``SPARQLTool`` holds no vector-store settings, so
    the budget arguments here are conservative literals, *not* the
    deployment's configured budget. ``OntologyPatchRetriever`` -- the
    production caller -- passes every one of them from
    ``ONTOLOGY_PATCH_INDUCED_SUBGRAPH_*``. Direct callers who want the
    configured behaviour should do the same rather than rely on these.

    Args:
        ontologies: Pre-fetched catalog to build from. When ``None`` only the
            ontologies named by ``ontology_iris`` are read from the triple
            store (the whole catalog when ``ontology_iris`` is empty).
        merged: Pre-merged ``(graph, prefix_map)``. Supplying it skips both
            the fetch and the merge entirely.
    """
    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()

    if merged is None and ontologies is None:
        ontologies = self._fetch_ontologies_sync(ontology_iris)
    result, metrics = SPARQLTool._build_induced_subgraph(
        ontologies or [],
        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,
        merged,
        type_promotion_score_factor,
        seed_order,
        entity_groups,
        extra_description_predicates,
    )
    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

build_candidate_subgraph_query(seed_irefs, graph_irefs, *, depth)

Build a CONSTRUCT for everything :func:_build_induced_subgraph may read.

Five branches, each a direct translation of a read pattern in the builder:

1. owl:Ontology header triples, which populate the ontology_subjects exclusion set — plus the sh:declare blank-node subtrees hanging off them, so persisted author prefix names reach the candidate path too.
2. Triples incident to any node within depth hops of a seed -- what :func:_bfs_expand_from_seed visits and materializes.
3. Triples incident to the rdfs:subClassOf ancestors of the seeds and of their types -- :func:_add_subclass_ancestor_closure after seed promotion. Unbounded * rather than the configured hop limit, deliberately: a superset is safe, a subset is not.
4. Definition triples of properties whose rdfs:domain/rdfs:range is a seed or a seed's type -- :func:_crosslink_property_seeds.

Not covered: the cross-component schema-path repair (:func:_find_schema_path_in_merged_graph) can search up to _SCHEMA_PATH_MAX_DEPTH hops from nodes that are themselves depth + 1 hops out, so a bridge may lie outside this candidate set. The consequence is a missing bridge -- a smaller, still-correct snapshot -- never a wrong triple.

Parameters:

Name	Type	Description	Default
seed_irefs	Sequence[str]	Seed IRIs, already escaped as <iri> IRIREFs.	required
graph_irefs	Sequence[str]	Named graph IRIs to restrict to, escaped as IRIREFs.	required
depth	int	Neighborhood hop count, matching the builder's depth.	required

Returns:

Name	Type	Description
str	str	A SPARQL CONSTRUCT query.

Source code in ontocast/tool/sparql.py

def build_candidate_subgraph_query(
    seed_irefs: Sequence[str],
    graph_irefs: Sequence[str],
    *,
    depth: int,
) -> str:
    """Build a CONSTRUCT for everything :func:`_build_induced_subgraph` may read.

    Five branches, each a direct translation of a read pattern in the builder:

    1. ``owl:Ontology`` header triples, which populate the ``ontology_subjects``
       exclusion set — plus the ``sh:declare`` blank-node subtrees hanging off
       them, so persisted author prefix names reach the candidate path too.
    2. Triples incident to any node within ``depth`` hops of a seed -- what
       :func:`_bfs_expand_from_seed` visits and materializes.
    3. Triples incident to the ``rdfs:subClassOf`` ancestors of the seeds and of
       their types -- :func:`_add_subclass_ancestor_closure` after seed promotion.
       Unbounded ``*`` rather than the configured hop limit, deliberately: a
       superset is safe, a subset is not.
    4. Definition triples of properties whose ``rdfs:domain``/``rdfs:range`` is a
       seed or a seed's type -- :func:`_crosslink_property_seeds`.

    Not covered: the cross-component schema-path repair
    (:func:`_find_schema_path_in_merged_graph`) can search up to
    ``_SCHEMA_PATH_MAX_DEPTH`` hops from nodes that are themselves ``depth + 1``
    hops out, so a bridge may lie outside this candidate set. The consequence is a
    *missing* bridge -- a smaller, still-correct snapshot -- never a wrong triple.

    Args:
        seed_irefs: Seed IRIs, already escaped as ``<iri>`` IRIREFs.
        graph_irefs: Named graph IRIs to restrict to, escaped as IRIREFs.
        depth: Neighborhood hop count, matching the builder's ``depth``.

    Returns:
        str: A SPARQL CONSTRUCT query.
    """
    step = _bidirectional_non_noisy_step()
    # Hop 0 repeats the seeds as ``VALUES ?node`` rather than ``BIND(?seed AS ?node)``:
    # a BIND in its own group cannot see ``?seed`` from the enclosing group, so it
    # would leave ``?node`` unbound and the incident pattern would match every
    # triple in the dataset.
    ball_branches = ["{{ VALUES ?node {{ {} }} }}".format(" ".join(seed_irefs))]
    ball_branches += [
        "{{ ?seed {} ?node }}".format("/".join([step] * hops))
        for hops in range(1, max(0, depth) + 1)
    ]
    incident = (
        "{ { ?node ?p ?o . BIND(?node AS ?s) } UNION "
        "{ ?s ?p ?node . BIND(?node AS ?o) } }"
    )
    # ``FROM``, not ``GRAPH ?g``: a GRAPH block binds one graph for the whole
    # pattern, so a path could never cross an ontology boundary -- which is exactly
    # the cross-ontology ``rdfs:subClassOf`` case this retrieval exists to follow.
    # ``FROM`` merges the selected graphs into the default graph first, matching
    # what :func:`merge_ontology_graphs` does in Python.
    from_clause = "\n".join(f"FROM {iref}" for iref in graph_irefs)
    return f"""
PREFIX owl: <http://www.w3.org/2002/07/owl#>
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX sh: <http://www.w3.org/ns/shacl#>
CONSTRUCT {{ ?s ?p ?o }}
{from_clause}
WHERE {{
  VALUES ?seed {{ {" ".join(seed_irefs)} }}
  {{ ?s a owl:Ontology . ?s ?p ?o }}
  UNION
  {{ ?onto a owl:Ontology . ?onto sh:declare ?s . ?s ?p ?o }}
  UNION
  {{ {" UNION ".join(ball_branches)} {incident} }}
  UNION
  {{ ?seed rdf:type?/rdfs:subClassOf* ?anc .
     {{ {{ ?anc ?p ?o . BIND(?anc AS ?s) }} UNION
       {{ ?s ?p ?anc . BIND(?anc AS ?o) }} }} }}
  UNION
  {{ ?seed rdf:type? ?cls .
     ?prop rdfs:domain|rdfs:range ?cls .
     ?prop ?p ?o . BIND(?prop AS ?s) }}
}}
"""

filter_overbroad_namespace_map(ns_map)

Drop namespace bindings whose URI is a strict prefix of another in the map.

Source code in ontocast/tool/sparql.py

def filter_overbroad_namespace_map(ns_map: dict[str, str]) -> dict[str, str]:
    """Drop namespace bindings whose URI is a strict prefix of another in the map."""
    all_ns_uris = set(ns_map.values())
    return {
        prefix: uri
        for prefix, uri in ns_map.items()
        if not any(other != uri and other.startswith(uri) for other in all_ns_uris)
    }

merge_ontology_graphs(ontologies)

Union ontology graphs into one graph carrying their prefix bindings.

Prefix bindings are harvested from each source graph's namespace manager -- they are serialization metadata rather than triples, so they only exist here because the sources were parsed from Turtle.

The result is treated as read-only by every consumer: the induced-subgraph builder reads it as an oracle and writes exclusively to its own result graph. That is what makes the merge safe to cache and share across content units.

Parameters:

Name	Type	Description	Default
ontologies	Sequence[Ontology]	Ontology versions to merge.	required

Returns:

Name	Type	Description
tuple	RDFGraph	The merged graph and the surviving prefix → namespace map, which
	dict[str, str]	the caller binds onto the snapshot it builds. The map is returned rather
	tuple[RDFGraph, dict[str, str]]	than re-read from the merged graph so callers see exactly the author
	tuple[RDFGraph, dict[str, str]]	bindings, not rdflib's built-in ones.

Source code in ontocast/tool/sparql.py

def merge_ontology_graphs(
    ontologies: Sequence[Ontology],
) -> tuple[RDFGraph, dict[str, str]]:
    """Union ontology graphs into one graph carrying their prefix bindings.

    Prefix bindings are harvested from each source graph's namespace manager --
    they are serialization metadata rather than triples, so they only exist here
    because the sources were parsed from Turtle.

    The result is treated as read-only by every consumer: the induced-subgraph
    builder reads it as an oracle and writes exclusively to its own result graph.
    That is what makes the merge safe to cache and share across content units.

    Args:
        ontologies: Ontology versions to merge.

    Returns:
        tuple: The merged graph and the surviving prefix → namespace map, which
        the caller binds onto the snapshot it builds. The map is returned rather
        than re-read from the merged graph so callers see exactly the author
        bindings, not rdflib's built-in ones.
    """
    all_ns_map: dict[str, str] = {}
    for ontology in ontologies:
        for prefix, namespace in ontology.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 ontology in ontologies:
        merged_graph += ontology.graph
    return merged_graph, filtered_ns

select_relevant_ontologies(ontologies, ontology_iris, ontology_version_filters, ontology_hash_filters)

Filter a catalog down to the ontologies an induced subgraph may draw on.

An empty ontology_iris means "no restriction". Version and hash filters only apply to IRIs they mention, so an ontology absent from both passes through untouched.

Generic over lineage-bearing records so the same predicate runs on graph-less :class:~ontocast.onto.ontology_header.OntologyHeader values -- which is what the SPARQL candidate path filters, having no graphs to filter.

Parameters:

Name	Type	Description	Default
ontologies	Sequence[LineageT]	Candidate catalog ontologies or headers.	required
ontology_iris	list[str] | None	Allowed ontology IRIs, or empty/None for all.	required
ontology_version_filters	dict[str, set[str]] | None	Allowed semantic versions per ontology IRI.	required
ontology_hash_filters	dict[str, set[str]] | None	Allowed content hashes per ontology IRI.	required

Returns:

Name	Type	Description
list	list[LineageT]	The surviving records, in input order.

Source code in ontocast/tool/sparql.py

def select_relevant_ontologies(
    ontologies: Sequence[LineageT],
    ontology_iris: list[str] | None,
    ontology_version_filters: dict[str, set[str]] | None,
    ontology_hash_filters: dict[str, set[str]] | None,
) -> list[LineageT]:
    """Filter a catalog down to the ontologies an induced subgraph may draw on.

    An empty ``ontology_iris`` means "no restriction". Version and hash filters
    only apply to IRIs they mention, so an ontology absent from both passes
    through untouched.

    Generic over lineage-bearing records so the same predicate runs on graph-less
    :class:`~ontocast.onto.ontology_header.OntologyHeader` values -- which is what
    the SPARQL candidate path filters, having no graphs to filter.

    Args:
        ontologies: Candidate catalog ontologies or headers.
        ontology_iris: Allowed ontology IRIs, or empty/None for all.
        ontology_version_filters: Allowed semantic versions per ontology IRI.
        ontology_hash_filters: Allowed content hashes per ontology IRI.

    Returns:
        list: The surviving records, in input order.
    """
    ontology_filter = set(ontology_iris or [])
    candidates: list[LineageT] = [
        ontology
        for ontology in ontologies
        if not ontology_filter or ontology.iri in ontology_filter
    ]
    by_iri: dict[str, list[LineageT]] = {}
    for ontology in candidates:
        by_iri.setdefault(ontology.iri, []).append(ontology)

    # Version/hash filters *select among* an IRI's catalog entries; they must never
    # discard an IRI wholesale. Atom payloads and catalog graphs are produced by
    # different processes, and graph hashes are not stable under serialization
    # round-trips (literal lexical forms are outside URDNA2015 canonicalization),
    # so an exact-hash requirement silently emptied whole ontologies out of the
    # prompt context. Relax per IRI: exact match → same-version → any catalog
    # entry, warning on each relaxation.
    kept_ids: set[int] = set()
    for iri, group in by_iri.items():
        if ontology_version_filters and iri in ontology_version_filters:
            allowed_versions = ontology_version_filters[iri]
            version_pass = [
                ontology
                for ontology in group
                if (str(ontology.version) if ontology.version is not None else None)
                in allowed_versions
            ]
            if not version_pass:
                logger.warning(
                    "Ontology %s: no catalog entry matches retrieval versions %s; "
                    "falling back to all %d catalog entr(ies) for this IRI",
                    iri,
                    sorted(allowed_versions),
                    len(group),
                )
                version_pass = list(group)
        else:
            version_pass = list(group)

        if ontology_hash_filters and iri in ontology_hash_filters:
            allowed_hashes = ontology_hash_filters[iri]
            hash_pass = [
                ontology for ontology in version_pass if ontology.hash in allowed_hashes
            ]
            if not hash_pass:
                logger.warning(
                    "Ontology %s: no catalog entry matches retrieval hashes "
                    "(catalog identity drift, e.g. serialization round-trip); "
                    "falling back to %d same-version entr(ies)",
                    iri,
                    len(version_pass),
                )
                hash_pass = version_pass
        else:
            hash_pass = version_pass
        kept_ids.update(id(ontology) for ontology in hash_pass)

    return [ontology for ontology in candidates if id(ontology) in kept_ids]