Skip to content

# `` 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`

    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115

    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`

    411
    412

    def __init__(self):
        pass

### `` parse

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

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

    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43

    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`

    49
    50

    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`

    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88

    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`

    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148

    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`

    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170

    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)))