jsonresults

A Serializer for SPARQL results in JSON:

http://www.w3.org/TR/rdf-sparql-json-res/

Bits and pieces borrowed from: http://projects.bigasterisk.com/sparqlhttp/

Authors: Drew Perttula, Gunnar Aastrand Grimnes

Classes:

• JSONResult –
• JSONResultParser –

  Parses SPARQL JSON results into a Result object.

• JSONResultSerializer –

  Serializes SPARQL results to JSON format.

Functions:

• parseJsonTerm –

  rdflib object (Literal, URIRef, BNode) for the given json-format dict.

• termToJSON –

JSONResult

JSONResult(json: Dict[str, Any])

Bases: Result

Attributes:

• askAnswer –
• bindings –
• json –
• vars –

Source code in rdflib/plugins/sparql/results/jsonresults.py

def __init__(self, json: Dict[str, Any]):
    self.json = json
    if "boolean" in json:
        type_ = "ASK"
    elif "results" in json:
        type_ = "SELECT"
    else:
        raise ResultException("No boolean or results in json!")

    Result.__init__(self, type_)

    if type_ == "ASK":
        self.askAnswer = bool(json["boolean"])
    else:
        self.bindings = self._get_bindings()
        self.vars = [Variable(x) for x in json["head"]["vars"]]

askAnswer instance-attribute

askAnswer = bool(json['boolean'])

bindings instance-attribute

bindings = _get_bindings()

json instance-attribute

json = json

vars instance-attribute

vars = [(Variable(x)) for x in (json['head']['vars'])]

JSONResultParser

JSONResultParser()

Bases: ResultParser

Parses SPARQL JSON results into a Result object.

Methods:

• parse –

Source code in rdflib/query.py

def __init__(self):
    pass

parse

parse(source: IO, content_type: Optional[str] = None) -> Result

Source code in rdflib/plugins/sparql/results/jsonresults.py

def parse(self, source: IO, content_type: Optional[str] = None) -> Result:  # type: ignore[override]
    inp = source.read()
    if _HAS_ORJSON:
        try:
            loaded = orjson.loads(inp)
        except Exception as e:
            raise ResultException(f"Failed to parse result: {e}")
    else:
        if isinstance(inp, bytes):
            inp = inp.decode("utf-8")
        loaded = json.loads(inp)
    return JSONResult(loaded)

JSONResultSerializer

JSONResultSerializer(result: Result)

Bases: ResultSerializer

Serializes SPARQL results to JSON format.

Methods:

• serialize –

Source code in rdflib/plugins/sparql/results/jsonresults.py

def __init__(self, result: Result):
    ResultSerializer.__init__(self, result)

serialize

serialize(stream: IO, encoding: str = None) -> None

Source code in rdflib/plugins/sparql/results/jsonresults.py

def serialize(self, stream: IO, encoding: str = None) -> None:  # type: ignore[override]
    res: Dict[str, Any] = {}
    if self.result.type == "ASK":
        res["head"] = {}
        res["boolean"] = self.result.askAnswer
    else:
        # select
        res["results"] = {}
        res["head"] = {}
        res["head"]["vars"] = self.result.vars
        res["results"]["bindings"] = [
            self._bindingToJSON(x) for x in self.result.bindings
        ]
    if _HAS_ORJSON:
        try:
            r_bytes = orjson.dumps(res, option=orjson.OPT_NON_STR_KEYS)
        except Exception as e:
            raise ResultException(f"Failed to serialize result: {e}")
        if encoding is not None:
            # Note, orjson will always write utf-8 even if
            # encoding is specified as something else.
            try:
                stream.write(r_bytes)
            except (TypeError, ValueError):
                stream.write(r_bytes.decode("utf-8"))
        else:
            stream.write(r_bytes.decode("utf-8"))
    else:
        r_str = json.dumps(res, allow_nan=False, ensure_ascii=False)
        if encoding is not None:
            try:
                stream.write(r_str.encode(encoding))
            except (TypeError, ValueError):
                stream.write(r_str)
        else:
            stream.write(r_str)

parseJsonTerm

parseJsonTerm(d: Dict[str, str]) -> Identifier

rdflib object (Literal, URIRef, BNode) for the given json-format dict.

input is like:

{ 'type': 'uri', 'value': 'http://famegame.com/2006/01/username' }
{ 'type': 'literal', 'value': 'drewp' }

Source code in rdflib/plugins/sparql/results/jsonresults.py

def parseJsonTerm(d: Dict[str, str]) -> Identifier:
    """rdflib object (Literal, URIRef, BNode) for the given json-format dict.

    input is like:

    ```json
    { 'type': 'uri', 'value': 'http://famegame.com/2006/01/username' }
    { 'type': 'literal', 'value': 'drewp' }
    ```
    """

    t = d["type"]
    if t == "uri":
        return URIRef(d["value"])
    elif t == "literal":
        return Literal(d["value"], datatype=d.get("datatype"), lang=d.get("xml:lang"))
    elif t == "typed-literal":
        return Literal(d["value"], datatype=URIRef(d["datatype"]))
    elif t == "bnode":
        return BNode(d["value"])
    else:
        raise NotImplementedError("json term type %r" % t)

termToJSON

termToJSON(self: JSONResultSerializer, term: Optional[Identifier]) -> Optional[Dict[str, str]]

Source code in rdflib/plugins/sparql/results/jsonresults.py

def termToJSON(
    self: JSONResultSerializer, term: Optional[Identifier]
) -> Optional[Dict[str, str]]:
    if isinstance(term, URIRef):
        return {"type": "uri", "value": str(term)}
    elif isinstance(term, Literal):
        r = {"type": "literal", "value": str(term)}

        if term.datatype is not None:
            r["datatype"] = str(term.datatype)
        if term.language is not None:
            r["xml:lang"] = term.language
        return r

    elif isinstance(term, BNode):
        return {"type": "bnode", "value": str(term)}
    elif term is None:
        return None
    else:
        raise ResultException("Unknown term type: %s (%s)" % (term, type(term)))