graflo.data_source.rdf

RDF data source hierarchy.

Provides two concrete data sources that share a common abstract parent:

• :class:RdfFileDataSource – reads local RDF files (Turtle, RDF/XML, N3, JSON-LD, …) via rdflib.
• :class:SparqlEndpointDataSource – queries a remote SPARQL endpoint (e.g. Apache Fuseki) via SPARQLWrapper.

Both convert RDF triples into flat dictionaries grouped by subject URI, one dict per rdf:Class instance. Each document has _uri (full subject URI) and _key (URI local name — the fragment or last path segment). All other identity decisions (e.g. using a domain-specific literal as the storage key) belong in the schema layer, not here.

Two kinds of subject have no IRI of their own to key on, and both sources treat them alike:

• A blank node is keyed on its content: _uri is _:<key> and _key the key, the same on every read. A reference to it carries that _uri.
• IRIs joined by owl:sameAs are one document under the smallest IRI; the others are listed in _same_as and references to them are rewritten. same_as="keep" reads the statements as an ordinary property instead.

Uses rdflib and SPARQLWrapper, which are core dependencies of graflo (see pyproject.toml).

Attributes

RDF_TYPE = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type' module-attribute

SparqlDataSource = SparqlEndpointDataSource module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

RdfDataSource

Bases: AbstractDataSource, ABC

Abstract base for RDF data sources (file and endpoint).

Captures the fields and batch-yielding logic shared by both :class:RdfFileDataSource and :class:SparqlEndpointDataSource.

Attributes:

Name	Type	Description
rdf_class	str | None	Optional URI of the rdf:Class to filter subjects by.
same_as	Literal['collapse', 'keep']	collapse (default) reads IRIs joined by owl:sameAs as one document; keep reads the statements as a property.

Source code in graflo/data_source/rdf.py

class RdfDataSource(AbstractDataSource, abc.ABC):
    """Abstract base for RDF data sources (file and endpoint).

    Captures the fields and batch-yielding logic shared by both
    :class:`RdfFileDataSource` and :class:`SparqlEndpointDataSource`.

    Attributes:
        rdf_class: Optional URI of the ``rdf:Class`` to filter subjects by.
        same_as: ``collapse`` (default) reads IRIs joined by ``owl:sameAs`` as
            one document; ``keep`` reads the statements as a property.
    """

    source_type: DataSourceType = DataSourceType.SPARQL
    rdf_class: str | None = Field(
        default=None, description="URI of the rdf:Class to filter by"
    )
    same_as: Literal["collapse", "keep"] = Field(
        default="collapse",
        description="How owl:sameAs statements between IRIs are read.",
    )
    typed_objects: list[str] = Field(
        default_factory=list,
        description=(
            "Properties (local names) whose objects are also read by class, as "
            "``<property>@<Class>``: for a property with several ranges."
        ),
    )

    @staticmethod
    def _yield_batches(
        docs: list[dict], batch_size: int, limit: int | None
    ) -> Iterator[list[dict]]:
        """Apply *limit*, then yield *docs* in chunks of *batch_size*."""
        if limit is not None:
            docs = docs[:limit]
        for i in range(0, max(len(docs), 1), batch_size):
            batch = docs[i : i + batch_size]
            if batch:
                yield batch

Attributes

rdf_class = Field(default=None, description='URI of the rdf:Class to filter by') class-attribute instance-attribute

same_as = Field(default='collapse', description='How owl:sameAs statements between IRIs are read.') class-attribute instance-attribute

source_type = DataSourceType.SPARQL class-attribute instance-attribute

typed_objects = Field(default_factory=list, description='Properties (local names) whose objects are also read by class, as ``<property>@<Class>``: for a property with several ranges.') class-attribute instance-attribute

RdfFileDataSource

Bases: RdfDataSource

Data source for local RDF files.

Parses RDF files using rdflib and yields flat dictionaries grouped by subject URI. Optionally filters by rdf_class so that only instances of a specific class are returned.

Attributes:

