Skip to content

# `` util

RDF4J utility functions.

Functions:

- `build_context_param` –

  Build the RDF4J http context query parameters dictionary.

- `build_infer_param` –

  Build the RDF4J http infer query parameters dictionary.

- `build_sparql_query_accept_header` –

  Build the SPARQL query accept header.

- `build_spo_param` –

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

- `rdf_payload_to_stream` –

  Convert an RDF payload into a file-like object.

- `validate_graph_name` –
- `validate_no_bnodes` –

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

## `` 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)