rdflib

Backend for RDFLib.

For developers: The usage of s, p, and o represent the different parts of an RDF Triple: subject, predicate, and object.

RdflibStrategy

Triplestore strategy for rdflib.

Parameters:

Name	Type	Description	Default
base_iri	Optional[str]	Unused by this backend.	None
database	Optional[str]	Unused - rdflib does not support multiple databases.	None
triplestore_url	Optional[str]	If given, initialise the triplestore from this storage. When close() is called, the storage will be overwritten with the current content of the triplestore.	None
format	Optional[str]	Format of storage specified with base_iri.	None
graph	Graph	A rdflib.Graph instance to expose with tripper, instead of creating a new empty Graph object.	None

Source code in tripper/backends/rdflib.py

class RdflibStrategy:
    """Triplestore strategy for rdflib.

    Arguments:
        base_iri: Unused by this backend.
        database: Unused - rdflib does not support multiple databases.
        triplestore_url: If given, initialise the triplestore from this
            storage.  When `close()` is called, the storage will be
            overwritten with the current content of the triplestore.
        format: Format of storage specified with `base_iri`.
        graph: A rdflib.Graph instance to expose with tripper, instead of
            creating a new empty Graph object.
    """

    def __init__(
        self,
        base_iri: "Optional[str]" = None,  # pylint: disable=unused-argument
        database: "Optional[str]" = None,
        triplestore_url: "Optional[str]" = None,
        format: "Optional[str]" = None,  # pylint: disable=redefined-builtin
        graph: "Graph" = None,
    ) -> None:
        # Note that although `base_iri` is unused in this backend, it may
        # still be used by calling Triplestore object.
        if database:
            warnings.warn("database", UnusedArgumentWarning, stacklevel=3)

        self.graph = graph if graph else Graph()
        self.triplestore_url = triplestore_url
        if self.triplestore_url is not None:
            if format is None:
                format = guess_format(self.triplestore_url)
            self.parse(location=self.triplestore_url, format=format)
        self.base_format = format

    def triples(self, triple: "Triple") -> "Generator[Triple, None, None]":
        """Returns a generator over matching triples."""
        for s, p, o in self.graph.triples(  # pylint: disable=not-an-iterable
            astriple(triple)
        ):
            yield (
                (
                    f"_:{s}"
                    if isinstance(s, BNode) and not s.startswith("_:")
                    else str(s)
                ),
                str(p),
                (
                    parse_literal(o)
                    if isinstance(o, rdflibLiteral)
                    else (
                        f"_:{o}"
                        if isinstance(o, BNode) and not o.startswith("_:")
                        else str(o)
                    )
                ),
            )

    def add_triples(self, triples: "Sequence[Triple]"):
        """Add a sequence of triples."""
        for triple in triples:
            self.graph.add(astriple(triple))

    def remove(self, triple: "Triple"):
        """Remove all matching triples from the backend."""
        self.graph.remove(astriple(triple))

    # Optional methods
    def close(self):
        """Close the internal RDFLib graph."""
        if self.triplestore_url:
            self.serialize(
                destination=self.triplestore_url, format=self.base_format
            )
        self.graph.close()

    def parse(
        self,
        source=None,
        location=None,
        data=None,
        format=None,  # pylint: disable=redefined-builtin
        **kwargs,
    ):
        """Parse source and add the resulting triples to triplestore.

        The source is specified using one of `source`, `location` or `data`.

        Parameters:
            source: File-like object or file name.
            location: String with relative or absolute URL to source.
            data: String containing the data to be parsed.
            format: Needed if format can not be inferred from source.
            kwargs: Additional less used keyword arguments.
                See https://rdflib.readthedocs.io/en/stable/apidocs/rdflib.html#rdflib.Graph.parse
        """
        self.graph.parse(
            source=source,
            location=location,
            data=data,
            format=format,
            **kwargs,
        )

    def serialize(
        self,
        destination=None,
        format="turtle",  # pylint: disable=redefined-builtin
        **kwargs,
    ) -> "Union[None, str]":
        """Serialise to destination.

        Parameters:
            destination: File name or object to write to. If None, the serialisation is
                returned.
            format: Format to serialise as. Supported formats, depends on the backend.
            kwargs: Passed to the rdflib.Graph.serialize() method.
                See https://rdflib.readthedocs.io/en/stable/apidocs/rdflib.html#rdflib.Graph.serialize

        Returns:
            Serialised string if `destination` is None.
        """
        result = self.graph.serialize(
            destination=destination, format=format, **kwargs
        )
        if destination is None:
            # Depending on the version of rdflib the return value of
            # graph.serialize() man either be a string or a bytes object...
            return result if isinstance(result, str) else result.decode()
        return None

    def query(self, query_object, **kwargs) -> "List[Tuple[str, ...]]":
        """SPARQL query.

        Parameters:
            query_object: String with the SPARQL query.
            kwargs: Keyword arguments passed to rdflib.Graph.query().

        Returns:
            List of tuples of IRIs for each matching row.

        """
        rows = self.graph.query(query_object=query_object, **kwargs)
        return [tuple(str(v) for v in row) for row in rows]

    def update(self, update_object, **kwargs) -> None:
        """Update triplestore with SPARQL.

        Parameters:
            update_object: String with the SPARQL query.
            kwargs: Keyword arguments passed to rdflib.Graph.update().

        Note:
            This method is intended for INSERT and DELETE queries. Use
            the query() method for SELECT queries.

        """
        return self.graph.update(update_object=update_object, **kwargs)

    def bind(self, prefix: str, namespace: str):
        """Bind prefix to namespace.

        Should only be defined if the backend supports namespaces.
        Called by triplestore.bind().
        """
        if namespace:
            self.graph.bind(prefix, namespace, replace=True)
        else:
            warnings.warn(
                "rdflib does not support removing namespace prefixes"
            )

    def namespaces(self) -> dict:
        """Returns a dict mapping prefixes to namespaces.

        Should only be defined if the backend supports namespaces.
        Used by triplestore.parse() to get prefixes after reading
        triples from an external source.
        """
        return {
            prefix: str(namespace)
            for prefix, namespace in self.graph.namespaces()
        }