Name	Type	Description
path	Path	Path to the RDF file.
rdf_format	str | None	Explicit rdflib format string (e.g. "turtle"). When None the format is guessed from the file extension.

Source code in graflo/data_source/rdf.py

class RdfFileDataSource(RdfDataSource):
    """Data source for local RDF files.

    Parses RDF files using *rdflib* and yields flat dictionaries grouped by
    subject URI.  Optionally filters by ``rdf_class`` so that only instances
    of a specific class are returned.

    Attributes:
        path: Path to the RDF file.
        rdf_format: Explicit rdflib format string (e.g. ``"turtle"``).
            When ``None`` the format is guessed from the file extension.
    """

    path: Path
    rdf_format: str | None = Field(
        default=None, description="rdflib serialization format"
    )

    def _resolve_format(self) -> str:
        """Return the rdflib format string, guessing from extension if needed."""
        if self.rdf_format:
            return self.rdf_format
        ext = self.path.suffix.lower()
        fmt = _EXT_FORMAT.get(ext)
        if fmt is None:
            raise ValueError(
                f"Cannot determine RDF format for extension '{ext}'. "
                f"Set rdf_format explicitly. Known: {list(_EXT_FORMAT.keys())}"
            )
        return fmt

    def iter_batches(
        self, batch_size: int = 1000, limit: int | None = None
    ) -> Iterator[list[dict]]:
        """Parse the RDF file and yield batches of flat dictionaries."""
        try:
            from rdflib import Graph
        except ImportError as exc:
            raise ImportError(
                "rdflib is required for RDF data sources. "
                "It is a core dependency of graflo; reinstall with "
                "`pip install --force-reinstall graflo` or install rdflib manually."
            ) from exc

        g = Graph()
        g.parse(str(self.path), format=self._resolve_format())
        logger.info(
            "Parsed %d triples from %s (format=%s)",
            len(g),
            self.path,
            self._resolve_format(),
        )

        docs = _triples_to_docs(
            g,
            rdf_class=self.rdf_class,
            same_as=self.same_as,
            typed_objects=self.typed_objects,
        )
        yield from self._yield_batches(docs, batch_size, limit)

Attributes

path instance-attribute

rdf_format = Field(default=None, description='rdflib serialization format') class-attribute instance-attribute

Methods:

iter_batches(batch_size=1000, limit=None)

Parse the RDF file and yield batches of flat dictionaries.

Source code in graflo/data_source/rdf.py

def iter_batches(
    self, batch_size: int = 1000, limit: int | None = None
) -> Iterator[list[dict]]:
    """Parse the RDF file and yield batches of flat dictionaries."""
    try:
        from rdflib import Graph
    except ImportError as exc:
        raise ImportError(
            "rdflib is required for RDF data sources. "
            "It is a core dependency of graflo; reinstall with "
            "`pip install --force-reinstall graflo` or install rdflib manually."
        ) from exc

    g = Graph()
    g.parse(str(self.path), format=self._resolve_format())
    logger.info(
        "Parsed %d triples from %s (format=%s)",
        len(g),
        self.path,
        self._resolve_format(),
    )

    docs = _triples_to_docs(
        g,
        rdf_class=self.rdf_class,
        same_as=self.same_as,
        typed_objects=self.typed_objects,
    )
    yield from self._yield_batches(docs, batch_size, limit)

SparqlEndpointDataSource

Bases: RdfDataSource

Data source that reads from a SPARQL endpoint.

Uses SPARQLWrapper to query an endpoint and returns flat dictionaries grouped by subject.

Attributes:

Name	Type	Description
config	SparqlSourceConfig	SPARQL source configuration.

Source code in graflo/data_source/rdf.py

