sparqlconnector

Classes:

• SPARQLConnector –

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

• SPARQLConnectorException –

__all__ module-attribute

__all__ = ['SPARQLConnector', 'SPARQLConnectorException']

log module-attribute

log = getLogger(__name__)

SPARQLConnector

SPARQLConnector(query_endpoint: Optional[str] = None, update_endpoint: Optional[str] = None, returnFormat: Optional[str] = 'xml', method: Literal['GET', 'POST', 'POST_FORM'] = 'GET', auth: Optional[Tuple[str, str]] = 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:

• query –
• response_mime_types –

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

• update –

Attributes:

• kwargs –
• method (str) –
• query_endpoint –
• returnFormat –
• update_endpoint –

Source code in rdflib/plugins/stores/sparqlconnector.py

def __init__(
    self,
    query_endpoint: Optional[str] = None,
    update_endpoint: Optional[str] = None,
    returnFormat: Optional[str] = "xml",  # noqa: N803
    method: te.Literal["GET", "POST", "POST_FORM"] = "GET",
    auth: Optional[Tuple[str, str]] = 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: Optional[str] = None, named_graph: Optional[str] = None) -> Result

Source code in rdflib/plugins/stores/sparqlconnector.py

def query(
    self,
    query: str,
    default_graph: Optional[str] = None,
    named_graph: Optional[str] = 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

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: Optional[str] = None, named_graph: Optional[str] = None) -> None

Source code in rdflib/plugins/stores/sparqlconnector.py

def update(
    self,
    query: str,
    default_graph: Optional[str] = None,
    named_graph: Optional[str] = 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