Source code for rdflib.plugins.sparql.results.jsonresults

"""A Serializer for SPARQL results in JSON:

Bits and pieces borrowed from:

Authors: Drew Perttula, Gunnar Aastrand Grimnes


from __future__ import annotations

import json
from typing import IO, Any, Dict, Mapping, MutableSequence, Optional

from rdflib.query import Result, ResultException, ResultParser, ResultSerializer
from rdflib.term import BNode, Identifier, Literal, URIRef, Variable

[docs]class JSONResultParser(ResultParser): # type error: Signature of "parse" incompatible with supertype "ResultParser"
[docs] def parse(self, source: IO, content_type: Optional[str] = None) -> Result: # type: ignore[override] inp = if isinstance(inp, bytes): inp = inp.decode("utf-8") return JSONResult(json.loads(inp))
[docs]class JSONResultSerializer(ResultSerializer):
[docs] def __init__(self, result: Result): ResultSerializer.__init__(self, result)
# type error: Signature of "serialize" incompatible with supertype "ResultSerializer"
[docs] 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 ] r = json.dumps(res, allow_nan=False, ensure_ascii=False) if encoding is not None: stream.write(r.encode(encoding)) else: stream.write(r)
def _bindingToJSON(self, b: Mapping[Variable, Identifier]) -> Dict[Variable, Any]: res = {} for var in b: j = termToJSON(self, b[var]) if j is not None: res[var] = termToJSON(self, b[var]) return res
[docs]class JSONResult(Result):
[docs] 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"]]
def _get_bindings(self) -> MutableSequence[Mapping[Variable, Identifier]]: ret: MutableSequence[Mapping[Variable, Identifier]] = [] for row in self.json["results"]["bindings"]: outRow: Dict[Variable, Identifier] = {} for k, v in row.items(): outRow[Variable(k)] = parseJsonTerm(v) ret.append(outRow) return ret
[docs]def parseJsonTerm(d: Dict[str, str]) -> Identifier: """rdflib object (Literal, URIRef, BNode) for the given json-format dict. input is like: { 'type': 'uri', 'value': '' } { '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)
[docs]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)))