Skip to content

util

RDF4J utility functions.

Functions:

build_context_param 

build_context_param(params: dict[str, str], graph_name: IdentifiedNode | Iterable[IdentifiedNode] | str | None = None) -> None

Build the RDF4J http context query parameters dictionary.

Note

This mutates the params dictionary key context.

Parameters:

  • params

    (dict[str, str]) –

    The httpx.Request parameter dictionary.

  • graph_name

    (IdentifiedNode | Iterable[IdentifiedNode] | str | None, default: None ) –

    The graph name or iterable of graph names.

    This is the context query parameter value.

Source code in rdflib/contrib/rdf4j/util.py 
16
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
def build_context_param(
    params: dict[str, str],
    graph_name: IdentifiedNode | t.Iterable[IdentifiedNode] | str | None = None,
) -> None:
    """Build the RDF4J http context query parameters dictionary.

    !!! Note
        This mutates the params dictionary key `context`.

    Parameters:
        params: The `httpx.Request` parameter dictionary.
        graph_name: The graph name or iterable of graph names.

            This is the `context` query parameter value.
    """
    if graph_name is not None and isinstance(graph_name, IdentifiedNode):
        if graph_name == DATASET_DEFAULT_GRAPH_ID:
            # Special RDF4J null value for context-less statements.
            params["context"] = "null"
        else:
            params["context"] = graph_name.n3()
    elif graph_name is not None and isinstance(graph_name, str):
        params["context"] = URIRef(graph_name).n3()
    elif graph_name is not None and isinstance(graph_name, t.Iterable):
        # type error: "str" has no attribute "n3"
        graph_names = ",".join([x.n3() for x in graph_name])  # type: ignore[attr-defined]
        params["context"] = graph_names

build_infer_param 

build_infer_param(params: dict[str, str], infer: bool = True) -> None

Build the RDF4J http infer query parameters dictionary.

Note

This mutates the params dictionary key infer.

Parameters:

  • params

    (dict[str, str]) –

    The httpx.Request parameter dictionary.

  • infer

    (bool, default: True ) –

    The infer query parameter value.

Source code in rdflib/contrib/rdf4j/util.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def build_infer_param(
    params: dict[str, str],
    infer: bool = True,
) -> None:
    """Build the RDF4J http infer query parameters dictionary.

    !!! Note
        This mutates the params dictionary key `infer`.

    Parameters:
        params: The `httpx.Request` parameter dictionary.
        infer: The `infer` query parameter value.
    """
    if not infer:
        params["infer"] = "false"

build_sparql_query_accept_header 

build_sparql_query_accept_header(query: str, headers: dict[str, str])

Build the SPARQL query accept header.

Note

This mutates the headers dictionary key Accept.

Parameters:

  • query

    (str) –

    The SPARQL query.

Source code in rdflib/contrib/rdf4j/util.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def build_sparql_query_accept_header(query: str, headers: dict[str, str]):
    """Build the SPARQL query accept header.

    !!! Note
        This mutates the headers dictionary key `Accept`.

    Parameters:
        query: The SPARQL query.
    """
    prepared_query = prepareQuery(query)
    if prepared_query.algebra.name in ("SelectQuery", "AskQuery"):
        headers["Accept"] = "application/sparql-results+json"
    elif prepared_query.algebra.name in ("ConstructQuery", "DescribeQuery"):
        headers["Accept"] = "application/n-triples"
    else:
        raise ValueError(f"Unsupported query type: {prepared_query.algebra.name}")

build_spo_param 

build_spo_param(params: dict[str, str], subj: SubjectType = None, pred: PredicateType = None, obj: ObjectType = None) -> None

Build the RDF4J http subj, predicate, and object query parameters dictionary.

Note

This mutates the params dictionary key subj, pred, and obj.

Parameters:

  • params

    (dict[str, str]) –

    The httpx.Request parameter dictionary.

  • subj

    (SubjectType, default: None ) –

    The subj query parameter value.

  • pred

    (PredicateType, default: None ) –

    The pred query parameter value.

  • obj

    (ObjectType, default: None ) –

    The obj query parameter value.

