Skip to content

query

Classes:

__all__ module-attribute 

__all__ = ['Processor', 'UpdateProcessor', 'Result', 'ResultRow', 'ResultParser', 'ResultSerializer', 'ResultException', 'EncodeOnlyUnicode']

EncodeOnlyUnicode 

EncodeOnlyUnicode(stream: BinaryIO)

This is a crappy work-around for http://bugs.python.org/issue11649

Methods:

Source code in rdflib/query.py 
98
99
def __init__(self, stream: BinaryIO):
    self.__stream = stream

__stream instance-attribute 

__stream = stream

__getattr__ 

__getattr__(name: str) -> Any
Source code in rdflib/query.py 
107
108
def __getattr__(self, name: str) -> Any:
    return getattr(self.__stream, name)

write 

write(arg)
Source code in rdflib/query.py 
101
102
103
104
105
def write(self, arg):
    if isinstance(arg, str):
        self.__stream.write(arg.encode("utf-8"))
    else:
        self.__stream.write(arg)

Processor 

Processor(graph: Graph)

Query plugin interface.

This module is useful for those wanting to write a query processor that can plugin to rdf. If you are wanting to execute a query you likely want to do so through the Graph class query method.

Methods:

Source code in rdflib/query.py 
54
55
def __init__(self, graph: Graph):
    pass

query 

query(strOrQuery: Union[str, Query], initBindings: Mapping[str, Identifier] = {}, initNs: Mapping[str, Any] = {}, DEBUG: bool = False) -> Mapping[str, Any]
Source code in rdflib/query.py 
58
59
60
61
62
63
64
65
def query(  # type: ignore[empty-body]
    self,
    strOrQuery: Union[str, Query],  # noqa: N803
    initBindings: Mapping[str, Identifier] = {},  # noqa: N803
    initNs: Mapping[str, Any] = {},  # noqa: N803
    DEBUG: bool = False,  # noqa: N803
) -> Mapping[str, Any]:
    pass

Result 

Result(type_: str)

A common class for representing query result.

There is a bit of magic here that makes this appear like different Python objects, depending on the type of result.

If the type is “SELECT”, iterating will yield lists of ResultRow objects

If the type is “ASK”, iterating will yield a single bool (or bool(result) will return the same bool)

If the type is “CONSTRUCT” or “DESCRIBE” iterating will yield the triples.

len(result) also works.

Methods:

Attributes:

Source code in rdflib/query.py 
213
214
215
216
217
218
219
220
221
222
223
224
def __init__(self, type_: str):
    if type_ not in ("CONSTRUCT", "DESCRIBE", "SELECT", "ASK"):
        raise ResultException("Unknown Result type: %s" % type_)

    self.type = type_
    #: variables contained in the result.
    self.vars: Optional[List[Variable]] = None
    """a list of variables contained in the result"""
    self._bindings: MutableSequence[Mapping[Variable, Identifier]] = None  # type: ignore[assignment]
    self._genbindings: Optional[Iterator[Mapping[Variable, Identifier]]] = None
    self.askAnswer: Optional[bool] = None
    self.graph: Optional[Graph] = None

askAnswer instance-attribute 

askAnswer: Optional[bool] = None

bindings property writable 

bindings: MutableSequence[Mapping[Variable, Identifier]]

a list of variable bindings as dicts

graph instance-attribute 

graph: Optional[Graph] = None

type instance-attribute 

type = type_

vars instance-attribute 

vars: Optional[List[Variable]] = None

a list of variables contained in the result

__bool__ 

__bool__() -> bool
Source code in rdflib/query.py 
351
352
353
354
355
356
def __bool__(self) -> bool:
    if self.type == "ASK":
        # type error: Incompatible return value type (got "Optional[bool]", expected "bool")
        return self.askAnswer  # type: ignore[return-value]
    else:
        return len(self) > 0

__eq__ 

__eq__(other: Any) -> bool
Source code in rdflib/query.py 
401
402
403
404
405
406
407
408
409
410
411
412
def __eq__(self, other: Any) -> bool:
    try:
        if self.type != other.type:
            return False
        if self.type == "ASK":
            return self.askAnswer == other.askAnswer
        elif self.type == "SELECT":
            return self.vars == other.vars and self.bindings == other.bindings
        else:
            return self.graph == other.graph
    except Exception:
        return False