class SparqlEndpointDataSource(RdfDataSource):
    """Data source that reads from a SPARQL endpoint.

    Uses ``SPARQLWrapper`` to query an endpoint and returns flat dictionaries
    grouped by subject.

    Attributes:
        config: SPARQL source configuration.
    """

    config: SparqlSourceConfig

    def _create_wrapper(self) -> Any:
        """Create a configured ``SPARQLWrapper`` instance."""
        try:
            from SPARQLWrapper import JSON, SPARQLWrapper
        except ImportError as exc:
            raise ImportError(
                "SPARQLWrapper is required for SPARQL endpoint data sources. "
                "It is a core dependency of graflo; reinstall with "
                "`pip install --force-reinstall graflo` or install SPARQLWrapper manually."
            ) from exc

        sparql = SPARQLWrapper(self.config.endpoint_url)
        sparql.setReturnFormat(JSON)
        if self.config.username and self.config.password:
            sparql.setCredentials(self.config.username, self.config.password)
        return sparql

    def _bindings(
        self, wrapper: Any, query_at: Callable[[int], str]
    ) -> Iterator[dict[str, Any]]:
        """The rows of a paged query, fetched a page at a time as they are read."""
        page_size = self.config.page_size
        offset = 0
        while True:
            query = query_at(offset)
            wrapper.setQuery(query)
            logger.debug("SPARQL query (offset=%d): %s", offset, query)
            bindings = wrapper.queryAndConvert().get("results", {}).get("bindings", [])
            yield from bindings
            if len(bindings) < page_size:
                return
            offset += page_size

    def iter_batches(
        self, batch_size: int = 1000, limit: int | None = None
    ) -> Iterator[list[dict]]:
        """Query the SPARQL endpoint and yield batches of flat dictionaries.

        Paginates with SPARQL LIMIT/OFFSET on **bindings** (triple rows), merges
        rows into subject documents in a streaming fashion, and stops fetching
        once *limit* subjects have been yielded (when set).

        Two more reads happen when the graph calls for them. The ``owl:sameAs``
        statements are read first (unless ``same_as`` is ``keep``); the members
        of a component are held back and yielded last, as one document. The
        graph's blank-node triples are read when the first blank node is met,
        which keys blank nodes on their content; this relies on the endpoint
        giving a blank node the same label in every query of the read.
        """
        wrapper = self._create_wrapper()
        config = self.config
        collapse = self.same_as == "collapse"
        components = SameAs(
            (binding["s"]["value"], binding["o"]["value"])
            for binding in (
                self._bindings(wrapper, config.same_as_query) if collapse else ()
            )
        )
        blank_triples = _BlankNodeTriples(
            lambda: self._bindings(wrapper, config.blank_node_query)
        )
        identity = _Identity(components, BlankNodeKeys(blank_triples.outgoing))

        def fill(doc: dict[str, Any], rows: Iterable[dict[str, Any]]) -> None:
            for binding in rows:
                _merge_sparql_binding_into_doc(
                    doc, binding, identity, collapse=collapse
                )

        batch: list[dict] = []
        emitted = 0
        # Documents of sameAs components, by canonical IRI, and the members
        # the query returned for each.
        held: dict[str, dict[str, Any]] = {}
        returned: dict[str, set[str]] = {}

        def full() -> bool:
            return limit is not None and emitted >= limit

        def types_of(iris: list[str]) -> dict[str, set[str]]:
            canonical = {
                member: iri for iri in iris for member in components.members(iri)
            }
            members = sorted(canonical)
            found: dict[str, set[str]] = {}
            for start in range(0, len(members), _SUBJECTS_PER_QUERY):
                chunk = members[start : start + _SUBJECTS_PER_QUERY]
                for binding in self._bindings(
                    wrapper,
                    lambda offset, chunk=chunk: config.types_query(chunk, offset),
                ):
                    if binding["o"]["type"] == "uri":
                        found.setdefault(canonical[binding["s"]["value"]], set()).add(
                            binding["o"]["value"]
                        )
            return found

        def out(docs: list[dict]) -> list[dict]:
            split_objects_by_type(docs, self.typed_objects, types_of)
            return docs

        rows_by_subject = groupby(
            self._bindings(wrapper, lambda offset: config.build_query(offset=offset)),
            key=lambda binding: (binding["s"]["type"], binding["s"]["value"]),
        )
        for _, rows in rows_by_subject:
            if full():
                break
            first = next(rows)
            subject = _binding_term(first["s"])
            if subject.kind == "bnode" and subject.value not in blank_triples:
                raise ValueError(
                    f"{config.endpoint_url} returned a blank node under a label "
                    "its blank-node triples do not carry. Blank nodes are keyed "
                    "on their triples, which needs blank-node labels that stay "
                    "the same across the queries of one read."
                )
            uri = identity.uri(subject)
            if subject.kind == "iri" and len(components.members(uri)) > 1:
                returned.setdefault(uri, set()).add(subject.value)
                doc = held.setdefault(uri, _new_doc(uri))
                fill(doc, [first, *rows])
                continue
            doc = _new_doc(uri)
            fill(doc, [first, *rows])
            batch.append(doc)
            emitted += 1
            if len(batch) >= batch_size:
                yield out(batch)
                batch = []

        for uri in sorted(held):
            if full():
                break
            members = components.members(uri)
            absent = [iri for iri in members if iri not in returned[uri]]
            for start in range(0, len(absent), _SUBJECTS_PER_QUERY):
                chunk = absent[start : start + _SUBJECTS_PER_QUERY]
                fill(
                    held[uri],
                    self._bindings(
                        wrapper,
                        lambda offset, chunk=chunk: config.subjects_query(
                            chunk, offset
                        ),
                    ),
                )
            held[uri]["_same_as"] = [iri for iri in members if iri != uri]
            batch.append(held[uri])
            emitted += 1
            if len(batch) >= batch_size:
                yield out(batch)
                batch = []

        if batch:
            yield out(batch)