Source code in rdflib/contrib/rdf4j/util.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def build_spo_param(
    params: dict[str, str],
    subj: SubjectType = None,
    pred: PredicateType = None,
    obj: ObjectType = None,
) -> None:
    """Build the RDF4J http subj, predicate, and object query parameters dictionary.

    !!! Note
        This mutates the params dictionary key `subj`, `pred`, and `obj`.

    Parameters:
        params: The `httpx.Request` parameter dictionary.
        subj: The `subj` query parameter value.
        pred: The `pred` query parameter value.
        obj: The `obj` query parameter value.
    """
    if subj is not None:
        params["subj"] = subj.n3()
    if pred is not None:
        params["pred"] = pred.n3()
    if obj is not None:
        params["obj"] = obj.n3()

rdf_payload_to_stream 

rdf_payload_to_stream(data: str | bytes | BinaryIO | Graph | Dataset) -> tuple[BinaryIO, bool]

Convert an RDF payload into a file-like object.

Parameters:

  • data

    (str | bytes | BinaryIO | Graph | Dataset) –

    The RDF payload.

    This can be a python str, bytes, BinaryIO, or a Graph or Dataset.

Returns:

  • BinaryIO –

    A tuple containing the file-like object and a boolean indicating whether the

  • bool –

    immediate caller should close the stream.

Source code in rdflib/contrib/rdf4j/util.py 
 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
def rdf_payload_to_stream(
    data: str | bytes | t.BinaryIO | Graph | Dataset,
) -> tuple[t.BinaryIO, bool]:
    """Convert an RDF payload into a file-like object.

    Parameters:
        data: The RDF payload.

            This can be a python `str`, `bytes`, `BinaryIO`, or a
            [`Graph`][rdflib.graph.Graph] or [`Dataset`][rdflib.graph.Dataset].

    Returns:
        A tuple containing the file-like object and a boolean indicating whether the
        immediate caller should close the stream.
    """
    stream: t.BinaryIO
    if isinstance(data, str):
        # Check if it looks like a file path. Assumes file path length is less than 260.
        if "\n" not in data and len(data) < 260:
            try:
                stream = open(data, "rb")
                should_close = True
            except (FileNotFoundError, OSError):
                # Treat as raw string content
                stream = io.BytesIO(data.encode("utf-8"))
                should_close = False
        else:
            # Treat as raw string content
            stream = io.BytesIO(data.encode("utf-8"))
            should_close = False
    elif isinstance(data, bytes):
        stream = io.BytesIO(data)
        should_close = False
    elif isinstance(data, (Graph, Dataset)):
        if data.context_aware:
            stream = io.BytesIO(
                data.serialize(format="application/n-quads", encoding="utf-8")
            )
        else:
            stream = io.BytesIO(
                data.serialize(format="application/n-triples", encoding="utf-8")
            )
        should_close = True
    else:
        # Assume it's already a file-like object
        stream = data
        should_close = False

    return stream, should_close

validate_graph_name 

validate_graph_name(graph_name: URIRef | Iterable[URIRef] | str | None)
Source code in rdflib/contrib/rdf4j/util.py 
156
157
158
159
160
161
162
def validate_graph_name(graph_name: URIRef | t.Iterable[URIRef] | str | None):
    if (
        isinstance(graph_name, BNode)
        or isinstance(graph_name, t.Iterable)
        and any(isinstance(x, BNode) for x in graph_name)
    ):
        raise ValueError("Graph name must not be a BNode.")

validate_no_bnodes 

validate_no_bnodes(subj: SubjectType, pred: PredicateType, obj: ObjectType, graph_name: URIRef | Iterable[URIRef] | str | None) -> None

Validate that the subject, predicate, and object are not BNodes.

Source code in rdflib/contrib/rdf4j/util.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def validate_no_bnodes(
    subj: SubjectType,
    pred: PredicateType,
    obj: ObjectType,
    graph_name: URIRef | t.Iterable[URIRef] | str | None,
) -> None:
    """Validate that the subject, predicate, and object are not BNodes."""
    if (
        isinstance(subj, BNode)
        or isinstance(pred, BNode)
        or isinstance(obj, BNode)
        or isinstance(graph_name, BNode)
    ):
        raise ValueError(
            "Subject, predicate, and object must not be a BNode: "
            f"{subj}, {pred}, {obj}"
        )
    validate_graph_name(graph_name)