__getattr__ 

__getattr__(name: str) -> Any
Source code in rdflib/query.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
def __getattr__(self, name: str) -> Any:
    if self.type in ("CONSTRUCT", "DESCRIBE") and self.graph is not None:
        # type error: "Graph" has no attribute "__getattr__"
        return self.graph.__getattr__(self, name)  # type: ignore[attr-defined]
    elif self.type == "SELECT" and name == "result":
        warnings.warn(
            "accessing the 'result' attribute is deprecated."
            " Iterate over the object instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        # copied from __iter__, above
        # type error: Item "None" of "Optional[List[Variable]]" has no attribute "__iter__" (not iterable)
        return [(tuple(b[v] for v in self.vars)) for b in self.bindings]  # type: ignore[union-attr]
    else:
        raise AttributeError("'%s' object has no attribute '%s'" % (self, name))

__iter__ 

__iter__() -> Iterator[Union[_TripleType, bool, ResultRow]]
Source code in rdflib/query.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def __iter__(
    self,
) -> Iterator[Union[_TripleType, bool, ResultRow]]:
    if self.type in ("CONSTRUCT", "DESCRIBE"):
        # type error: Item "None" of "Optional[Graph]" has no attribute "__iter__" (not iterable)
        for t in self.graph:  # type: ignore[union-attr]
            yield t
    elif self.type == "ASK":
        # type error: Incompatible types in "yield" (actual type "Optional[bool]", expected type "Union[Tuple[Identifier, Identifier, Identifier], bool, ResultRow]")  [misc]
        yield self.askAnswer  # type: ignore[misc]
    elif self.type == "SELECT":
        # this iterates over ResultRows of variable bindings

        if self._genbindings:
            for b in self._genbindings:
                if b:  # don't add a result row in case of empty binding {}
                    self._bindings.append(b)
                    # type error: Argument 2 to "ResultRow" has incompatible type "Optional[List[Variable]]"; expected "List[Variable]"
                    yield ResultRow(b, self.vars)  # type: ignore[arg-type]
            self._genbindings = None
        else:
            for b in self._bindings:
                if b:  # don't add a result row in case of empty binding {}
                    # type error: Argument 2 to "ResultRow" has incompatible type "Optional[List[Variable]]"; expected "List[Variable]"
                    yield ResultRow(b, self.vars)  # type: ignore[arg-type]

__len__ 

__len__() -> int
Source code in rdflib/query.py 
342
343
344
345
346
347
348
349
def __len__(self) -> int:
    if self.type == "ASK":
        return 1
    elif self.type == "SELECT":
        return len(self.bindings)
    else:
        # type error: Argument 1 to "len" has incompatible type "Optional[Graph]"; expected "Sized"
        return len(self.graph)  # type: ignore[arg-type]

parse staticmethod 

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

Parse a query result from a source.

Source code in rdflib/query.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@staticmethod
def parse(
    source: Optional[IO] = None,
    format: Optional[str] = None,
    content_type: Optional[str] = None,
    **kwargs: Any,
) -> Result:
    """Parse a query result from a source."""
    from rdflib import plugin

    if format:
        plugin_key = format
    elif content_type:
        plugin_key = content_type.split(";", 1)[0]
    else:
        plugin_key = "xml"

    parser = plugin.get(plugin_key, ResultParser)()

    # type error: Argument 1 to "parse" of "ResultParser" has incompatible type "Optional[IO[Any]]"; expected "IO[Any]"
    return parser.parse(
        source, content_type=content_type, **kwargs  # type:ignore[arg-type]
    )

serialize 

serialize(destination: Optional[Union[str, IO]] = None, encoding: str = 'utf-8', format: str = 'xml', **args: Any) -> Optional[bytes]

Serialize the query result.

Parameters:

  • destination

    (Optional[Union[str, IO]], default: None ) –

    Path of file output or BufferedIOBase object to write the output to. If None this function will return the output as bytes

  • encoding

    (str, default: 'utf-8' ) –

    Encoding of output.

  • format

    (str, default: 'xml' ) –

    The format used for serialization. See sparql.results module for all builtin SPARQL result serialization. Further serializer can be loaded as plugin. Some example formats are csv, json, txt or xml

Returns:

  • Optional[bytes] –

    Serialized result, when destination is not given.

Source code in rdflib/query.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def serialize(
    self,
    destination: Optional[Union[str, IO]] = None,
    encoding: str = "utf-8",
    format: str = "xml",
    **args: Any,
) -> Optional[bytes]:
    """
    Serialize the query result.


    Args:
        destination: Path of file output or BufferedIOBase object
            to write the output to. If `None` this function
            will return the output as  `bytes`
        encoding: Encoding of output.
        format: The format used for serialization.
            See [sparql.results module][rdflib.plugins.sparql.results]
            for all builtin SPARQL result serialization.
            Further serializer can be loaded [as plugin][rdflib.plugin].
            Some example formats are
            [csv][rdflib.plugins.sparql.results.csvresults.CSVResultSerializer],
            [json][rdflib.plugins.sparql.results.jsonresults.JSONResultSerializer],
            [txt][rdflib.plugins.sparql.results.txtresults.TXTResultSerializer]
            or
            [xml][rdflib.plugins.sparql.results.xmlresults.XMLResultSerializer]

    Returns:
        Serialized result, when destination is not given.
    """
    if self.type in ("CONSTRUCT", "DESCRIBE"):
        # type error: Item "None" of "Optional[Graph]" has no attribute "serialize"
        # type error: Incompatible return value type (got "Union[bytes, str, Graph, Any]", expected "Optional[bytes]")
        return self.graph.serialize(  # type: ignore[union-attr,return-value]
            destination, encoding=encoding, format=format, **args
        )

    """stolen wholesale from graph.serialize"""
    from rdflib import plugin

    serializer = plugin.get(format, ResultSerializer)(self)
    if destination is None:
        streamb: BytesIO = BytesIO()
        stream2 = EncodeOnlyUnicode(streamb)  # TODO: Remove the need for this
        # TODO: All QueryResult serializers should write to a Bytes Stream.
        # type error: Argument 1 to "serialize" of "ResultSerializer" has incompatible type "EncodeOnlyUnicode"; expected "IO[Any]"
        serializer.serialize(stream2, encoding=encoding, **args)  # type: ignore[arg-type]
        return streamb.getvalue()
    if hasattr(destination, "write"):
        stream = cast(IO[bytes], destination)
        serializer.serialize(stream, encoding=encoding, **args)
    else:
        location = cast(str, destination)
        scheme, netloc, path, params, query, fragment = urlparse(location)
        if scheme == "file":
            if netloc != "":
                raise ValueError(
                    f"the file URI {location!r} has an authority component which is not supported"
                )
            os_path = url2pathname(path)
        else:
            os_path = location
        with open(os_path, "wb") as stream:
            serializer.serialize(stream, encoding=encoding, **args)
    return None

ResultException

Bases: Exception

ResultParser 

ResultParser()

Methods:

  • parse –

    return a Result object

Source code in rdflib/query.py 
416
417
def __init__(self):
    pass

parse 

parse(source: IO, **kwargs: Any) -> Result

return a Result object

Source code in rdflib/query.py 
420
421
422
def parse(self, source: IO, **kwargs: Any) -> Result:  # type: ignore[empty-body]
    """return a Result object"""
    pass  # abstract

ResultRow

Bases: Tuple[Identifier, ...]

A single result row allows accessing bindings as attributes or with []

>>> from rdflib import URIRef, Variable
>>> rr=ResultRow({ Variable('a'): URIRef('urn:cake') }, [Variable('a')])

>>> rr[0]
rdflib.term.URIRef('urn:cake')
>>> rr[1]
Traceback (most recent call last):
    ...
IndexError: tuple index out of range

>>> rr.a
rdflib.term.URIRef('urn:cake')
>>> rr.b
Traceback (most recent call last):
    ...
AttributeError: b

>>> rr['a']
rdflib.term.URIRef('urn:cake')
>>> rr['b']
Traceback (most recent call last):
    ...
KeyError: 'b'

>>> rr[Variable('a')]
rdflib.term.URIRef('urn:cake')

New in version 4.0

Methods:

Attributes:

  • labels (Mapping[str, int]) –

labels instance-attribute 

labels: Mapping[str, int]

__getattr__ 

__getattr__(name: str) -> Identifier
Source code in rdflib/query.py 
156
157
158
159
def __getattr__(self, name: str) -> Identifier:
    if name not in self.labels:
        raise AttributeError(name)
    return tuple.__getitem__(self, self.labels[name])

__getitem__ 

__getitem__(name: Union[str, int, Any]) -> Identifier
Source code in rdflib/query.py 
163
164
165
166
167
168
169
170
171
172
173
def __getitem__(self, name: Union[str, int, Any]) -> Identifier:  # type: ignore[override]
    try:
        # type error: Invalid index type "Union[str, int, Any]" for "tuple"; expected type "int"
        return tuple.__getitem__(self, name)  # type: ignore[index]
    except TypeError:
        if name in self.labels:
            # type error: Invalid index type "Union[str, int, slice, Any]" for "Mapping[str, int]"; expected type "str"
            return tuple.__getitem__(self, self.labels[name])  # type: ignore[index]
        if str(name) in self.labels:  # passing in variable object
            return tuple.__getitem__(self, self.labels[str(name)])
        raise KeyError(name)

__new__ 

__new__(values: Mapping[Variable, Identifier], labels: List[Variable])
Source code in rdflib/query.py 
149
150
151
152
153
154
def __new__(cls, values: Mapping[Variable, Identifier], labels: List[Variable]):
    # type error: Value of type variable "Self" of "__new__" of "tuple" cannot be "ResultRow"  [type-var]
    # type error: Generator has incompatible item type "Optional[Identifier]"; expected "_T_co"  [misc]
    instance = super(ResultRow, cls).__new__(cls, (values.get(v) for v in labels))  # type: ignore[type-var, misc, unused-ignore]
    instance.labels = dict((str(x[1]), x[0]) for x in enumerate(labels))
    return instance

asdict 

asdict() -> Dict[str, Identifier]
Source code in rdflib/query.py 
191
192
def asdict(self) -> Dict[str, Identifier]:
    return dict((v, self[v]) for v in self.labels if self[v] is not None)

get 

get(name: str, default: Identifier) -> Identifier

get(name: str, default: Optional[Identifier] = ...) -> Optional[Identifier]

get(name: str, default: Optional[Identifier] = None) -> Optional[Identifier]
Source code in rdflib/query.py 
183
184
185
186
187
188
189
def get(
    self, name: str, default: Optional[Identifier] = None
) -> Optional[Identifier]:
    try:
        return self[name]
    except KeyError:
        return default

ResultSerializer 

ResultSerializer(result: Result)

Methods:

  • serialize –

    return a string properly serialized

Attributes:

Source code in rdflib/query.py 
426
427
def __init__(self, result: Result):
    self.result = result

result instance-attribute 

result = result

serialize 

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

return a string properly serialized

Source code in rdflib/query.py 
429
430
431
def serialize(self, stream: IO, encoding: str = "utf-8", **kwargs: Any) -> None:
    """return a string properly serialized"""
    pass  # abstract

UpdateProcessor 

UpdateProcessor(graph: Graph)

Update plugin interface.

This module is useful for those wanting to write an update processor that can plugin to rdflib. If you are wanting to execute an update statement you likely want to do so through the Graph class update method.

New in version 4.0

Methods:

Source code in rdflib/query.py 
79
80
def __init__(self, graph: Graph):
    pass

update 

update(strOrQuery: Union[str, Update], initBindings: Mapping[str, Identifier] = {}, initNs: Mapping[str, Any] = {}) -> None
Source code in rdflib/query.py 
82
83
84
85
86
87
88
def update(
    self,
    strOrQuery: Union[str, Update],  # noqa: N803
    initBindings: Mapping[str, Identifier] = {},  # noqa: N803
    initNs: Mapping[str, Any] = {},  # noqa: N803
) -> None:
    pass