Skip to content

gsp

SPARQL Graph Store Protocol operations

GraphStore

Encapsulates Graph Store Protocol configuration for executing operations.

Source code in ogc/na/gsp.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 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
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 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
class GraphStore:
    """
    Encapsulates Graph Store Protocol configuration for executing operations.
    """

    def __init__(self, url: str, auth_details: tuple[str, str] | None = None):
        """
        Constructs a new GraphStore
        :param url: SPARQL Graph Store Protocol URL
        :param auth_details: tuple in the form ('username', 'password') for authentication
        """
        self._url = url
        self.auth_details = auth_details
        self.put = self.replace
        self.post = self.add

    def _post_or_put(self, method: str, graph_uri, source: str | bytes | Path, format: str = 'text/turtle'):
        if graph_uri:
            params = {'graph': str(graph_uri)}
        else:
            params = 'default'

        if isinstance(source, Path):
            with open(source, 'r') as f:
                data = f.read()
        elif isinstance(source, IO) or isinstance(source, IOBase):
            data = source.read()
        else:
            data = source

        if isinstance(data, str):
            data = data.encode('utf-8')

        method = getattr(requests, method)
        r = method(
            self._url,
            params=params,
            headers={
                'Content-type': format,
            },
            auth=self.auth_details,
            data=data,
        )

        if isinstance(data, Generator):
            data.close()

        r.raise_for_status()

    def delete(self, graph_uri: str | None, ignore_404=True):
        """
        Deletes a graph from the Graph Store
        :param graph_uri: URI for the graph (if None, the default graph will be used)
        :param ignore_404: Whether to ignore HTTP 404 errors (otherwise, an Exception will be thrown)
        :return:
        """
        if graph_uri:
            params = {'graph': str(graph_uri)}
        else:
            params = 'default'

        r = requests.delete(
            self._url,
            params=params,
            auth=self.auth_details,
        )
        if not (ignore_404 and r.status_code == 404):
            r.raise_for_status()

    def replace(self, graph_uri: str | None, source: str | bytes | Path, format: str = 'text/turtle'):
        """
        Replaces the data for a Graph Store graph with the provided source (HTTP PUT operation)
        :param graph_uri: URI for the graph (if None, the default graph will be used)
        :param source: Source data or file name
        :param format: Media type to provide to the Graph Store
        :return:
        """
        self._post_or_put('put', graph_uri, source, format)

    def add(self, graph_uri: str | None, source: str | bytes | Path, format: str = 'text/turtle'):
        """
        Adds data from the provided source to a Graph Store graph (HTTP PUT operation)
        :param graph_uri: URI for the graph (if None, the default graph will be used)
        :param source: Source data or file name
        :param format: Media type to provide to the Graph Store
        :return:
        """
        self._post_or_put('post', graph_uri, source, format)

    def get(self, graph_uri) -> Graph:
        """
        Retrieves the data from a graph in the Graph Store
        :param graph_uri: URI for the graph (if None, the default graph will be used)
        :return: An [RDFLib Graph][rdflib.Graph]
        """
        if graph_uri:
            params = {'graph': str(graph_uri)}
        else:
            params = 'default'

        r = requests.get(
            self._url,
            params=params,
            auth=self.auth_details,
        )
        r.raise_for_status()
        g = Graph().parse(r.content)
        return g

__init__(url, auth_details=None)

Constructs a new GraphStore

Parameters:

Name Type Description Default
url str

SPARQL Graph Store Protocol URL

 required
auth_details tuple[str, str] | None

tuple in the form ('username', 'password') for authentication

 None
Source code in ogc/na/gsp.py 
22
23
24
25
26
27
28
29
30
31
def __init__(self, url: str, auth_details: tuple[str, str] | None = None):
    """
    Constructs a new GraphStore
    :param url: SPARQL Graph Store Protocol URL
    :param auth_details: tuple in the form ('username', 'password') for authentication
    """
    self._url = url
    self.auth_details = auth_details
    self.put = self.replace
    self.post = self.add

add(graph_uri, source, format='text/turtle')

Adds data from the provided source to a Graph Store graph (HTTP PUT operation)

Parameters:

Name Type Description Default
graph_uri str | None

URI for the graph (if None, the default graph will be used)

 required
source str | bytes | Path

Source data or file name

 required
format str

Media type to provide to the Graph Store

 'text/turtle'

Returns:

Type Description
Source code in ogc/na/gsp.py 
 96
 97
 98
 99
100
101
102
103
104
def add(self, graph_uri: str | None, source: str | bytes | Path, format: str = 'text/turtle'):
    """
    Adds data from the provided source to a Graph Store graph (HTTP PUT operation)
    :param graph_uri: URI for the graph (if None, the default graph will be used)
    :param source: Source data or file name
    :param format: Media type to provide to the Graph Store
    :return:
    """
    self._post_or_put('post', graph_uri, source, format)

delete(graph_uri, ignore_404=True)

Deletes a graph from the Graph Store

Parameters:

Name Type Description Default
graph_uri str | None

URI for the graph (if None, the default graph will be used)

 required
ignore_404

Whether to ignore HTTP 404 errors (otherwise, an Exception will be thrown)

 True

Returns:

Type Description
Source code in ogc/na/gsp.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def delete(self, graph_uri: str | None, ignore_404=True):
    """
    Deletes a graph from the Graph Store
    :param graph_uri: URI for the graph (if None, the default graph will be used)
    :param ignore_404: Whether to ignore HTTP 404 errors (otherwise, an Exception will be thrown)
    :return:
    """
    if graph_uri:
        params = {'graph': str(graph_uri)}
    else:
        params = 'default'

    r = requests.delete(
        self._url,
        params=params,
        auth=self.auth_details,
    )
    if not (ignore_404 and r.status_code == 404):
        r.raise_for_status()

get(graph_uri)

Retrieves the data from a graph in the Graph Store

Parameters:

Name Type Description Default
graph_uri

URI for the graph (if None, the default graph will be used)

 required

Returns:

Type Description
Graph

An RDFLib Graph
Source code in ogc/na/gsp.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def get(self, graph_uri) -> Graph:
    """
    Retrieves the data from a graph in the Graph Store
    :param graph_uri: URI for the graph (if None, the default graph will be used)
    :return: An [RDFLib Graph][rdflib.Graph]
    """
    if graph_uri:
        params = {'graph': str(graph_uri)}
    else:
        params = 'default'

    r = requests.get(
        self._url,
        params=params,
        auth=self.auth_details,
    )
    r.raise_for_status()
    g = Graph().parse(r.content)
    return g

replace(graph_uri, source, format='text/turtle')

Replaces the data for a Graph Store graph with the provided source (HTTP PUT operation)

Parameters:

Name Type Description Default
graph_uri str | None

URI for the graph (if None, the default graph will be used)

 required
source str | bytes | Path

Source data or file name

 required
format str

Media type to provide to the Graph Store

 'text/turtle'

Returns:

Type Description
Source code in ogc/na/gsp.py 
86
87
88
89
90
91
92
93
94
def replace(self, graph_uri: str | None, source: str | bytes | Path, format: str = 'text/turtle'):
    """
    Replaces the data for a Graph Store graph with the provided source (HTTP PUT operation)
    :param graph_uri: URI for the graph (if None, the default graph will be used)
    :param source: Source data or file name
    :param format: Media type to provide to the Graph Store
    :return:
    """
    self._post_or_put('put', graph_uri, source, format)