Attributes

config instance-attribute

Methods:

iter_batches(batch_size=1000, limit=None)

Query the SPARQL endpoint and yield batches of flat dictionaries.

Paginates with SPARQL LIMIT/OFFSET on bindings (triple rows), merges rows into subject documents in a streaming fashion, and stops fetching once limit subjects have been yielded (when set).

Two more reads happen when the graph calls for them. The owl:sameAs statements are read first (unless same_as is keep); the members of a component are held back and yielded last, as one document. The graph's blank-node triples are read when the first blank node is met, which keys blank nodes on their content; this relies on the endpoint giving a blank node the same label in every query of the read.

Source code in graflo/data_source/rdf.py

def iter_batches(
    self, batch_size: int = 1000, limit: int | None = None
) -> Iterator[list[dict]]:
    """Query the SPARQL endpoint and yield batches of flat dictionaries.

    Paginates with SPARQL LIMIT/OFFSET on **bindings** (triple rows), merges
    rows into subject documents in a streaming fashion, and stops fetching
    once *limit* subjects have been yielded (when set).

    Two more reads happen when the graph calls for them. The ``owl:sameAs``
    statements are read first (unless ``same_as`` is ``keep``); the members
    of a component are held back and yielded last, as one document. The
    graph's blank-node triples are read when the first blank node is met,
    which keys blank nodes on their content; this relies on the endpoint
    giving a blank node the same label in every query of the read.
    """
    wrapper = self._create_wrapper()
    config = self.config
    collapse = self.same_as == "collapse"
    components = SameAs(
        (binding["s"]["value"], binding["o"]["value"])
        for binding in (
            self._bindings(wrapper, config.same_as_query) if collapse else ()
        )
    )
    blank_triples = _BlankNodeTriples(
        lambda: self._bindings(wrapper, config.blank_node_query)
    )
    identity = _Identity(components, BlankNodeKeys(blank_triples.outgoing))

    def fill(doc: dict[str, Any], rows: Iterable[dict[str, Any]]) -> None:
        for binding in rows:
            _merge_sparql_binding_into_doc(
                doc, binding, identity, collapse=collapse
            )

    batch: list[dict] = []
    emitted = 0
    # Documents of sameAs components, by canonical IRI, and the members
    # the query returned for each.
    held: dict[str, dict[str, Any]] = {}
    returned: dict[str, set[str]] = {}

    def full() -> bool:
        return limit is not None and emitted >= limit

    def types_of(iris: list[str]) -> dict[str, set[str]]:
        canonical = {
            member: iri for iri in iris for member in components.members(iri)
        }
        members = sorted(canonical)
        found: dict[str, set[str]] = {}
        for start in range(0, len(members), _SUBJECTS_PER_QUERY):
            chunk = members[start : start + _SUBJECTS_PER_QUERY]
            for binding in self._bindings(
                wrapper,
                lambda offset, chunk=chunk: config.types_query(chunk, offset),
            ):
                if binding["o"]["type"] == "uri":
                    found.setdefault(canonical[binding["s"]["value"]], set()).add(
                        binding["o"]["value"]
                    )
        return found

    def out(docs: list[dict]) -> list[dict]:
        split_objects_by_type(docs, self.typed_objects, types_of)
        return docs

    rows_by_subject = groupby(
        self._bindings(wrapper, lambda offset: config.build_query(offset=offset)),
        key=lambda binding: (binding["s"]["type"], binding["s"]["value"]),
    )
    for _, rows in rows_by_subject:
        if full():
            break
        first = next(rows)
        subject = _binding_term(first["s"])
        if subject.kind == "bnode" and subject.value not in blank_triples:
            raise ValueError(
                f"{config.endpoint_url} returned a blank node under a label "
                "its blank-node triples do not carry. Blank nodes are keyed "
                "on their triples, which needs blank-node labels that stay "
                "the same across the queries of one read."
            )
        uri = identity.uri(subject)
        if subject.kind == "iri" and len(components.members(uri)) > 1:
            returned.setdefault(uri, set()).add(subject.value)
            doc = held.setdefault(uri, _new_doc(uri))
            fill(doc, [first, *rows])
            continue
        doc = _new_doc(uri)
        fill(doc, [first, *rows])
        batch.append(doc)
        emitted += 1
        if len(batch) >= batch_size:
            yield out(batch)
            batch = []

    for uri in sorted(held):
        if full():
            break
        members = components.members(uri)
        absent = [iri for iri in members if iri not in returned[uri]]
        for start in range(0, len(absent), _SUBJECTS_PER_QUERY):
            chunk = absent[start : start + _SUBJECTS_PER_QUERY]
            fill(
                held[uri],
                self._bindings(
                    wrapper,
                    lambda offset, chunk=chunk: config.subjects_query(
                        chunk, offset
                    ),
                ),
            )
        held[uri]["_same_as"] = [iri for iri in members if iri != uri]
        batch.append(held[uri])
        emitted += 1
        if len(batch) >= batch_size:
            yield out(batch)
            batch = []

    if batch:
        yield out(batch)

