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:

Functions:

  • parseTerm –

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

Attributes:

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:

Attributes:

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:

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:

Source code in rdflib/query.py 
416
417
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:

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)