Skip to content

# `` hext

This is a rdflib plugin for parsing Hextuple files, which are Newline-Delimited JSON (ndjson) files, into Conjunctive. The store that backs the graph must be able to handle contexts, i.e. multiple graphs.

Classes:

- `HextuplesParser` –

  An RDFLib parser for Hextuples

## `` __all__ `module-attribute`

    __all__ = ['HextuplesParser']

## `` HextuplesParser

    HextuplesParser()

Bases: `Parser`

An RDFLib parser for Hextuples

Methods:

- `parse` –

Attributes:

- `default_context` (`Optional[Graph]`) –
- `skolemize` –

Source code in `rdflib/plugins/parsers/hext.py`

    37
    38
    39
    40

    def __init__(self):
        super(HextuplesParser, self).__init__()
        self.default_context: Optional[Graph] = None
        self.skolemize = False

### `` default_context `instance-attribute`

    default_context: Optional[Graph] = None

### `` skolemize `instance-attribute`

    skolemize = False

### `` parse

    parse(source: InputSource, graph: Graph, skolemize: bool = False, **kwargs: Any) -> None

Source code in `rdflib/plugins/parsers/hext.py`

     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
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172

    def parse(self, source: InputSource, graph: Graph, skolemize: bool = False, **kwargs: Any) -> None:  # type: ignore[override]
        if kwargs.get("encoding") not in [None, "utf-8"]:
            warnings.warn(
                f"Hextuples files are always utf-8 encoded, "
                f"I was passed: {kwargs.get('encoding')}, "
                "but I'm still going to use utf-8"
            )

        assert (
            graph.store.context_aware
        ), "Hextuples Parser needs a context-aware store!"

        self.skolemize = skolemize
        # Set default_union to True to mimic ConjunctiveGraph behavior
        ds = Dataset(store=graph.store, default_union=True)
        ds_default = ds.default_context  # the DEFAULT_DATASET_GRAPH_ID
        if isinstance(graph, (Dataset, ConjunctiveGraph)):
            self.default_context = graph.default_context
        elif graph.identifier is not None:
            if graph.identifier == ds_default.identifier:
                self.default_context = graph
            else:
                self.default_context = ds.get_context(graph.identifier)
        else:
            # mypy thinks this is unreachable, but graph.identifier can be None
            self.default_context = ds_default  # type: ignore[unreachable]
        if self.default_context is not ds_default:
            ds.default_context = self.default_context
            ds.remove_graph(ds_default)  # remove the original unused default graph

        try:
            text_stream: Optional[TextIO] = source.getCharacterStream()
        except (AttributeError, LookupError):
            text_stream = None
        try:
            binary_stream: Optional[BinaryIO] = source.getByteStream()
        except (AttributeError, LookupError):
            binary_stream = None

        if text_stream is None and binary_stream is None:
            raise ValueError(
                f"Source does not have a character stream or a byte stream and cannot be used {type(source)}"
            )
        if TYPE_CHECKING:
            assert text_stream is not None or binary_stream is not None
        use_stream: Union[TextIO, BinaryIO]
        if _HAS_ORJSON:
            if binary_stream is not None:
                use_stream = binary_stream
            else:
                if TYPE_CHECKING:
                    assert isinstance(text_stream, TextIOWrapper)
                use_stream = text_stream
            loads = orjson.loads
        else:
            if text_stream is not None:
                use_stream = text_stream
            else:
                if TYPE_CHECKING:
                    assert isinstance(binary_stream, BufferedReader)
                use_stream = TextIOWrapper(binary_stream, encoding="utf-8")
            loads = json.loads

        for line in use_stream:  # type: Union[str, bytes]
            if len(line) == 0 or line.isspace():
                # Skipping empty lines because this is what was being done before for the first and last lines, albeit in an rather indirect way.
                # The result is that we accept input that would otherwise be invalid.
                # Possibly we should just let this result in an error.
                continue
            # this complex handing is because the 'value' component is
            # allowed to be "" but not None
            # all other "" values are treated as None
            raw_line: List[str] = loads(line)
            hex_tuple_line = [x if x != "" else None for x in raw_line]
            if raw_line[2] == "":
                hex_tuple_line[2] = ""
            self._parse_hextuple(ds, hex_tuple_line)