Skip to content

# `` processor

Code for tying SPARQL Engine into RDFLib

These should be automatically registered with RDFLib

Classes:

- `SPARQLProcessor` –
- `SPARQLResult` –
- `SPARQLUpdateProcessor` –

Functions:

- `prepareQuery` –

  Parse and translate a SPARQL Query

- `prepareUpdate` –

  Parse and translate a SPARQL Update

- `processUpdate` –

  Process a SPARQL Update Request

## `` SPARQLProcessor

    SPARQLProcessor(graph)

Bases: `Processor`

Methods:

- `query` –

  Evaluate a query with the given initial bindings, and initial

Attributes:

- `graph` –

Source code in `rdflib/plugins/sparql/processor.py`

    109
    110

    def __init__(self, graph):
        self.graph = graph

### `` graph `instance-attribute`

    graph = graph

### `` query

    query(strOrQuery: Union[str, Query], initBindings: Optional[Mapping[str, Identifier]] = None, initNs: Optional[Mapping[str, Any]] = None, base: Optional[str] = None, DEBUG: bool = False) -> Mapping[str, Any]

Evaluate a query with the given initial bindings, and initial namespaces. The given base is used to resolve relative URIs in the query and will be overridden by any BASE given in the query.

Caution

This method can access indirectly requested network endpoints, for example, query processing will attempt to access network endpoints specified in `SERVICE` directives.

When processing untrusted or potentially malicious queries, measures should be taken to restrict network and file access.

For information on available security measures, see the RDFLib Security Considerations documentation.

Source code in `rdflib/plugins/sparql/processor.py`

    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146

    def query(  # type: ignore[override]
        self,
        strOrQuery: Union[str, Query],
        initBindings: Optional[Mapping[str, Identifier]] = None,
        initNs: Optional[Mapping[str, Any]] = None,
        base: Optional[str] = None,
        DEBUG: bool = False,
    ) -> Mapping[str, Any]:
        """
        Evaluate a query with the given initial bindings, and initial
        namespaces. The given base is used to resolve relative URIs in
        the query and will be overridden by any BASE given in the query.

        !!! warning "Caution"

            This method can access indirectly requested network endpoints, for
            example, query processing will attempt to access network endpoints
            specified in `SERVICE` directives.

            When processing untrusted or potentially malicious queries, measures
            should be taken to restrict network and file access.

            For information on available security measures, see the RDFLib
            [Security Considerations](../security_considerations.md)
            documentation.
        """

        if isinstance(strOrQuery, str):
            strOrQuery = translateQuery(parseQuery(strOrQuery), base, initNs)

        return evalQuery(self.graph, strOrQuery, initBindings, base)

## `` SPARQLResult

    SPARQLResult(res: Mapping[str, Any])

Bases: `Result`

Attributes:

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

Source code in `rdflib/plugins/sparql/processor.py`

    68
    69
    70
    71
    72
    73
    74

    def __init__(self, res: Mapping[str, Any]):
        Result.__init__(self, res["type_"])
        self.vars = res.get("vars_")
        # type error: Incompatible types in assignment (expression has type "Optional[Any]", variable has type "MutableSequence[Mapping[Variable, Identifier]]")
        self.bindings = res.get("bindings")  # type: ignore[assignment]
        self.askAnswer = res.get("askAnswer")
        self.graph = res.get("graph")

### `` askAnswer `instance-attribute`

    askAnswer = get('askAnswer')

### `` bindings `instance-attribute`

    bindings = get('bindings')

### `` graph `instance-attribute`

    graph = get('graph')

### `` vars `instance-attribute`

    vars = get('vars_')

## `` SPARQLUpdateProcessor

    SPARQLUpdateProcessor(graph)

Bases: `UpdateProcessor`

Methods:

- `update` –

  Caution

Attributes:

- `graph` –

Source code in `rdflib/plugins/sparql/processor.py`

    78
    79

    def __init__(self, graph):
        self.graph = graph

### `` graph `instance-attribute`

    graph = graph

### `` update

    update(strOrQuery: Union[str, Update], initBindings: Optional[Mapping[str, Identifier]] = None, initNs: Optional[Mapping[str, Any]] = None) -> None

Caution

This method can access indirectly requested network endpoints, for example, query processing will attempt to access network endpoints specified in `SERVICE` directives.

When processing untrusted or potentially malicious queries, measures should be taken to restrict network and file access.

For information on available security measures, see the RDFLib Security Considerations documentation.

Source code in `rdflib/plugins/sparql/processor.py`

     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

    def update(
        self,
        strOrQuery: Union[str, Update],
        initBindings: Optional[Mapping[str, Identifier]] = None,
        initNs: Optional[Mapping[str, Any]] = None,
    ) -> None:
        """
        !!! warning "Caution"

            This method can access indirectly requested network endpoints, for
            example, query processing will attempt to access network endpoints
            specified in `SERVICE` directives.

            When processing untrusted or potentially malicious queries, measures
            should be taken to restrict network and file access.

            For information on available security measures, see the RDFLib
            [Security Considerations](../security_considerations.md)
            documentation.
        """

        if isinstance(strOrQuery, str):
            strOrQuery = translateUpdate(parseUpdate(strOrQuery), initNs=initNs)

        return evalUpdate(self.graph, strOrQuery, initBindings)

## `` prepareQuery

    prepareQuery(queryString: str, initNs: Optional[Mapping[str, Any]] = None, base: Optional[str] = None) -> Query

Parse and translate a SPARQL Query

Source code in `rdflib/plugins/sparql/processor.py`

    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33

    def prepareQuery(
        queryString: str,
        initNs: Optional[Mapping[str, Any]] = None,
        base: Optional[str] = None,
    ) -> Query:
        """
        Parse and translate a SPARQL Query
        """
        if initNs is None:
            initNs = {}
        ret = translateQuery(parseQuery(queryString), base, initNs)
        ret._original_args = (queryString, initNs, base)
        return ret

## `` prepareUpdate

    prepareUpdate(updateString: str, initNs: Optional[Mapping[str, Any]] = None, base: Optional[str] = None) -> Update

Parse and translate a SPARQL Update

Source code in `rdflib/plugins/sparql/processor.py`

    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48

    def prepareUpdate(
        updateString: str,
        initNs: Optional[Mapping[str, Any]] = None,
        base: Optional[str] = None,
    ) -> Update:
        """
        Parse and translate a SPARQL Update
        """
        if initNs is None:
            initNs = {}
        ret = translateUpdate(parseUpdate(updateString), base, initNs)
        ret._original_args = (updateString, initNs, base)
        return ret

## `` processUpdate

    processUpdate(graph: Graph, updateString: str, initBindings: Optional[Mapping[str, Identifier]] = None, initNs: Optional[Mapping[str, Any]] = None, base: Optional[str] = None) -> None

Process a SPARQL Update Request returns Nothing on success or raises Exceptions on error

Source code in `rdflib/plugins/sparql/processor.py`

    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64

    def processUpdate(
        graph: Graph,
        updateString: str,
        initBindings: Optional[Mapping[str, Identifier]] = None,
        initNs: Optional[Mapping[str, Any]] = None,
        base: Optional[str] = None,
    ) -> None:
        """
        Process a SPARQL Update Request
        returns Nothing on success or raises Exceptions on error
        """
        evalUpdate(
            graph, translateUpdate(parseUpdate(updateString), base, initNs), initBindings
        )