Skip to content

# `` xmlresults

A Parser for SPARQL results in XML:

http://www.w3.org/TR/rdf-sparql-XMLres/

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

Authors: Drew Perttula, Gunnar Aastrand Grimnes

Classes:

- `SPARQLXMLWriter` –

  Python saxutils-based SPARQL XML Writer

- `XMLResult` –
- `XMLResultParser` –

  A Parser for SPARQL results in XML.

- `XMLResultSerializer` –

  Serializes SPARQL results into XML format.

Functions:

- `parseTerm` –

  rdflib object (Literal, URIRef, BNode) for the given

Attributes:

- `FOUND_LXML` –
- `RESULTS_NS_ET` –
- `SPARQL_XML_NAMESPACE` –
- `log` –

## `` FOUND_LXML `module-attribute`

    FOUND_LXML = True

## `` RESULTS_NS_ET `module-attribute`

    RESULTS_NS_ET = '{%s}' % SPARQL_XML_NAMESPACE

## `` SPARQL_XML_NAMESPACE `module-attribute`

    SPARQL_XML_NAMESPACE = 'http://www.w3.org/2005/sparql-results#'

## `` log `module-attribute`

    log = getLogger(__name__)

## `` SPARQLXMLWriter

    SPARQLXMLWriter(output: IO, encoding: str = 'utf-8')

Python saxutils-based SPARQL XML Writer

Methods:

- `close` –
- `write_ask` –
- `write_binding` –
- `write_end_result` –
- `write_header` –
- `write_results_header` –
- `write_start_result` –

Attributes:

- `writer` –

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

    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200

    def __init__(self, output: IO, encoding: str = "utf-8"):
        writer = XMLGenerator(output, encoding)
        writer.startDocument()
        writer.startPrefixMapping("", SPARQL_XML_NAMESPACE)
        writer.startPrefixMapping("xml", XML_NAMESPACE)
        writer.startElementNS(
            (SPARQL_XML_NAMESPACE, "sparql"), "sparql", AttributesNSImpl({}, {})
        )
        self.writer = writer
        self._output = output
        self._encoding = encoding
        self._results = False

### `` writer `instance-attribute`

    writer = writer

### `` close

    close() -> None

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

    301
    302
    303
    304
    305

    def close(self) -> None:
        if self._results:
            self.writer.endElementNS((SPARQL_XML_NAMESPACE, "results"), "results")
        self.writer.endElementNS((SPARQL_XML_NAMESPACE, "sparql"), "sparql")
        self.writer.endDocument()

### `` write_ask

    write_ask(val: bool) -> None

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

    223
    224
    225
    226
    227
    228

    def write_ask(self, val: bool) -> None:
        self.writer.startElementNS(
            (SPARQL_XML_NAMESPACE, "boolean"), "boolean", AttributesNSImpl({}, {})
        )
        self.writer.characters(str(val).lower())
        self.writer.endElementNS((SPARQL_XML_NAMESPACE, "boolean"), "boolean")

### `` write_binding

    write_binding(name: Variable, val: Identifier) -> None

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

    247
    248
    249
    250
    251
    252
    253
    254
    255
    256
    257
    258
    259
    260
    261
    262
    263
    264
    265
    266
    267
    268
    269
    270
    271
    272
    273
    274
    275
    276
    277
    278
    279
    280
    281
    282
    283
    284
    285
    286
    287
    288
    289
    290
    291
    292
    293
    294
    295
    296
    297
    298
    299

    def write_binding(self, name: Variable, val: Identifier) -> None:
        assert self._resultStarted

        attr_vals: Dict[Tuple[Optional[str], str], str] = {
            (None, "name"): str(name),
        }
        attr_qnames: Dict[Tuple[Optional[str], str], str] = {
            (None, "name"): "name",
        }
        self.writer.startElementNS(
            (SPARQL_XML_NAMESPACE, "binding"),
            "binding",
            # type error: Argument 1 to "AttributesNSImpl" has incompatible type "Dict[Tuple[None, str], str]"; expected "Mapping[Tuple[str, str], str]"
            # type error: Argument 2 to "AttributesNSImpl" has incompatible type "Dict[Tuple[None, str], str]"; expected "Mapping[Tuple[str, str], str]"
            AttributesNSImpl(attr_vals, attr_qnames),  # type: ignore[arg-type, unused-ignore]
        )

        if isinstance(val, URIRef):
            self.writer.startElementNS(
                (SPARQL_XML_NAMESPACE, "uri"), "uri", AttributesNSImpl({}, {})
            )
            self.writer.characters(val)
            self.writer.endElementNS((SPARQL_XML_NAMESPACE, "uri"), "uri")
        elif isinstance(val, BNode):
            self.writer.startElementNS(
                (SPARQL_XML_NAMESPACE, "bnode"), "bnode", AttributesNSImpl({}, {})
            )
            self.writer.characters(val)
            self.writer.endElementNS((SPARQL_XML_NAMESPACE, "bnode"), "bnode")
        elif isinstance(val, Literal):
            attr_vals = {}
            attr_qnames = {}
            if val.language:
                attr_vals[(XML_NAMESPACE, "lang")] = val.language
                attr_qnames[(XML_NAMESPACE, "lang")] = "xml:lang"
            elif val.datatype:
                attr_vals[(None, "datatype")] = val.datatype
                attr_qnames[(None, "datatype")] = "datatype"

            self.writer.startElementNS(
                (SPARQL_XML_NAMESPACE, "literal"),
                "literal",
                # type error: Argument 1 to "AttributesNSImpl" has incompatible type "Dict[Tuple[Optional[str], str], str]"; expected "Mapping[Tuple[str, str], str]"
                # type error: Argument 2 to "AttributesNSImpl" has incompatible type "Dict[Tuple[Optional[str], str], str]"; expected "Mapping[Tuple[str, str], str]"
                AttributesNSImpl(attr_vals, attr_qnames),  # type: ignore[arg-type, unused-ignore]
            )
            self.writer.characters(val)
            self.writer.endElementNS((SPARQL_XML_NAMESPACE, "literal"), "literal")

        else:
            raise Exception("Unsupported RDF term: %s" % val)

        self.writer.endElementNS((SPARQL_XML_NAMESPACE, "binding"), "binding")