add_triples(self, triples)

Add a sequence of triples.

Source code in tripper/backends/rdflib.py

def add_triples(self, triples: "Sequence[Triple]"):
    """Add a sequence of triples."""
    for triple in triples:
        self.graph.add(astriple(triple))

bind(self, prefix, namespace)

Bind prefix to namespace.

Should only be defined if the backend supports namespaces. Called by triplestore.bind().

Source code in tripper/backends/rdflib.py

def bind(self, prefix: str, namespace: str):
    """Bind prefix to namespace.

    Should only be defined if the backend supports namespaces.
    Called by triplestore.bind().
    """
    if namespace:
        self.graph.bind(prefix, namespace, replace=True)
    else:
        warnings.warn(
            "rdflib does not support removing namespace prefixes"
        )

close(self)

Close the internal RDFLib graph.

Source code in tripper/backends/rdflib.py

def close(self):
    """Close the internal RDFLib graph."""
    if self.triplestore_url:
        self.serialize(
            destination=self.triplestore_url, format=self.base_format
        )
    self.graph.close()

namespaces(self)

Returns a dict mapping prefixes to namespaces.

Should only be defined if the backend supports namespaces. Used by triplestore.parse() to get prefixes after reading triples from an external source.

Source code in tripper/backends/rdflib.py

def namespaces(self) -> dict:
    """Returns a dict mapping prefixes to namespaces.

    Should only be defined if the backend supports namespaces.
    Used by triplestore.parse() to get prefixes after reading
    triples from an external source.
    """
    return {
        prefix: str(namespace)
        for prefix, namespace in self.graph.namespaces()
    }

parse(self, source=None, location=None, data=None, format=None, **kwargs)

Parse source and add the resulting triples to triplestore.

The source is specified using one of source, location or data.

Parameters:

Name	Type	Description	Default
source		File-like object or file name.	None
location		String with relative or absolute URL to source.	None
data		String containing the data to be parsed.	None
format		Needed if format can not be inferred from source.	None
kwargs		Additional less used keyword arguments. See https://rdflib.readthedocs.io/en/stable/apidocs/rdflib.html#rdflib.Graph.parse	{}

Source code in tripper/backends/rdflib.py

def parse(
    self,
    source=None,
    location=None,
    data=None,
    format=None,  # pylint: disable=redefined-builtin
    **kwargs,
):
    """Parse source and add the resulting triples to triplestore.

    The source is specified using one of `source`, `location` or `data`.

    Parameters:
        source: File-like object or file name.
        location: String with relative or absolute URL to source.
        data: String containing the data to be parsed.
        format: Needed if format can not be inferred from source.
        kwargs: Additional less used keyword arguments.
            See https://rdflib.readthedocs.io/en/stable/apidocs/rdflib.html#rdflib.Graph.parse
    """
    self.graph.parse(
        source=source,
        location=location,
        data=data,
        format=format,
        **kwargs,
    )

query(self, query_object, **kwargs)

SPARQL query.

Parameters:

Name	Type	Description	Default
query_object		String with the SPARQL query.	required
kwargs		Keyword arguments passed to rdflib.Graph.query().	{}

Returns:

Type	Description
List[Tuple[str, ...]]	List of tuples of IRIs for each matching row.

Source code in tripper/backends/rdflib.py

def query(self, query_object, **kwargs) -> "List[Tuple[str, ...]]":
    """SPARQL query.

    Parameters:
        query_object: String with the SPARQL query.
        kwargs: Keyword arguments passed to rdflib.Graph.query().

    Returns:
        List of tuples of IRIs for each matching row.

    """
    rows = self.graph.query(query_object=query_object, **kwargs)
    return [tuple(str(v) for v in row) for row in rows]

remove(self, triple)

Remove all matching triples from the backend.

Source code in tripper/backends/rdflib.py

def remove(self, triple: "Triple"):
    """Remove all matching triples from the backend."""
    self.graph.remove(astriple(triple))

serialize(self, destination=None, format='turtle', **kwargs)

Serialise to destination.

Parameters:

Name	Type	Description	Default
destination		File name or object to write to. If None, the serialisation is returned.	None
format		Format to serialise as. Supported formats, depends on the backend.	'turtle'
kwargs		Passed to the rdflib.Graph.serialize() method. See https://rdflib.readthedocs.io/en/stable/apidocs/rdflib.html#rdflib.Graph.serialize	{}

Returns:

Type	Description
Union[None, str]	Serialised string if destination is None.

Source code in tripper/backends/rdflib.py

def serialize(
    self,
    destination=None,
    format="turtle",  # pylint: disable=redefined-builtin
    **kwargs,
) -> "Union[None, str]":
    """Serialise to destination.

    Parameters:
        destination: File name or object to write to. If None, the serialisation is
            returned.
        format: Format to serialise as. Supported formats, depends on the backend.
        kwargs: Passed to the rdflib.Graph.serialize() method.
            See https://rdflib.readthedocs.io/en/stable/apidocs/rdflib.html#rdflib.Graph.serialize

    Returns:
        Serialised string if `destination` is None.
    """
    result = self.graph.serialize(
        destination=destination, format=format, **kwargs
    )
    if destination is None:
        # Depending on the version of rdflib the return value of
        # graph.serialize() man either be a string or a bytes object...
        return result if isinstance(result, str) else result.decode()
    return None

triples(self, triple)

Returns a generator over matching triples.

Source code in tripper/backends/rdflib.py

def triples(self, triple: "Triple") -> "Generator[Triple, None, None]":
    """Returns a generator over matching triples."""
    for s, p, o in self.graph.triples(  # pylint: disable=not-an-iterable
        astriple(triple)
    ):
        yield (
            (
                f"_:{s}"
                if isinstance(s, BNode) and not s.startswith("_:")
                else str(s)
            ),
            str(p),
            (
                parse_literal(o)
                if isinstance(o, rdflibLiteral)
                else (
                    f"_:{o}"
                    if isinstance(o, BNode) and not o.startswith("_:")
                    else str(o)
                )
            ),
        )

update(self, update_object, **kwargs)

Update triplestore with SPARQL.

Parameters:

Name	Type	Description	Default
update_object		String with the SPARQL query.	required
kwargs		Keyword arguments passed to rdflib.Graph.update().	{}

Note

This method is intended for INSERT and DELETE queries. Use the query() method for SELECT queries.

Source code in tripper/backends/rdflib.py

def update(self, update_object, **kwargs) -> None:
    """Update triplestore with SPARQL.

    Parameters:
        update_object: String with the SPARQL query.
        kwargs: Keyword arguments passed to rdflib.Graph.update().

    Note:
        This method is intended for INSERT and DELETE queries. Use
        the query() method for SELECT queries.

    """
    return self.graph.update(update_object=update_object, **kwargs)

astriple(triple)

Help function converting a triple to rdflib triple.

Source code in tripper/backends/rdflib.py

def astriple(triple: "Triple"):
    """Help function converting a triple to rdflib triple."""
    s, p, o = triple
    return asuri(s), asuri(p), asuri(o)

asuri(value)

Help function converting a spo-value to proper rdflib type.

Source code in tripper/backends/rdflib.py

def asuri(value: "Union[None, Literal, str]"):
    """Help function converting a spo-value to proper rdflib type."""
    if value is None:
        return None
    if isinstance(value, Literal):
        return rdflibLiteral(
            value.value, lang=value.lang, datatype=value.datatype
        )
    if value.startswith("_:"):
        return BNode(value[2:])
    return URIRef(value)