SparqlSourceConfig

Bases: ConfigBaseModel

Configuration for a SPARQL endpoint data source.

Attributes:

Name	Type	Description
endpoint_url	str	Full SPARQL query endpoint URL (e.g. http://localhost:3030/dataset/sparql)
rdf_class	str | None	URI of the rdf:Class whose instances to fetch
graph_uri	str | None	Named graph to restrict the query to (optional)
sparql_query	str | None	Custom SPARQL query override (optional)
username	str | None	HTTP basic-auth username (optional)
password	str | None	HTTP basic-auth password (optional)
page_size	int	Number of results per SPARQL LIMIT/OFFSET page

Source code in graflo/data_source/rdf.py

class SparqlSourceConfig(ConfigBaseModel):
    """Configuration for a SPARQL endpoint data source.

    Attributes:
        endpoint_url: Full SPARQL query endpoint URL
            (e.g. ``http://localhost:3030/dataset/sparql``)
        rdf_class: URI of the rdf:Class whose instances to fetch
        graph_uri: Named graph to restrict the query to (optional)
        sparql_query: Custom SPARQL query override (optional)
        username: HTTP basic-auth username (optional)
        password: HTTP basic-auth password (optional)
        page_size: Number of results per SPARQL LIMIT/OFFSET page
    """

    endpoint_url: str
    rdf_class: str | None = None
    graph_uri: str | None = None
    sparql_query: str | None = None
    username: str | None = None
    password: str | None = None
    page_size: int = Field(default=10_000, description="SPARQL pagination page size")

    def _select(self, pattern: str) -> str:
        """``SELECT ?s ?p ?o`` over *pattern*, inside the named graph when one is set."""
        if self.graph_uri:
            pattern = f"GRAPH <{self.graph_uri}> {{ {pattern} }}"
        return f"SELECT ?s ?p ?o WHERE {{ {pattern} }}"

    def _page(self, base: str, offset: int, limit: int | None = None) -> str:
        effective_limit = limit if limit is not None else self.page_size
        return f"{base} LIMIT {effective_limit} OFFSET {offset}"

    def build_query(self, offset: int = 0, limit: int | None = None) -> str:
        """Build a SPARQL SELECT query.

        If *sparql_query* is set it is returned with LIMIT/OFFSET appended.
        Otherwise generates::

            SELECT ?s ?p ?o WHERE { ?s a <rdf_class> . ?s ?p ?o . }
        """
        if self.sparql_query:
            base = self.sparql_query.rstrip().rstrip(";")
        else:
            class_filter = f"?s a <{self.rdf_class}> . " if self.rdf_class else ""
            base = self._select(f"{class_filter}?s ?p ?o .")

        # Group bindings by subject during streaming pagination; requires all
        # triple rows for one ?s to appear contiguously in the result.
        order_clause = "" if "ORDER BY" in base.upper() else " ORDER BY ?s"
        return self._page(f"{base}{order_clause}", offset, limit)

    def same_as_query(self, offset: int = 0) -> str:
        """The ``owl:sameAs`` statements between IRIs, as ``?s`` and ``?o``."""
        base = self._select(f"?s <{OWL_SAME_AS}> ?o . FILTER(isIRI(?s) && isIRI(?o))")
        return self._page(f"{base} ORDER BY ?s ?o", offset)

    def blank_node_query(self, offset: int = 0) -> str:
        """Every triple whose subject is a blank node."""
        base = self._select("?s ?p ?o . FILTER(isBlank(?s))")
        return self._page(f"{base} ORDER BY ?s ?p ?o", offset)

    def types_query(self, iris: list[str], offset: int = 0) -> str:
        """The ``rdf:type`` of each of *iris*, as ``?s`` and ``?o``."""
        values = " ".join(f"<{iri}>" for iri in iris)
        base = self._select(
            f"VALUES ?s {{ {values} }} ?s ?p ?o . FILTER(?p = <{RDF_TYPE}>)"
        )
        return self._page(f"{base} ORDER BY ?s ?o", offset)

    def subjects_query(self, iris: list[str], offset: int = 0) -> str:
        """Every triple of the subjects *iris*."""
        values = " ".join(f"<{iri}>" for iri in iris)
        base = self._select(f"VALUES ?s {{ {values} }} ?s ?p ?o .")
        return self._page(f"{base} ORDER BY ?s", offset)

Attributes

endpoint_url instance-attribute

graph_uri = None class-attribute instance-attribute

page_size = Field(default=10000, description='SPARQL pagination page size') class-attribute instance-attribute

password = None class-attribute instance-attribute

rdf_class = None class-attribute instance-attribute

sparql_query = None class-attribute instance-attribute

username = None class-attribute instance-attribute

Methods:

blank_node_query(offset=0)

Every triple whose subject is a blank node.

Source code in graflo/data_source/rdf.py

def blank_node_query(self, offset: int = 0) -> str:
    """Every triple whose subject is a blank node."""
    base = self._select("?s ?p ?o . FILTER(isBlank(?s))")
    return self._page(f"{base} ORDER BY ?s ?p ?o", offset)

build_query(offset=0, limit=None)

Build a SPARQL SELECT query.

If sparql_query is set it is returned with LIMIT/OFFSET appended. Otherwise generates::

SELECT ?s ?p ?o WHERE { ?s a <rdf_class> . ?s ?p ?o . }

Source code in graflo/data_source/rdf.py

def build_query(self, offset: int = 0, limit: int | None = None) -> str:
    """Build a SPARQL SELECT query.

    If *sparql_query* is set it is returned with LIMIT/OFFSET appended.
    Otherwise generates::

        SELECT ?s ?p ?o WHERE { ?s a <rdf_class> . ?s ?p ?o . }
    """
    if self.sparql_query:
        base = self.sparql_query.rstrip().rstrip(";")
    else:
        class_filter = f"?s a <{self.rdf_class}> . " if self.rdf_class else ""
        base = self._select(f"{class_filter}?s ?p ?o .")

    # Group bindings by subject during streaming pagination; requires all
    # triple rows for one ?s to appear contiguously in the result.
    order_clause = "" if "ORDER BY" in base.upper() else " ORDER BY ?s"
    return self._page(f"{base}{order_clause}", offset, limit)

same_as_query(offset=0)

The owl:sameAs statements between IRIs, as ?s and ?o.

Source code in graflo/data_source/rdf.py

def same_as_query(self, offset: int = 0) -> str:
    """The ``owl:sameAs`` statements between IRIs, as ``?s`` and ``?o``."""
    base = self._select(f"?s <{OWL_SAME_AS}> ?o . FILTER(isIRI(?s) && isIRI(?o))")
    return self._page(f"{base} ORDER BY ?s ?o", offset)

subjects_query(iris, offset=0)

Every triple of the subjects iris.

Source code in graflo/data_source/rdf.py

def subjects_query(self, iris: list[str], offset: int = 0) -> str:
    """Every triple of the subjects *iris*."""
    values = " ".join(f"<{iri}>" for iri in iris)
    base = self._select(f"VALUES ?s {{ {values} }} ?s ?p ?o .")
    return self._page(f"{base} ORDER BY ?s", offset)

types_query(iris, offset=0)

The rdf:type of each of iris, as ?s and ?o.

Source code in graflo/data_source/rdf.py

def types_query(self, iris: list[str], offset: int = 0) -> str:
    """The ``rdf:type`` of each of *iris*, as ``?s`` and ``?o``."""
    values = " ".join(f"<{iri}>" for iri in iris)
    base = self._select(
        f"VALUES ?s {{ {values} }} ?s ?p ?o . FILTER(?p = <{RDF_TYPE}>)"
    )
    return self._page(f"{base} ORDER BY ?s ?o", offset)

Functions:

split_objects_by_type(docs, properties, types_of)

Add <property>@<Class> beside each listed property, in place.

The new field holds the property's objects typed <Class> (the class IRI's local name); an object of several classes is under each, one of none under no field. types_of maps IRIs to their rdf:type IRIs.

Source code in graflo/data_source/rdf.py

def split_objects_by_type(
    docs: list[dict[str, Any]],
    properties: Iterable[str],
    types_of: Callable[[list[str]], dict[str, set[str]]],
) -> None:
    """Add ``<property>@<Class>`` beside each listed property, in place.

    The new field holds the property's objects typed ``<Class>`` (the class
    IRI's local name); an object of several classes is under each, one of none
    under no field. *types_of* maps IRIs to their ``rdf:type`` IRIs.
    """
    wanted = list(dict.fromkeys(properties))
    if not wanted:
        return
    objects = sorted(
        {
            value
            for doc in docs
            for name in wanted
            for value in _values(doc.get(name))
            if isinstance(value, str) and not value.startswith(BLANK_PREFIX)
        }
    )
    if not objects:
        return
    types = types_of(objects)
    for doc in docs:
        for name in wanted:
            for value in _values(doc.get(name)):
                for cls in sorted(types.get(value, ())):
                    _add_value(doc, f"{name}@{_local_name(cls)}", value)