sparqlwrapper

Backend for SPARQLWrapper

SparqlwrapperStrategy

Triplestore strategy for SPARQLWrapper.

Parameters:

Name	Type	Description	Default
base_iri	str	URI of SPARQL endpoint.	required
kwargs		Additional arguments passed to the SPARQLWrapper constructor.	{}

Source code in tripper/backends/sparqlwrapper.py

class SparqlwrapperStrategy:
    """Triplestore strategy for SPARQLWrapper.

    Arguments:
        base_iri: URI of SPARQL endpoint.
        kwargs: Additional arguments passed to the SPARQLWrapper constructor.

    """

    prefer_sparql = True

    def __init__(self, base_iri: str, **kwargs) -> None:
        kwargs.pop(
            "database", None
        )  # database is not used in the SPARQLWrapper backend
        self.sparql = SPARQLWrapper(endpoint=base_iri, **kwargs)

    def triples(self, triple: "Triple") -> "Generator[Triple, None, None]":
        """Returns a generator over matching triples."""
        variables = [
            f"?{triple_name}"
            for triple_name, triple_value in zip("spo", triple)
            if triple_value is None
        ]
        where_spec = " ".join(
            (
                f"?{triple_name}"
                if triple_value is None
                else (
                    triple_value
                    if triple_value.startswith("<")
                    else f"<{triple_value}>"
                )
            )
            for triple_name, triple_value in zip("spo", triple)
        )
        query = "\n".join(
            [
                f"SELECT {' '.join(variables)} WHERE {{",
                f"  {where_spec} .",
                "}",
            ]
        )
        self.sparql.setReturnFormat(JSON)
        self.sparql.setMethod(GET)
        self.sparql.setQuery(query)
        ret = self.sparql.queryAndConvert()
        for binding in ret["results"]["bindings"]:
            yield tuple(
                (
                    convert_json_entrydict(binding[name])
                    if name in binding
                    else value
                )
                for name, value in zip("spo", triple)
            )

    def add_triples(self, triples: "Sequence[Triple]") -> "QueryResult":
        """Add a sequence of triples."""
        spec = "\n".join(
            "  "
            + " ".join(
                (
                    value.n3()
                    if isinstance(value, Literal)
                    else value if value.startswith("<") else f"<{value}>"
                )
                for value in triple
            )
            + " ."
            for triple in triples
        )
        query = f"INSERT DATA {{\n{spec}\n}}"
        self.sparql.setReturnFormat(RDFXML)
        self.sparql.setMethod(POST)
        self.sparql.setQuery(query)
        return self.sparql.query()

    def remove(self, triple: "Triple") -> "QueryResult":
        """Remove all matching triples from the backend."""
        spec = " ".join(
            (
                f"?{name}"
                if value is None
                else (
                    value.n3()
                    if isinstance(value, Literal)
                    else value if value.startswith("<") else f"<{value}>"
                )
            )
            for name, value in zip("spo", triple)
        )
        query = f"DELETE {{ {spec} }}"
        self.sparql.setReturnFormat(RDFXML)
        self.sparql.setMethod(POST)
        self.sparql.setQuery(query)
        return self.sparql.query()

add_triples(self, triples)

Add a sequence of triples.

Source code in tripper/backends/sparqlwrapper.py

def add_triples(self, triples: "Sequence[Triple]") -> "QueryResult":
    """Add a sequence of triples."""
    spec = "\n".join(
        "  "
        + " ".join(
            (
                value.n3()
                if isinstance(value, Literal)
                else value if value.startswith("<") else f"<{value}>"
            )
            for value in triple
        )
        + " ."
        for triple in triples
    )
    query = f"INSERT DATA {{\n{spec}\n}}"
    self.sparql.setReturnFormat(RDFXML)
    self.sparql.setMethod(POST)
    self.sparql.setQuery(query)
    return self.sparql.query()

remove(self, triple)

Remove all matching triples from the backend.

Source code in tripper/backends/sparqlwrapper.py

def remove(self, triple: "Triple") -> "QueryResult":
    """Remove all matching triples from the backend."""
    spec = " ".join(
        (
            f"?{name}"
            if value is None
            else (
                value.n3()
                if isinstance(value, Literal)
                else value if value.startswith("<") else f"<{value}>"
            )
        )
        for name, value in zip("spo", triple)
    )
    query = f"DELETE {{ {spec} }}"
    self.sparql.setReturnFormat(RDFXML)
    self.sparql.setMethod(POST)
    self.sparql.setQuery(query)
    return self.sparql.query()

triples(self, triple)

Returns a generator over matching triples.

Source code in tripper/backends/sparqlwrapper.py

def triples(self, triple: "Triple") -> "Generator[Triple, None, None]":
    """Returns a generator over matching triples."""
    variables = [
        f"?{triple_name}"
        for triple_name, triple_value in zip("spo", triple)
        if triple_value is None
    ]
    where_spec = " ".join(
        (
            f"?{triple_name}"
            if triple_value is None
            else (
                triple_value
                if triple_value.startswith("<")
                else f"<{triple_value}>"
            )
        )
        for triple_name, triple_value in zip("spo", triple)
    )
    query = "\n".join(
        [
            f"SELECT {' '.join(variables)} WHERE {{",
            f"  {where_spec} .",
            "}",
        ]
    )
    self.sparql.setReturnFormat(JSON)
    self.sparql.setMethod(GET)
    self.sparql.setQuery(query)
    ret = self.sparql.queryAndConvert()
    for binding in ret["results"]["bindings"]:
        yield tuple(
            (
                convert_json_entrydict(binding[name])
                if name in binding
                else value
            )
            for name, value in zip("spo", triple)
        )

convert_json_entrydict(entrydict)

Convert SPARQLWrapper json entry dict (representing a single IRI or literal) to a tripper type.

Source code in tripper/backends/sparqlwrapper.py

def convert_json_entrydict(entrydict: "Dict[str, str]") -> str:
    """Convert SPARQLWrapper json entry dict (representing a single IRI or
    literal) to a tripper type."""
    if entrydict["type"] == "uri":
        return entrydict["value"]

    if entrydict["type"] == "literal":
        return Literal(
            entrydict["value"],
            lang=entrydict.get("xml:lang"),
            datatype=entrydict.get("datatype"),
        )

    if entrydict["type"] == "bnode":
        return (
            entrydict["value"]
            if entrydict["value"].startswith("_:")
            else f"_:{entrydict['value']}"
        )

    raise ValueError(f"unexpected type in entrydict: {entrydict}")