Skip to content

# `` query

Classes:

- `EncodeOnlyUnicode` –

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

- `Processor` –

  Query plugin interface.

- `Result` –

  A common class for representing query result.

- `ResultException` –
- `ResultParser` –
- `ResultRow` –

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

- `ResultSerializer` –
- `UpdateProcessor` –

  Update plugin interface.

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

- `__getattr__` –
- `write` –

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:

- `query` –

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:

- `__bool__` –
- `__eq__` –
- `__getattr__` –
- `__iter__` –
- `__len__` –
- `parse` –

  Parse a query result from a source.

- `serialize` –

  Serialize the query result.

Attributes:

- `askAnswer` (`Optional[bool]`) –
- `bindings` (`MutableSequence[Mapping[Variable, Identifier]]`) –

  a list of variable bindings as dicts

- `graph` (`Optional[Graph]`) –
- `type` –
- `vars` (`Optional[List[Variable]]`) –

  a list of variables contained in the result

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`

    346
    347
    348
    349
    350
    351

    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`

    396
    397
    398
    399
    400
    401
    402
    403
    404
    405
    406
    407

    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`

    379
    380
    381
    382
    383
    384
    385
    386
    387
    388
    389
    390
    391
    392
    393
    394

    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`

    353
    354
    355
    356
    357
    358
    359
    360
    361
    362
    363
    364
    365
    366
    367
    368
    369
    370
    371
    372
    373
    374
    375
    376
    377

    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`

    337
    338
    339
    340
    341
    342
    343
    344

    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.

The `format` argument determines the Serializer class to use.

- csv: `CSVResultSerializer`
- json: `JSONResultSerializer`
- txt: `TXTResultSerializer`
- xml: `XMLResultSerializer`

Parameters:

- #### `destination`

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

  Path of file output or BufferedIOBase object to write the output to.

- #### `encoding`

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

  Encoding of output.

- #### `format`

  (`str`, default: `'xml'` ) –

  One of [‘csv’, ‘json’, ‘txt’, xml’]

Returns:

- `Optional[bytes]` –

  bytes

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

    def serialize(
        self,
        destination: Optional[Union[str, IO]] = None,
        encoding: str = "utf-8",
        format: str = "xml",
        **args: Any,
    ) -> Optional[bytes]:
        """
        Serialize the query result.

        The `format` argument determines the Serializer class to use.

        - csv: [`CSVResultSerializer`][rdflib.plugins.sparql.results.csvresults.CSVResultSerializer]
        - json: [`JSONResultSerializer`][rdflib.plugins.sparql.results.jsonresults.JSONResultSerializer]
        - txt: [`TXTResultSerializer`][rdflib.plugins.sparql.results.txtresults.TXTResultSerializer]
        - xml: [`XMLResultSerializer`][rdflib.plugins.sparql.results.xmlresults.XMLResultSerializer]

        Args:
            destination: Path of file output or BufferedIOBase object to write the output to.
            encoding: Encoding of output.
            format: One of ['csv', 'json', 'txt', xml']

        Returns:
            bytes
        """
        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`

    411
    412

    def __init__(self):
        pass

### `` parse

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

return a Result object

Source code in `rdflib/query.py`

    415
    416
    417

    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:

- `__getattr__` –
- `__getitem__` –
- `__new__` –
- `asdict` –
- `get` –

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

- `result` –

Source code in `rdflib/query.py`

    421
    422

    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`

    424
    425
    426

    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:

- `update` –

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