### `` write_end_result

    write_end_result() -> None

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

    242
    243
    244
    245

    def write_end_result(self) -> None:
        assert self._resultStarted
        self.writer.endElementNS((SPARQL_XML_NAMESPACE, "result"), "result")
        self._resultStarted = False

### `` write_header

    write_header(allvarsL: Sequence[Variable]) -> None

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

    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219
    220
    221

    def write_header(self, allvarsL: Sequence[Variable]) -> None:
        self.writer.startElementNS(
            (SPARQL_XML_NAMESPACE, "head"), "head", AttributesNSImpl({}, {})
        )
        for i in range(0, len(allvarsL)):
            attr_vals = {
                (None, "name"): str(allvarsL[i]),
            }
            attr_qnames = {
                (None, "name"): "name",
            }
            self.writer.startElementNS(
                (SPARQL_XML_NAMESPACE, "variable"),
                "variable",
                # type error: Argument 1 to "AttributesNSImpl" has incompatible type "Dict[Tuple[None, str], str]"; expected "Mapping[Tuple[str, str], str]"
                # type error: Argument 2 to "AttributesNSImpl" has incompatible type "Dict[Tuple[None, str], str]"; expected "Mapping[Tuple[str, str], str]"  [arg-type]
                AttributesNSImpl(attr_vals, attr_qnames),  # type: ignore[arg-type]
            )
            self.writer.endElementNS((SPARQL_XML_NAMESPACE, "variable"), "variable")
        self.writer.endElementNS((SPARQL_XML_NAMESPACE, "head"), "head")

### `` write_results_header

    write_results_header() -> None

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

    230
    231
    232
    233
    234

    def write_results_header(self) -> None:
        self.writer.startElementNS(
            (SPARQL_XML_NAMESPACE, "results"), "results", AttributesNSImpl({}, {})
        )
        self._results = True

### `` write_start_result

    write_start_result() -> None

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

    236
    237
    238
    239
    240

    def write_start_result(self) -> None:
        self.writer.startElementNS(
            (SPARQL_XML_NAMESPACE, "result"), "result", AttributesNSImpl({}, {})
        )
        self._resultStarted = True

## `` XMLResult

    XMLResult(source: IO, content_type: Optional[str] = None)

Bases: `Result`

Attributes:

- `askAnswer` –
- `bindings` –
- `vars` –

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

     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
     89
     90
     91
     92
     93
     94
     95
     96
     97
     98
     99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127

    def __init__(self, source: IO, content_type: Optional[str] = None):
        parser_encoding: Optional[str] = None
        if hasattr(source, "encoding"):
            if TYPE_CHECKING:
                assert isinstance(source, TextIO)
            parser_encoding = "utf-8"
            source_str = source.read()
            source = BytesIO(source_str.encode(parser_encoding))
        else:
            if TYPE_CHECKING:
                assert isinstance(source, BinaryIO)

        if FOUND_LXML:
            lxml_parser = lxml_etree.XMLParser(huge_tree=True, encoding=parser_encoding)
            tree = cast(
                xml_etree.ElementTree,
                lxml_etree.parse(source, parser=lxml_parser),
            )
        else:
            xml_parser = xml_etree.XMLParser(encoding=parser_encoding)
            tree = xml_etree.parse(source, parser=xml_parser)

        boolean = tree.find(RESULTS_NS_ET + "boolean")
        results = tree.find(RESULTS_NS_ET + "results")

        if boolean is not None:
            type_ = "ASK"
        elif results is not None:
            type_ = "SELECT"
        else:
            raise ResultException("No RDF result-bindings or boolean answer found!")

        Result.__init__(self, type_)

        if type_ == "SELECT":
            self.bindings = []
            for result in results:  # type: ignore[union-attr]
                if result.tag != f"{RESULTS_NS_ET}result":
                    # This is here because with lxml this also gets comments,
                    # not just elements. Also this should not operate on non
                    # "result" elements.
                    continue
                r = {}
                for binding in result:
                    if binding.tag != f"{RESULTS_NS_ET}binding":
                        # This is here because with lxml this also gets
                        # comments, not just elements. Also this should not
                        # operate on non "binding" elements.
                        continue
                    # type error: error: Argument 1 to "Variable" has incompatible type "Union[str, None, Any]"; expected "str"
                    # NOTE on type error: Element.get() can return None, and
                    # this will invariably fail if passed into Variable
                    # constructor as value
                    r[Variable(binding.get("name"))] = parseTerm(binding[0])  # type: ignore[arg-type] # FIXME
                self.bindings.append(r)

            self.vars = [
                # type error: Argument 1 to "Variable" has incompatible type "Optional[str]"; expected "str"
                # NOTE on type error: Element.get() can return None, and this
                # will invariably fail if passed into Variable constructor as
                # value
                Variable(x.get("name"))  # type: ignore[arg-type] # FIXME
                for x in tree.findall(
                    "./%shead/%svariable" % (RESULTS_NS_ET, RESULTS_NS_ET)
                )
            ]

        else:
            self.askAnswer = boolean.text.lower().strip() == "true"  # type: ignore[union-attr]

### `` askAnswer `instance-attribute`

    askAnswer = strip() == 'true'

### `` bindings `instance-attribute`

    bindings = []

### `` vars `instance-attribute`

    vars = [(Variable(get('name'))) for x in (findall('./%shead/%svariable' % (RESULTS_NS_ET, RESULTS_NS_ET)))]

## `` XMLResultParser

    XMLResultParser()

Bases: `ResultParser`

A Parser for SPARQL results in XML.

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/xmlresults.py`

    54
    55

    def parse(self, source: IO, content_type: Optional[str] = None) -> Result:  # type: ignore[override]
        return XMLResult(source)

## `` XMLResultSerializer

    XMLResultSerializer(result: Result)

Bases: `ResultSerializer`

Serializes SPARQL results into XML format.

Methods:

- `serialize` –

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

    160
    161

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

### `` serialize

    serialize(stream: IO, encoding: str = 'utf-8', **kwargs: Any) -> None

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

    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180

    def serialize(self, stream: IO, encoding: str = "utf-8", **kwargs: Any) -> None:
        writer = SPARQLXMLWriter(stream, encoding)
        if self.result.type == "ASK":
            writer.write_header([])
            # type error: Argument 1 to "write_ask" of "SPARQLXMLWriter" has incompatible type "Optional[bool]"; expected "bool"
            writer.write_ask(self.result.askAnswer)  # type: ignore[arg-type]
        else:
            # type error: Argument 1 to "write_header" of "SPARQLXMLWriter" has incompatible type "Optional[List[Variable]]"; expected "Sequence[Variable]"
            writer.write_header(self.result.vars)  # type: ignore[arg-type]
            writer.write_results_header()
            for b in self.result.bindings:
                writer.write_start_result()
                for key, val in b.items():
                    writer.write_binding(key, val)

                writer.write_end_result()

        writer.close()

## `` parseTerm

    parseTerm(element: Element) -> Union[URIRef, Literal, BNode]

rdflib object (Literal, URIRef, BNode) for the given elementtree element

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

    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154

    def parseTerm(element: xml_etree.Element) -> Union[URIRef, Literal, BNode]:
        """rdflib object (Literal, URIRef, BNode) for the given
        elementtree element"""
        tag, text = element.tag, element.text
        if tag == RESULTS_NS_ET + "literal":
            if text is None:
                text = ""
            datatype = None
            lang = None
            if element.get("datatype", None):
                # type error: Argument 1 to "URIRef" has incompatible type "Optional[str]"; expected "str"
                datatype = URIRef(element.get("datatype"))  # type: ignore[arg-type]
            elif element.get("{%s}lang" % XML_NAMESPACE, None):
                lang = element.get("{%s}lang" % XML_NAMESPACE)

            ret = Literal(text, datatype=datatype, lang=lang)

            return ret
        elif tag == RESULTS_NS_ET + "uri":
            # type error: Argument 1 to "URIRef" has incompatible type "Optional[str]"; expected "str"
            return URIRef(text)  # type: ignore[arg-type]
        elif tag == RESULTS_NS_ET + "bnode":
            return BNode(text)
        else:
            raise TypeError("unknown binding type %r" % element)