`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.

Attributes¶

`logger = logging.getLogger(name)` `module-attribute` ¶

Classes¶

`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.last_finalize_metrics: dict[str, int] = {}

    @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

Attributes¶

`last_finalize_metrics = {}` `instance-attribute` ¶

`triple_store_manager = triple_store_manager` `instance-attribute` ¶

Methods:¶

`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.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

`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

Functions:¶

`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:

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.
Triples incident to any node within depth hops of a seed -- what :func:_bfs_expand_from_seed visits and materializes.
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.
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]

ontocast.tool.sparql¶

Attributes¶

logger = logging.getLogger(__name__) module-attribute ¶

Classes¶

SPARQLTool ¶

Attributes¶

last_finalize_metrics = {} instance-attribute ¶

triple_store_manager = triple_store_manager instance-attribute ¶

Methods:¶

__init__(triple_store_manager=None) ¶

Functions:¶

build_candidate_subgraph_query(seed_irefs, graph_irefs, *, depth) ¶

filter_overbroad_namespace_map(ns_map) ¶

merge_ontology_graphs(ontologies) ¶

select_relevant_ontologies(ontologies, ontology_iris, ontology_version_filters, ontology_hash_filters) ¶

`ontocast.tool.sparql`¶

`logger = logging.getLogger(name)` `module-attribute` ¶

`SPARQLTool` ¶

`last_finalize_metrics = {}` `instance-attribute` ¶

`triple_store_manager = triple_store_manager` `instance-attribute` ¶

`init(triple_store_manager=None)` ¶

`build_candidate_subgraph_query(seed_irefs, graph_irefs, *, depth)` ¶

`filter_overbroad_namespace_map(ns_map)` ¶

`merge_ontology_graphs(ontologies)` ¶

`select_relevant_ontologies(ontologies, ontology_iris, ontology_version_filters, ontology_hash_filters)` ¶