Skip to content

sparqlconnector

Classes:

SUPPORTED_FORMATS module-attribute 

SUPPORTED_FORMATS = Literal['xml', 'json', 'csv', 'tsv', 'application/rdf+xml']

SUPPORTED_METHODS module-attribute 

SUPPORTED_METHODS = Literal['GET', 'POST', 'POST_FORM']

__all__ module-attribute 

__all__ = ['SPARQLConnector', 'SPARQLConnectorException']

log module-attribute 

log = getLogger(__name__)

SPARQLConnector 

SPARQLConnector(query_endpoint: str | None = None, update_endpoint: str | None = None, returnFormat: SUPPORTED_FORMATS = 'xml', method: SUPPORTED_METHODS = 'GET', auth: tuple[str, str] | None = None, **kwargs)

this class deals with nitty gritty details of talking to a SPARQL server

Any additional keyword arguments will be passed to to the request, and can be used to setup timeouts etc.

Methods:

Attributes:

Source code in rdflib/plugins/stores/sparqlconnector.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    query_endpoint: str | None = None,
    update_endpoint: str | None = None,
    returnFormat: SUPPORTED_FORMATS = "xml",  # noqa: N803
    method: SUPPORTED_METHODS = "GET",
    auth: tuple[str, str] | None = None,
    **kwargs,
):
    """
    auth, if present, must be a tuple of (username, password) used for Basic Authentication

    Any additional keyword arguments will be passed to to the request, and can be used to setup timeouts etc.
    """
    self._method: str
    self.returnFormat = returnFormat
    self.query_endpoint = query_endpoint
    self.update_endpoint = update_endpoint
    self.kwargs = kwargs
    self.method = method
    if auth is not None:
        if type(auth) is not tuple:
            raise SPARQLConnectorException("auth must be a tuple")
        if len(auth) != 2:
            raise SPARQLConnectorException("auth must be a tuple (user, password)")
        base64string = base64.b64encode(bytes("%s:%s" % auth, "ascii"))
        self.kwargs.setdefault("headers", {})
        self.kwargs["headers"].update(
            {"Authorization": "Basic %s" % base64string.decode("utf-8")}
        )

kwargs instance-attribute 

kwargs = kwargs

method property writable 

method: str

query_endpoint instance-attribute 

query_endpoint = query_endpoint

returnFormat instance-attribute 

returnFormat = returnFormat

update_endpoint instance-attribute 

update_endpoint = update_endpoint

query 

query(query: str, default_graph: str | None = None, named_graph: str | None = None) -> Result
Source code in rdflib/plugins/stores/sparqlconnector.py 
 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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def query(
    self,
    query: str,
    default_graph: str | None = None,
    named_graph: str | None = None,
) -> Result:
    if not self.query_endpoint:
        raise SPARQLConnectorException("Query endpoint not set!")

    params = {}
    # this test ensures we don't have a useless (BNode) default graph URI, which calls to Graph().query() will add
    if default_graph is not None and type(default_graph) is not BNode:
        params["default-graph-uri"] = default_graph

    headers = {"Accept": self.response_mime_types()}

    args = copy.deepcopy(self.kwargs)

    # merge params/headers dicts
    args.setdefault("params", {})

    args.setdefault("headers", {})
    args["headers"].update(headers)

    if self.method == "GET":
        params["query"] = query
        args["params"].update(params)
        qsa = "?" + urlencode(args["params"])
        try:
            res = urlopen(
                Request(self.query_endpoint + qsa, headers=args["headers"])
            )
        except Exception as e:  # noqa: F841
            raise ValueError(
                "You did something wrong formulating either the URI or your SPARQL query"
            )
    elif self.method == "POST":
        args["headers"].update({"Content-Type": "application/sparql-query"})
        args["params"].update(params)
        qsa = "?" + urlencode(args["params"])
        try:
            res = urlopen(
                Request(
                    self.query_endpoint + qsa,
                    data=query.encode(),
                    headers=args["headers"],
                )
            )
        except HTTPError as e:
            # type error: Incompatible return value type (got "Tuple[int, str, None]", expected "Result")
            return e.code, str(e), None  # type: ignore[return-value]
    elif self.method == "POST_FORM":
        params["query"] = query
        args["params"].update(params)
        try:
            res = urlopen(
                Request(
                    self.query_endpoint,
                    data=urlencode(args["params"]).encode(),
                    headers=args["headers"],
                )
            )
        except HTTPError as e:
            # type error: Incompatible return value type (got "Tuple[int, str, None]", expected "Result")
            return e.code, str(e), None  # type: ignore[return-value]
    else:
        raise SPARQLConnectorException("Unknown method %s" % self.method)
    return Result.parse(
        BytesIO(res.read()), content_type=res.headers["Content-Type"].split(";")[0]
    )

response_mime_types 

response_mime_types() -> str

Construct a HTTP-Header Accept field to reflect the supported mime types.

If the return_format parameter is set, the mime types are restricted to these accordingly.

Source code in rdflib/plugins/stores/sparqlconnector.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def response_mime_types(self) -> str:
    """Construct a HTTP-Header Accept field to reflect the supported mime types.

    If the return_format parameter is set, the mime types are restricted to these accordingly.
    """
    sparql_format_mimetype_map = {
        k: FORMAT_MIMETYPE_MAP.get(k, [])
        + RESPONSE_TABLE_FORMAT_MIMETYPE_MAP.get(k, [])
        for k in list(FORMAT_MIMETYPE_MAP.keys())
        + list(RESPONSE_TABLE_FORMAT_MIMETYPE_MAP.keys())
    }

    supported_formats = set()
    for plugin in plugins(name=self.returnFormat, kind=ResultParser):
        if "/" not in plugin.name:
            supported_formats.update(
                sparql_format_mimetype_map.get(plugin.name, [])
            )
        else:
            supported_formats.add(plugin.name)
    return ", ".join(supported_formats)

update 

update(query: str, default_graph: str | None = None, named_graph: str | None = None) -> None
Source code in rdflib/plugins/stores/sparqlconnector.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def update(
    self,
    query: str,
    default_graph: str | None = None,
    named_graph: str | None = None,
) -> None:
    if not self.update_endpoint:
        raise SPARQLConnectorException("Query endpoint not set!")

    params = {}

    if default_graph is not None:
        params["using-graph-uri"] = default_graph

    if named_graph is not None:
        params["using-named-graph-uri"] = named_graph

    headers = {
        "Accept": self.response_mime_types(),
        "Content-Type": "application/sparql-update; charset=UTF-8",
    }

    args = copy.deepcopy(self.kwargs)  # other QSAs

    args.setdefault("params", {})
    args["params"].update(params)
    args.setdefault("headers", {})
    args["headers"].update(headers)

    qsa = "?" + urlencode(args["params"])
    res = urlopen(  # noqa: F841
        Request(
            self.update_endpoint + qsa, data=query.encode(), headers=args["headers"]
        )
    )

SPARQLConnectorException

Bases: Exception