API reference

pyjelly

Modules:

Name	Description
errors
integrations
jelly
options
parse
serialize

errors

Classes:

Name	Description
JellyConformanceError	Raised when Jelly conformance is violated.
JellyAssertionError	Raised when a recommended assertion from the specification fails.
JellyNotImplementedError	Raised when a future feature is not yet implemented.
JellyIncompleteResultsError	Raised when the producer of SPARQL results reports that they are incomplete.
JellyTruncatedResultsWarning	Issued when a stream of SPARQL results ends without a trailer.

JellyConformanceError

Bases: Exception

Raised when Jelly conformance is violated.

JellyAssertionError

Bases: AssertionError

Raised when a recommended assertion from the specification fails.

JellyNotImplementedError

Bases: NotImplementedError

Raised when a future feature is not yet implemented.

JellyIncompleteResultsError(message, error, result=None)

Bases: Exception

Raised when the producer of SPARQL results reports that they are incomplete.

For example, the query failed midway. The rows read before the error are valid.

Source code in pyjelly/errors.py

def __init__(self, message: str, error: str, result: object = None) -> None:
    super().__init__(message)
    #: Error message from the producer
    self.error = error
    #: Results read before the error, if the reader keeps them
    self.result = result

JellyTruncatedResultsWarning

Bases: UserWarning

Issued when a stream of SPARQL results ends without a trailer.

Some results may be missing, e.g., because the writer stopped or the connection was lost.

integrations

Modules:

Name	Description
generic
rdflib

generic

Modules:

Name	Description
decode	Adapters that make generic terms and statements from decoded Jelly rows.
generic_sink
parse
serialize
sparql	Jelly-SPARQL support for the generic API (without rdflib).

decode

Adapters that make generic terms and statements from decoded Jelly rows.

Compiled with mypyc, like the decoder that calls them.

Classes:

Name	Description
GenericStatementSinkAdapter	Implement Adapter for generic statements.
GenericTriplesAdapter	Triples adapted implementation for GenericStatementSink.
GenericQuadsAdapter	Extends GenericQuadsBaseAdapter for QUADS physical type.
GenericGraphsAdapter	Extends GenericQuadsBaseAdapter for GRAPHS physical type.

GenericStatementSinkAdapter(options, parsing_mode=ParsingMode.FLAT)

Bases: Adapter

Implement Adapter for generic statements.

Notes: Returns custom RDF terms expected by GenericStatementSink, handles namespace declarations, and quoted triples.

Args: Adapter (type): base Adapter class

Source code in pyjelly/parse/decode.py

def __init__(
    self, options: ParserOptions, parsing_mode: ParsingMode = ParsingMode.FLAT
) -> None:
    self.options = options
    self.parsing_mode = parsing_mode

GenericTriplesAdapter(options)

Bases: GenericStatementSinkAdapter

Triples adapted implementation for GenericStatementSink.

Args: GenericStatementSinkAdapter (type): base GenericStatementSink adapter implementation that handles terms and namespaces.

Source code in pyjelly/integrations/generic/decode.py

def __init__(
    self,
    options: ParserOptions,
) -> None:
    super().__init__(options=options)

GenericQuadsAdapter(options)

Bases: GenericQuadsBaseAdapter

Extends GenericQuadsBaseAdapter for QUADS physical type.

Args: GenericQuadsBaseAdapter (type): quads adapter that handles base quads processing.

Source code in pyjelly/integrations/generic/decode.py

def __init__(self, options: ParserOptions) -> None:
    super().__init__(options=options)

GenericGraphsAdapter(options)

Bases: GenericQuadsBaseAdapter

Extends GenericQuadsBaseAdapter for GRAPHS physical type.

Notes: introduces graph start/end, checks if graph exists.

Args: GenericQuadsBaseAdapter (type): quads adapter that handles base quads processing.

Raises: JellyConformanceError: raised if graph start message was not received.

Source code in pyjelly/integrations/generic/decode.py

def __init__(
    self,
    options: ParserOptions,
) -> None:
    super().__init__(options=options)
    self._graph_id = None

generic_sink

Classes:

Name	Description
BlankNode	Class for blank nodes, storing BN's identifier as a string.
IRI	Class for IRIs, storing IRI as a string.
Literal	Class for literals.
Triple	Class for RDF triples.
Quad	Class for RDF quads.
Prefix	Class for generic namespace declaration.
GenericStatementSink	A graph or dataset of generic statements, kept in memory in order.

Attributes:

Name	Type	Description
DefaultGraph		The name of the default graph, used as the graph name of quads in it.

DefaultGraph = _DefaultGraph()

The name of the default graph, used as the graph name of quads in it.

BlankNode(identifier)

Class for blank nodes, storing BN's identifier as a string.

Attributes:

Name	Type	Description
identifier	str	The identifier of the blank node, without the _: prefix.

Source code in pyjelly/integrations/generic/generic_sink.py

def __init__(self, identifier: str) -> None:
    # Not `_identifier`: mypyc would name its C field `__identifier`, which
    # is a reserved word for the MSVC compiler
    self._id: str = identifier

identifier

The identifier of the blank node, without the _: prefix.

IRI(iri)

Class for IRIs, storing IRI as a string.

Attributes:

Name	Type	Description
iri	str	The IRI as a string, without angle brackets.

Source code in pyjelly/integrations/generic/generic_sink.py

def __init__(self, iri: str) -> None:
    self._iri: str = iri

iri

The IRI as a string, without angle brackets.

Literal(lex, langtag=None, datatype=None, direction=None)

Class for literals.

Notes: Consists of: lexical form, and optional language tag and datatype. A language-tagged string may also have a base direction, "ltr" or "rtl" (RDF 1.2). All parts of literal are stored as strings.

Attributes:

Name	Type	Description
lex	str	The lexical form, without quotes or escapes.
langtag	str | None	The language tag, or None if there is none.
datatype	str | None	The datatype IRI as a string, or None if it is not given.
direction	str | None	The base direction, "ltr" or "rtl", or None if there is none.

Source code in pyjelly/integrations/generic/generic_sink.py

def __init__(
    self,
    lex: str,
    langtag: str | None = None,
    datatype: str | None = None,
    direction: str | None = None,
) -> None:
    self._lex: str = lex
    self._langtag: str | None = langtag
    self._datatype: str | None = datatype
    self._direction: str | None = direction

lex

The lexical form, without quotes or escapes.

langtag

The language tag, or None if there is none.

datatype

The datatype IRI as a string, or None if it is not given.

direction

The base direction, "ltr" or "rtl", or None if there is none.

Triple

Bases: NamedTuple

Class for RDF triples.

Attributes:

Name	Type	Description
s	Node	The subject.
p	Node	The predicate.
o	Node	The object.

s

The subject.

p

The predicate.

o

The object.

Quad

Bases: NamedTuple

Class for RDF quads.

Attributes:

Name	Type	Description
s	Node	The subject.
p	Node	The predicate.
o	Node	The object.
g	GraphName	The graph name, DefaultGraph for the default graph.

s

The subject.

p

The predicate.

o

The object.

g

The graph name, DefaultGraph for the default graph.

Prefix

Bases: NamedTuple

Class for generic namespace declaration.

Attributes:

Name	Type	Description
prefix	str	The prefix.
iri	IRI	The namespace IRI.

prefix

The prefix.

iri

The namespace IRI.

GenericStatementSink(identifier=DefaultGraph)

A graph or dataset of generic statements, kept in memory in order.

It holds triples or quads, and the namespaces bound to it.

Notes: _store preserves the order of statements.

Args: identifier (str, optional): Identifier for a sink. Defaults to DefaultGraph.

Methods:

Name	Description
add	Add a statement at the end of the sink.
bind	Bind a prefix to a namespace, replacing an earlier binding of it.
parse	Read a Jelly stream into the sink.
serialize	Write the sink as a Jelly stream, with the default options.

Attributes:

Name	Type	Description
namespaces	Generator[tuple[str, IRI]]	The bound namespaces, as (prefix, namespace IRI) pairs.
identifier	GraphName	The name of the graph, DefaultGraph if it is not given.
store	Generator[Triple | Quad]	The statements, in the order they were added.
is_triples_sink	bool	Check if the sink contains triples or quads.

Source code in pyjelly/integrations/generic/generic_sink.py

def __init__(self, identifier: GraphName = DefaultGraph) -> None:
    """
    Initialize statements storage, namespaces dictionary, and parser.

    Notes:
        _store preserves the order of statements.

    Args:
        identifier (str, optional): Identifier for a sink.
            Defaults to DefaultGraph.

    """
    self._store: deque[Triple | Quad] = deque()
    self._namespaces: dict[str, IRI] = {}
    # Not `_identifier`, see BlankNode
    self._id = identifier

namespaces

The bound namespaces, as (prefix, namespace IRI) pairs.

identifier

The name of the graph, DefaultGraph if it is not given.

store

The statements, in the order they were added.

is_triples_sink

Check if the sink contains triples or quads.

Returns: bool: true, if length of statement is 3.

add(statement)

Add a statement at the end of the sink.

Source code in pyjelly/integrations/generic/generic_sink.py

def add(self, statement: Triple | Quad) -> None:
    """Add a statement at the end of the sink."""
    self._store.append(statement)

bind(prefix, namespace)

Bind a prefix to a namespace, replacing an earlier binding of it.

Source code in pyjelly/integrations/generic/generic_sink.py

def bind(self, prefix: str, namespace: IRI) -> None:
    """Bind a prefix to a namespace, replacing an earlier binding of it."""
    self._namespaces.update({prefix: namespace})

parse(input_file)

Read a Jelly stream into the sink.

The statements, namespaces and identifier of the sink are replaced with those read from the stream.

Args: input_file (IO[bytes]): binary file to read.

Source code in pyjelly/integrations/generic/generic_sink.py

def parse(self, input_file: IO[bytes]) -> None:
    """
    Read a Jelly stream into the sink.

    The statements, namespaces and identifier of the sink are replaced with
    those read from the stream.

    Args:
        input_file (IO[bytes]): binary file to read.

    """
    from pyjelly.integrations.generic.parse import (  # noqa: PLC0415
        parse_jelly_to_graph,
    )

    parsed_result = parse_jelly_to_graph(input_file)
    self._store = parsed_result._store
    self._namespaces = parsed_result._namespaces
    self._id = parsed_result._id

serialize(output_file)

Write the sink as a Jelly stream, with the default options.

To choose the options, use flat_stream_to_file or grouped_stream_to_file from pyjelly.integrations.generic.serialize.

Args: output_file (IO[bytes]): binary file to write.

Source code in pyjelly/integrations/generic/generic_sink.py

def serialize(self, output_file: IO[bytes]) -> None:
    """
    Write the sink as a Jelly stream, with the default options.

    To choose the options, use `flat_stream_to_file` or
    `grouped_stream_to_file` from `pyjelly.integrations.generic.serialize`.

    Args:
        output_file (IO[bytes]): binary file to write.

    """
    from pyjelly.integrations.generic.serialize import (  # noqa: PLC0415
        grouped_stream_to_file,
    )

    grouped_stream_to_file((sink for sink in [self]), output_file)

parse

Functions:

Name	Description
parse_triples_stream	Parse flat triple stream.
parse_quads_stream	Parse flat quads stream.
parse_jelly_grouped	Take a jelly file and return generators of generic statements sinks.
parse_jelly_to_graph	Add statements from Generator to GenericStatementSink.
parse_jelly_flat	Parse jelly file with FLAT logical type into a Generator of stream events.

parse_triples_stream(frames, options, frame_metadata=None)

Parse flat triple stream.

Args: frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames options (ParserOptions): stream options frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable used for extracting frame metadata

Yields: Generator[Iterable[Triple | Prefix]]: Generator of iterables of Triple or Prefix objects, one iterable per frame.

Source code in pyjelly/integrations/generic/parse.py

def parse_triples_stream(
    frames: Iterable[jelly.RdfStreamFrame],
    options: ParserOptions,
    frame_metadata: ContextVar[MutableMapping[str, bytes]] | None = None,
) -> Generator[Iterable[Triple | Prefix]]:
    """
    Parse flat triple stream.

    Args:
        frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames
        options (ParserOptions): stream options
        frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable
                used for extracting frame metadata

    Yields:
        Generator[Iterable[Triple | Prefix]]:
            Generator of iterables of Triple or Prefix objects,
            one iterable per frame.

    """
    adapter = GenericTriplesAdapter(options)
    decoder = Decoder(adapter=adapter)
    for frame in frames:
        if frame_metadata is not None:
            frame_metadata.set(
                frame.metadata
            ) if frame.metadata else frame_metadata.set({})
        yield decoder.iter_rows(frame)
    if options.params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES:
        yield decoder.end_of_stream()

parse_quads_stream(frames, options, frame_metadata=None)

Parse flat quads stream.

Args: frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames options (ParserOptions): stream options frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable used for extracting frame metadata

Yields: Generator[Iterable[Quad | Prefix]]: Generator of iterables of Quad or Prefix objects, one iterable per frame.

Source code in pyjelly/integrations/generic/parse.py

def parse_quads_stream(
    frames: Iterable[jelly.RdfStreamFrame],
    options: ParserOptions,
    frame_metadata: ContextVar[MutableMapping[str, bytes]] | None = None,
) -> Generator[Iterable[Quad | Prefix]]:
    """
    Parse flat quads stream.

    Args:
        frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames
        options (ParserOptions): stream options
        frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable
                used for extracting frame metadata

    Yields:
        Generator[Iterable[Quad | Prefix]]:
            Generator of iterables of Quad or Prefix objects,
            one iterable per frame.

    """
    adapter_class: type[GenericQuadsBaseAdapter]
    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_QUADS:
        adapter_class = GenericQuadsAdapter
    else:
        adapter_class = GenericGraphsAdapter
    adapter = adapter_class(options=options)
    decoder = Decoder(adapter=adapter)
    for frame in frames:
        if frame_metadata is not None:
            frame_metadata.set(
                frame.metadata
            ) if frame.metadata else frame_metadata.set({})
        yield decoder.iter_rows(frame)
    if options.params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES:
        yield decoder.end_of_stream()

parse_jelly_grouped(inp, sink_factory=GenericStatementSink, *, logical_type_strict=False, frame_metadata=None)

Take a jelly file and return generators of generic statements sinks.

Yields one generic statements sink per frame.

Args: inp (IO[bytes]): input jelly buffered binary stream sink_factory (Callable): lambda to construct a statement sink. By default, creates an empty in-memory GenericStatementSink. logical_type_strict (bool): If True, validate the logical type in stream options and require a grouped logical type. Otherwise, only the physical type is used to route parsing. frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable used for extracting frame metadata

Raises: NotImplementedError: is raised if a physical type is not implemented

Yields: Generator[GenericStatementSink]: returns generators for GenericStatementSink, regardless of stream type.

Source code in pyjelly/integrations/generic/parse.py

def parse_jelly_grouped(
    inp: IO[bytes],
    sink_factory: Callable[[], GenericStatementSink] = GenericStatementSink,
    *,
    logical_type_strict: bool = False,
    frame_metadata: ContextVar[MutableMapping[str, bytes]] | None = None,
) -> Generator[GenericStatementSink]:
    """
    Take a jelly file and return generators of generic statements sinks.

    Yields one generic statements sink per frame.

    Args:
        inp (IO[bytes]): input jelly buffered binary stream
        sink_factory (Callable): lambda to construct a statement sink.
            By default, creates an empty in-memory GenericStatementSink.
        logical_type_strict (bool): If True, validate the *logical* type
            in stream options and require a grouped logical type.
            Otherwise, only the physical type is used to route parsing.
        frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable
                used for extracting frame metadata

    Raises:
        NotImplementedError: is raised if a physical type is not implemented

    Yields:
        Generator[GenericStatementSink]:
            returns generators for GenericStatementSink, regardless of stream type.

    """
    options, frames = get_options_and_frames(inp)

    st = getattr(options, "stream_types", None)
    if logical_type_strict and (
        st is None
        or st.logical_type == jelly.LOGICAL_STREAM_TYPE_UNSPECIFIED
        or st.flat
    ):
        lt_name = (
            "UNSPECIFIED"
            if st is None
            else jelly.LogicalStreamType.Name(st.logical_type)
        )

        msg = (
            "strict logical type check requires options.stream_types"
            if st is None
            else f"expected GROUPED logical type, got {lt_name}"
        )
        raise JellyConformanceError(msg)

    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_TRIPLES:
        for graph in grouped_rows(
            parse_triples_stream(
                frames=frames, options=options, frame_metadata=frame_metadata
            ),
            options,
        ):
            sink = sink_factory()
            for graph_item in graph:
                if isinstance(graph_item, Prefix):
                    sink.bind(graph_item.prefix, graph_item.iri)
                else:
                    sink.add(graph_item)
            yield sink
        return
    elif options.stream_types.physical_type in (
        jelly.PHYSICAL_STREAM_TYPE_QUADS,
        jelly.PHYSICAL_STREAM_TYPE_GRAPHS,
    ):
        for dataset in grouped_rows(
            parse_quads_stream(
                frames=frames, options=options, frame_metadata=frame_metadata
            ),
            options,
        ):
            sink = sink_factory()
            for item in dataset:
                if isinstance(item, Prefix):
                    sink.bind(item.prefix, item.iri)
                else:
                    sink.add(item)
            yield sink
        return

    physical_type_name = jelly.PhysicalStreamType.Name(
        options.stream_types.physical_type
    )
    msg = f"the stream type {physical_type_name} is not supported "
    raise NotImplementedError(msg)

parse_jelly_to_graph(inp, sink_factory=GenericStatementSink)

Add statements from Generator to GenericStatementSink.

Args: inp (IO[bytes]): input jelly stream. sink_factory (Callable[[], GenericStatementSink]): factory to create statement sink. By default creates an empty in-memory GenericStatementSink. Has no division for datasets/graphs, utilizes the same underlying data structures.

Returns: GenericStatementSink: GenericStatementSink with statements.

Source code in pyjelly/integrations/generic/parse.py

def parse_jelly_to_graph(
    inp: IO[bytes],
    sink_factory: Callable[[], GenericStatementSink] = GenericStatementSink,
) -> GenericStatementSink:
    """
    Add statements from Generator to GenericStatementSink.

    Args:
        inp (IO[bytes]): input jelly stream.
        sink_factory (Callable[[], GenericStatementSink]): factory to create
            statement sink.
            By default creates an empty in-memory GenericStatementSink.
            Has no division for datasets/graphs,
            utilizes the same underlying data structures.

    Returns:
        GenericStatementSink: GenericStatementSink with statements.

    """
    options, frames = get_options_and_frames(inp)
    sink = sink_factory()

    for item in parse_jelly_flat(
        inp=inp, frames=frames, options=options, logical_type_strict=False
    ):
        if isinstance(item, Prefix):
            sink.bind(item.prefix, item.iri)  # type: ignore[union-attr, unused-ignore]
        else:
            sink.add(item)
    return sink

parse_jelly_flat(inp, frames=None, options=None, *, logical_type_strict=False)

Parse jelly file with FLAT logical type into a Generator of stream events.

Args: inp (IO[bytes]): input jelly buffered binary stream. frames (Iterable[jelly.RdfStreamFrame | None): jelly frames if read before. options (ParserOptions | None): stream options if read before. logical_type_strict (bool): If True, validate the logical type in stream options and require FLAT (TRIPLES/QUADS). Otherwise, only the physical type is used to route parsing.

Raises: NotImplementedError: if physical type is not supported

Yields: Generator[Statement | Prefix]: Generator of stream events

Source code in pyjelly/integrations/generic/parse.py

def parse_jelly_flat(
    inp: IO[bytes],
    frames: Iterable[jelly.RdfStreamFrame] | None = None,
    options: ParserOptions | None = None,
    *,
    logical_type_strict: bool = False,
) -> Generator[Statement | Prefix]:  # type: ignore[valid-type, unused-ignore]
    """
    Parse jelly file with FLAT logical type into a Generator of stream events.

    Args:
        inp (IO[bytes]): input jelly buffered binary stream.
        frames (Iterable[jelly.RdfStreamFrame | None):
            jelly frames if read before.
        options (ParserOptions | None): stream options
            if read before.
        logical_type_strict (bool): If True, validate the *logical* type
            in stream options and require FLAT (TRIPLES/QUADS).
            Otherwise, only the physical type is used to route parsing.

    Raises:
        NotImplementedError: if physical type is not supported

    Yields:
        Generator[Statement | Prefix]: Generator of stream events

    """
    if frames is None or options is None:
        options, frames = get_options_and_frames(inp)

    st = getattr(options, "stream_types", None)
    if logical_type_strict and (st is None or not st.flat):
        lt_name = (
            "UNSPECIFIED"
            if st is None
            else jelly.LogicalStreamType.Name(st.logical_type)
        )

        msg = (
            "strict logical type check requires options.stream_types"
            if st is None
            else f"expected FLAT logical type (TRIPLES/QUADS), got {lt_name}"
        )
        raise JellyConformanceError(msg)

    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_TRIPLES:
        yield from flat_rows(
            parse_triples_stream(frames=frames, options=options), options
        )
        return
    if options.stream_types.physical_type in (
        jelly.PHYSICAL_STREAM_TYPE_QUADS,
        jelly.PHYSICAL_STREAM_TYPE_GRAPHS,
    ):
        yield from flat_rows(
            parse_quads_stream(frames=frames, options=options), options
        )
        return
    physical_type_name = jelly.PhysicalStreamType.Name(
        options.stream_types.physical_type
    )
    msg = f"the stream type {physical_type_name} is not supported "
    raise NotImplementedError(msg)

serialize

Classes:

Name	Description
GenericSinkTermEncoder

Functions:

Name	Description
triples_stream_frames	Serialize a GenericStatementSink into frames using physical type triples stream.
quads_stream_frames	Serialize a GenericStatementSink into jelly frames using physical type quads stream.
guess_options	Guess the serializer options based on the store type.
guess_stream	Return an appropriate stream implementation for the given options.
grouped_stream_to_frames	Transform multiple GenericStatementSinks into Jelly frames.
grouped_stream_to_file	Write stream of GenericStatementSink to a binary file.
flat_stream_to_frames	Serialize a stream of raw GenericStatementSink's triples or quads into Jelly frames.
flat_stream_to_file	Write Triple or Quad events to a binary file.

GenericSinkTermEncoder(lookup_preset=None)

Bases: TermEncoder

Methods:

Name	Description
same_term	Check whether two terms are equal, to find repeated terms.
encode_spo	Encode term based on its GenericSink object.
encode_graph	Encode graph term based on its GenericSink object.

Source code in pyjelly/serialize/encode.py

def __init__(
    self,
    lookup_preset: options.LookupPreset | None = None,
) -> None:
    if lookup_preset is None:
        lookup_preset = options.LookupPreset()
    self.lookup_preset = lookup_preset
    self.names = LookupEncoder(lookup_size=lookup_preset.max_names)
    self.prefixes = LookupEncoder(lookup_size=lookup_preset.max_prefixes)
    self.datatypes = LookupEncoder(lookup_size=lookup_preset.max_datatypes)
    # Where rows for new lookup entries go in the row layout. A stream replaces
    # it with its own.
    self.flow: FrameFlow = ManualFrameFlow()
    self.iri_cache: dict[str, _CachedIri] = {}
    # Cleared when full, so that a long stream doesn't grow it without bound
    self.iri_cache_size = max(4 * lookup_preset.max_names, 1024)
    # The encoded subject, predicate, object and graph of the statement
    self.encoded = [EncodedTerm() for _ in range(4)]
    # Set by streams of Jelly-RDF 1.2. Then IRIs are encoded with the indices of
    # their entries, new entries are packed (see PackedEntries), and quoted
    # triples are encoded as triple terms.
    self.column_layout = False
    # The RDF version that a Jelly-RDF 1.2 stream declares, which limits its terms
    self.rdf_version: int = jelly.RDF_VERSION_UNSPECIFIED
    # The IRI sequence of the triple terms of a frame (only the object column can
    # have them): the indices of the last IRI. -1 makes the first IRI state its
    # prefix.
    self.triple_prefix_index = -1
    self.triple_name_index = 0
    # How deep the triple term being encoded is nested, and how many IRIs the
    # outermost one has so far
    self.triple_depth = 0
    self.triple_iris = 0

same_term(a, b)

Check whether two terms are equal, to find repeated terms.

IRIs, blank nodes and literals are compared field by field, which is cheaper than calling their eq.

Source code in pyjelly/integrations/generic/serialize.py

def same_term(self, a: object, b: object) -> bool:
    """
    Check whether two terms are equal, to find repeated terms.

    IRIs, blank nodes and literals are compared field by field, which is
    cheaper than calling their __eq__.
    """
    if a is b:
        return True
    cls = type(a)
    if cls is not type(b):
        return False
    if cls is IRI:
        return cast(IRI, a)._iri == cast(IRI, b)._iri
    if cls is BlankNode:
        return cast(BlankNode, a)._id == cast(BlankNode, b)._id
    if cls is Literal:
        x = cast(Literal, a)
        y = cast(Literal, b)
        return (
            x._lex == y._lex
            and x._langtag == y._langtag
            and x._datatype == y._datatype
            and x._direction == y._direction
        )
    return a == b

encode_spo(term, out)

Encode term based on its GenericSink object.

Args: term (object): term to encode out (EncodedTerm): where to put the encoded term.

Source code in pyjelly/integrations/generic/serialize.py

def encode_spo(self, term: object, out: EncodedTerm) -> None:
    """
    Encode term based on its GenericSink object.

    Args:
        term (object): term to encode
        out (EncodedTerm): where to put the encoded term.

    """
    if isinstance(term, IRI):
        self.encode_iri(term._iri, out)
        return

    if isinstance(term, Literal):
        self.encode_literal(
            lex=term._lex,
            language=term._langtag,
            datatype=term._datatype,
            direction=term._direction,
            out=out,
        )
        return

    if isinstance(term, BlankNode):
        self.encode_bnode(term._id, out)
        return

    if isinstance(term, Triple):
        self.encode_quoted_triple(term, out)
        return

    super().encode_spo(term, out)  # raises NotImplementedError

encode_graph(term, out)

Encode graph term based on its GenericSink object.

Args: term (object): term to encode out (EncodedTerm): where to put the encoded graph name.

Source code in pyjelly/integrations/generic/serialize.py

def encode_graph(self, term: object, out: EncodedTerm) -> None:
    """
    Encode graph term based on its GenericSink object.

    Args:
        term (object): term to encode
        out (EncodedTerm): where to put the encoded graph name.

    """
    if term is DefaultGraph:
        self.encode_default_graph(out)
        return
    if isinstance(term, IRI):
        self.encode_iri(term._iri, out)
        return

    if isinstance(term, Literal):
        self.encode_literal(
            lex=term._lex,
            language=term._langtag,
            datatype=term._datatype,
            out=out,
        )
        return

    if isinstance(term, BlankNode):
        self.encode_bnode(term._id, out)
        return
    super().encode_graph(term, out)  # raises NotImplementedError

triples_stream_frames(stream, data)

Serialize a GenericStatementSink into frames using physical type triples stream.

Args: stream (TripleStream): stream that specifies triples processing data (GenericStatementSink | Generator[Triple]): GenericStatementSink/Statements to serialize.

Yields: Generator[jelly.RdfStreamFrame]: jelly frames.

Source code in pyjelly/integrations/generic/serialize.py

@stream_frames.register(TripleStream)
def triples_stream_frames(
    stream: TripleStream,
    data: GenericStatementSink | Generator[Triple],
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a GenericStatementSink into frames using physical type triples stream.

    Args:
        stream (TripleStream): stream that specifies triples processing
        data (GenericStatementSink | Generator[Triple]):
            GenericStatementSink/Statements to serialize.

    Yields:
        Generator[jelly.RdfStreamFrame]: jelly frames.

    """
    stream.enroll()
    if (
        isinstance(data, GenericStatementSink)
        and stream.options.params.namespace_declarations
    ):
        yield from namespace_declarations(data, stream)

    graphs = (data,)
    for graph in graphs:
        for terms in graph:
            if frame := stream.triple(terms):
                yield frame
        if frame := stream.flow.frame_from_graph():
            yield frame
    if stream.stream_types.flat and (frame := stream.flow.to_stream_frame()):
        yield frame

quads_stream_frames(stream, data)

Serialize a GenericStatementSink into jelly frames using physical type quads stream.

Args: stream (QuadStream): stream that specifies quads processing data (GenericStatementSink | Generator[Quad]): Dataset to serialize.

Yields: Generator[jelly.RdfStreamFrame]: jelly frames

Source code in pyjelly/integrations/generic/serialize.py

@stream_frames.register(QuadStream)
def quads_stream_frames(
    stream: QuadStream,
    data: GenericStatementSink | Generator[Quad],
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a GenericStatementSink into jelly frames using physical type quads stream.

    Args:
        stream (QuadStream): stream that specifies quads processing
        data (GenericStatementSink | Generator[Quad]): Dataset to serialize.

    Yields:
        Generator[jelly.RdfStreamFrame]: jelly frames

    """
    stream.enroll()
    if stream.options.params.namespace_declarations:
        yield from namespace_declarations(data, stream)  # type: ignore[arg-type]

    iterator: Generator[Quad]
    if isinstance(data, GenericStatementSink):
        iterator = cast(Generator[Quad], data.store)
    else:
        iterator = data

    for terms in iterator:
        if frame := stream.quad(terms):
            yield frame
    if frame := stream.flow.frame_from_dataset():
        yield frame
    if stream.stream_types.flat and (frame := stream.flow.to_stream_frame()):
        yield frame

guess_options(sink, *, grouped=False)

Guess the serializer options based on the store type.

For a stream of graphs or datasets (grouped), each is an RDF Message in Jelly-RDF 1.2, and one frame in Jelly-RDF 1.1.

Source code in pyjelly/integrations/generic/serialize.py

def guess_options(
    sink: GenericStatementSink, *, grouped: bool = False
) -> SerializerOptions:
    """
    Guess the serializer options based on the store type.

    For a stream of graphs or datasets (`grouped`), each is an RDF Message in
    Jelly-RDF 1.2, and one frame in Jelly-RDF 1.1.
    """
    if sink.is_triples_sink:
        logical_type = (
            jelly.LOGICAL_STREAM_TYPE_GRAPHS
            if grouped
            else jelly.LOGICAL_STREAM_TYPE_FLAT_TRIPLES
        )
    else:
        logical_type = (
            jelly.LOGICAL_STREAM_TYPE_DATASETS
            if grouped
            else jelly.LOGICAL_STREAM_TYPE_FLAT_QUADS
        )
    # Generic sink supports both RDF-star and generalized statements by default
    # as it can handle any term types including quoted triples and generalized RDF terms
    params = StreamParameters(generalized_statements=True, rdf_star=True)
    return SerializerOptions(logical_type=logical_type, params=params)

guess_stream(options, sink)

Return an appropriate stream implementation for the given options.

Notes: if base(!) logical type is GRAPHS and sink.is_triples_sink is false, initializes TripleStream

Source code in pyjelly/integrations/generic/serialize.py

def guess_stream(options: SerializerOptions, sink: GenericStatementSink) -> Stream:
    """
    Return an appropriate stream implementation for the given options.

    Notes: if base(!) logical type is GRAPHS and sink.is_triples_sink is false,
        initializes TripleStream
    """
    stream_cls: type[Stream]
    if (
        options.logical_type % 10
    ) != jelly.LOGICAL_STREAM_TYPE_GRAPHS and not sink.is_triples_sink:
        stream_cls = QuadStream
    else:
        stream_cls = TripleStream
    if options is not None:
        lookup_preset = options.lookup_preset
    return stream_cls(
        encoder=GenericSinkTermEncoder(lookup_preset=lookup_preset),
        options=options,
    )

grouped_stream_to_frames(sink_generator, options=None)

Transform multiple GenericStatementSinks into Jelly frames.

Notes: One frame per GenericStatementSink.

Note: options are guessed if not provided.

Args: sink_generator (Generator[GenericStatementSink]): Generator of GenericStatementSink to transform. options (SerializerOptions | None, optional): stream options to use. Options are guessed based on the sink store type. Defaults to None.

Yields: Generator[jelly.RdfStreamFrame]: produced Jelly frames

Source code in pyjelly/integrations/generic/serialize.py

def grouped_stream_to_frames(
    sink_generator: Generator[GenericStatementSink],
    options: SerializerOptions | None = None,
) -> Generator[jelly.RdfStreamFrame]:
    """
    Transform multiple GenericStatementSinks into Jelly frames.

    Notes:
        One frame per GenericStatementSink.

    Note: options are guessed if not provided.

    Args:
        sink_generator (Generator[GenericStatementSink]): Generator of
            GenericStatementSink to transform.
        options (SerializerOptions | None, optional): stream options to use.
            Options are guessed based on the sink store type. Defaults to None.

    Yields:
        Generator[jelly.RdfStreamFrame]: produced Jelly frames

    """
    stream = None
    for sink in sink_generator:
        if not stream:
            if options is None:
                options = guess_options(sink, grouped=True)
            stream = guess_stream(options, sink)
        yield from stream_frames(stream, sink)

grouped_stream_to_file(stream, output_file, **kwargs)

Write stream of GenericStatementSink to a binary file.

Args: stream (Generator[GenericStatementSink]): Generator of GenericStatementSink to serialize. output_file (IO[bytes]): output buffered writer. **kwargs (Any): options to pass to stream.

Source code in pyjelly/integrations/generic/serialize.py

def grouped_stream_to_file(
    stream: Generator[GenericStatementSink],
    output_file: IO[bytes],
    **kwargs: Any,
) -> None:
    """
    Write stream of GenericStatementSink to a binary file.

    Args:
        stream (Generator[GenericStatementSink]): Generator of
            GenericStatementSink to serialize.
        output_file (IO[bytes]): output buffered writer.
        **kwargs (Any): options to pass to stream.

    """
    for frame in grouped_stream_to_frames(stream, **kwargs):
        write_delimited(frame, output_file)

flat_stream_to_frames(statements, options=None)

Serialize a stream of raw GenericStatementSink's triples or quads into Jelly frames.

Args: statements (Generator[Triple | Quad]): s/p/o triples or s/p/o/g quads to serialize. options (SerializerOptions | None, optional): if omitted, guessed based on the first tuple.

Yields: Generator[jelly.RdfStreamFrame]: generated frames.

Source code in pyjelly/integrations/generic/serialize.py

def flat_stream_to_frames(
    statements: Generator[Triple | Quad],
    options: SerializerOptions | None = None,
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a stream of raw GenericStatementSink's triples or quads into Jelly frames.

    Args:
        statements (Generator[Triple | Quad]):
          s/p/o triples or s/p/o/g quads to serialize.
        options (SerializerOptions | None, optional):
            if omitted, guessed based on the first tuple.

    Yields:
        Generator[jelly.RdfStreamFrame]: generated frames.

    """
    first = next(statements, None)
    if first is None:
        return

    sink = GenericStatementSink()
    sink.add(first)
    if options is None:
        options = guess_options(sink)
    stream = guess_stream(options, sink)

    combined: Generator[Triple | Quad] | GenericStatementSink = (
        item for item in chain([first], statements)
    )

    yield from stream_frames(stream, combined)

flat_stream_to_file(statements, output_file, options=None)

Write Triple or Quad events to a binary file.

Args: statements (Generator[Triple | Quad]): statements to serialize. output_file (IO[bytes]): output buffered writer. options (SerializerOptions | None, optional): stream options.

Source code in pyjelly/integrations/generic/serialize.py

def flat_stream_to_file(
    statements: Generator[Triple | Quad],
    output_file: IO[bytes],
    options: SerializerOptions | None = None,
) -> None:
    """
    Write Triple or Quad events to a binary file.

    Args:
        statements (Generator[Triple | Quad]): statements to serialize.
        output_file (IO[bytes]): output buffered writer.
        options (SerializerOptions | None, optional): stream options.

    """
    for frame in flat_stream_to_frames(statements, options):
        write_delimited(frame, output_file)

sparql

Jelly-SPARQL support for the generic API (without rdflib).

Functions:

Name	Description
parse_jelly_sparql	Read SPARQL query results from Jelly-SPARQL.
parse_jelly_sparql_sequence	Read a sequence of SPARQL query results from one Jelly-SPARQL stream.
sparql_stream	Create a Jelly-SPARQL writer for generic API terms.
serialize_jelly_sparql	Write a solution sequence (the result of a SELECT query) in Jelly-SPARQL.
serialize_jelly_sparql_ask	Write a boolean result (of an ASK query) in Jelly-SPARQL.
serialize_jelly_sparql_sequence	Write a sequence of SPARQL query results in one Jelly-SPARQL stream.

parse_jelly_sparql(inp)

Read SPARQL query results from Jelly-SPARQL.

Rows are read during iteration (only once), so the input must stay open until then.

Args: inp (IO[bytes] | bytes): the stream, delimited or not. It must have one result set (the FLAT stream type, the default).

Returns: SparqlResultSet: the results. Iterating it gives the rows, as tuples with one value per variable, None if the variable is unbound.

Raises: JellyIncompleteResultsError: if the writer reported that the results are incomplete. Raised at the end of the iteration, or right away for a boolean result.

Source code in pyjelly/integrations/generic/sparql.py

def parse_jelly_sparql(inp: IO[bytes] | bytes) -> SparqlResultSet:
    """
    Read SPARQL query results from Jelly-SPARQL.

    Rows are read during iteration (only once), so the input must stay open
    until then.

    Args:
        inp (IO[bytes] | bytes): the stream, delimited or not. It must have one
            result set (the FLAT stream type, the default).

    Returns:
        SparqlResultSet: the results. Iterating it gives the rows, as tuples with
            one value per variable, None if the variable is unbound.

    Raises:
        JellyIncompleteResultsError: if the writer reported that the results are
            incomplete. Raised at the end of the iteration, or right away for a
            boolean result.

    """
    return read_result_set(inp, GenericStatementSinkAdapter)

parse_jelly_sparql_sequence(inp)

Read a sequence of SPARQL query results from one Jelly-SPARQL stream.

This reads PUNCTUATED streams, which hold any number of results. A FLAT stream (the default) holds one. Rows are read during iteration. If the next result set is read first, the unread rows are kept in memory.

Args: inp (IO[bytes] | bytes): the stream, delimited.

Yields: SparqlResultSet: the results, one per result set.

Source code in pyjelly/integrations/generic/sparql.py

def parse_jelly_sparql_sequence(inp: IO[bytes] | bytes) -> Iterator[SparqlResultSet]:
    """
    Read a sequence of SPARQL query results from one Jelly-SPARQL stream.

    This reads PUNCTUATED streams, which hold any number of results. A FLAT
    stream (the default) holds one. Rows are read during iteration. If the next
    result set is read first, the unread rows are kept in memory.

    Args:
        inp (IO[bytes] | bytes): the stream, delimited.

    Yields:
        SparqlResultSet: the results, one per result set.

    """
    return read_result_sets(inp, GenericStatementSinkAdapter)

sparql_stream(options=None)

Create a Jelly-SPARQL writer for generic API terms.

Source code in pyjelly/integrations/generic/sparql.py

def sparql_stream(options: SparqlOptions | None = None) -> SparqlStream:
    """Create a Jelly-SPARQL writer for generic API terms."""
    if options is None:
        options = SparqlOptions()
    encoder = GenericSinkTermEncoder(lookup_preset=options.lookup_preset)
    return SparqlStream(encoder, options)

serialize_jelly_sparql(variables, rows, out, options=None, *, links=())

Write a solution sequence (the result of a SELECT query) in Jelly-SPARQL.

If the rows fail midway (e.g., the query fails), the stream is ended with the error so readers know the results are incomplete, and the exception is re-raised.

Args: variables (Iterable[str]): variable names, in projection order. rows (Iterable[Sequence[Node | None]]): the rows, with one value per variable, None if the variable is unbound. out (IO[bytes]): output stream. options (SparqlOptions | None): options of the stream. links (Sequence[str]): link IRIs for the results.

Source code in pyjelly/integrations/generic/sparql.py

def serialize_jelly_sparql(
    variables: Iterable[str],
    rows: Iterable[Sequence[Node | None]],
    out: IO[bytes],
    options: SparqlOptions | None = None,
    *,
    links: Sequence[str] = (),
) -> None:
    """
    Write a solution sequence (the result of a SELECT query) in Jelly-SPARQL.

    If the rows fail midway (e.g., the query fails), the stream is ended with the
    error so readers know the results are incomplete, and the exception is
    re-raised.

    Args:
        variables (Iterable[str]): variable names, in projection order.
        rows (Iterable[Sequence[Node | None]]): the rows, with one value per
            variable, None if the variable is unbound.
        out (IO[bytes]): output stream.
        options (SparqlOptions | None): options of the stream.
        links (Sequence[str]): link IRIs for the results.

    """
    stream = sparql_stream(options)
    writer = FrameWriter(out, delimited=stream.options.delimited)
    write_solutions(stream, variables, rows, writer, links=links)

serialize_jelly_sparql_ask(value, out, options=None, *, links=())

Write a boolean result (of an ASK query) in Jelly-SPARQL.

Args: value (bool): the result. out (IO[bytes]): output stream. options (SparqlOptions | None): options of the stream. links (Sequence[str]): link IRIs for the result.

Source code in pyjelly/integrations/generic/sparql.py

def serialize_jelly_sparql_ask(
    value: bool,  # noqa: FBT001
    out: IO[bytes],
    options: SparqlOptions | None = None,
    *,
    links: Sequence[str] = (),
) -> None:
    """
    Write a boolean result (of an ASK query) in Jelly-SPARQL.

    Args:
        value (bool): the result.
        out (IO[bytes]): output stream.
        options (SparqlOptions | None): options of the stream.
        links (Sequence[str]): link IRIs for the result.

    """
    stream = sparql_stream(options)
    FrameWriter(out, delimited=stream.options.delimited).write(
        stream.ask(value, links=links)
    )

serialize_jelly_sparql_sequence(results, out, options=None)

Write a sequence of SPARQL query results in one Jelly-SPARQL stream.

The stream is PUNCTUATED, with one result set per result.

Args: results (Iterable[SequenceItem]): the results, each a (variables, rows) pair for a SELECT query, or a bool for an ASK query. out (IO[bytes]): output stream. options (SparqlOptions | None): options of the stream, which must be punctuated. Defaults to SparqlOptions(punctuated=True).

Source code in pyjelly/integrations/generic/sparql.py

def serialize_jelly_sparql_sequence(
    results: Iterable[SequenceItem],
    out: IO[bytes],
    options: SparqlOptions | None = None,
) -> None:
    """
    Write a sequence of SPARQL query results in one Jelly-SPARQL stream.

    The stream is PUNCTUATED, with one result set per result.

    Args:
        results (Iterable[SequenceItem]): the results, each a (variables, rows)
            pair for a SELECT query, or a bool for an ASK query.
        out (IO[bytes]): output stream.
        options (SparqlOptions | None): options of the stream, which must be
            punctuated. Defaults to SparqlOptions(punctuated=True).

    """
    if options is None:
        options = SparqlOptions(punctuated=True)
    if not options.punctuated:
        msg = "a sequence of results needs SparqlOptions(punctuated=True)"
        raise ValueError(msg)
    stream = sparql_stream(options)
    writer = FrameWriter(out, delimited=True)
    for result in results:
        if isinstance(result, bool):
            writer.write(stream.ask(result))
        else:
            variables, rows = result
            write_solutions(stream, variables, rows, writer)

rdflib

Modules:

Name	Description
decode	Adapters that make rdflib terms and statements from decoded Jelly rows.
encode	Encoding of rdflib terms. Compiled with mypyc, like the other encoders.
parse
serialize
sparql	Jelly-SPARQL support for rdflib's SPARQL query results (rdflib.query.Result).
statements	The statements and prefixes that the rdflib parser yields.
walk	Walk rdflib graphs grouped by graph, subject and predicate (gspo order).

Functions:

Name	Description
register_extension_to_rdflib	Make rdflib.util.guess_format discover Jelly format.

register_extension_to_rdflib(extension='.jelly')

Make rdflib.util.guess_format discover Jelly format.

rdflib.util.guess_format("foo.jelly") register_extension_to_rdflib() rdflib.util.guess_format("foo.jelly") 'jelly'

Source code in pyjelly/integrations/rdflib/__init__.py

def register_extension_to_rdflib(extension: str = ".jelly") -> None:
    """
    Make [rdflib.util.guess_format][] discover Jelly format.

    >>> rdflib.util.guess_format("foo.jelly")
    >>> register_extension_to_rdflib()
    >>> rdflib.util.guess_format("foo.jelly")
    'jelly'
    """
    rdflib.util.SUFFIX_FORMAT_MAP[extension.removeprefix(".")] = "jelly"

decode

Adapters that make rdflib terms and statements from decoded Jelly rows.

Compiled with mypyc, like the decoder that calls them. rdflib is not installed when mypyc builds pyjelly, so its classes are only Any to the type checker here.

Classes:

Name	Description
RDFLibAdapter	RDFLib adapter class, is extended by triples and quads implementations.
RDFLibTriplesAdapter	Triples adapter RDFLib implementation.
RDFLibQuadsAdapter	Extended RDFLib adapter for the QUADS physical type.
RDFLibGraphsAdapter	Extension of RDFLibQuadsBaseAdapter for the GRAPHS physical type.
BindingRows	Row factory that turns the cells of Jelly-SPARQL frames into rdflib bindings.

RDFLibAdapter(options, *, validate_iris=False)

Bases: Adapter

RDFLib adapter class, is extended by triples and quads implementations.

Args: options (ParserOptions): options of the stream validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is still read, rdflib only logs a warning. Checking slows down parsing.

Source code in pyjelly/integrations/rdflib/decode.py

def __init__(self, options: ParserOptions, *, validate_iris: bool = False) -> None:
    super().__init__(options=options)
    self.validate_iris = validate_iris

RDFLibTriplesAdapter(options, *, validate_iris=False)

Bases: RDFLibAdapter

Triples adapter RDFLib implementation.

Notes: returns triple/namespace declaration as soon as receives them.

Source code in pyjelly/integrations/rdflib/decode.py

def __init__(self, options: ParserOptions, *, validate_iris: bool = False) -> None:
    super().__init__(options=options, validate_iris=validate_iris)

RDFLibQuadsAdapter(options, *, validate_iris=False)

Bases: RDFLibQuadsBaseAdapter

Extended RDFLib adapter for the QUADS physical type.

Args: RDFLibQuadsBaseAdapter (RDFLibAdapter): base quads adapter (shared with graphs physical type)

Source code in pyjelly/integrations/rdflib/decode.py

def __init__(self, options: ParserOptions, *, validate_iris: bool = False) -> None:
    super().__init__(options=options, validate_iris=validate_iris)

RDFLibGraphsAdapter(options, *, validate_iris=False)

Bases: RDFLibQuadsBaseAdapter

Extension of RDFLibQuadsBaseAdapter for the GRAPHS physical type.

Notes: introduces graph start/end, checks if graph exists.

Args: RDFLibQuadsBaseAdapter (RDFLibAdapter): base adapter for quads management.

Raises: JellyConformanceError: if no graph_start was encountered

Source code in pyjelly/integrations/rdflib/decode.py

def __init__(self, options: ParserOptions, *, validate_iris: bool = False) -> None:
    super().__init__(options=options, validate_iris=validate_iris)
    self._graph_id = None

BindingRows()

Row factory that turns the cells of Jelly-SPARQL frames into rdflib bindings.

A binding is a dict from each bound Variable to its value, as in rdflib's query results. Equal names share one Variable object (see variables()).

Methods:

Name	Description
variables	Get the Variable objects for these names.

Source code in pyjelly/integrations/rdflib/decode.py

def __init__(self) -> None:
    self.names: list[str] = []
    self.vars: list[Variable] = []

variables(names)

Get the Variable objects for these names.

Source code in pyjelly/integrations/rdflib/decode.py

def variables(self, names: list[str]) -> list[Variable]:
    """Get the Variable objects for these names."""
    if names != self.names:
        self.names = names
        self.vars = [Variable(name) for name in names]
    return self.vars

encode

Encoding of rdflib terms. Compiled with mypyc, like the other encoders.

rdflib is not installed when mypyc builds pyjelly, so its classes are only Any to the type checker here.

Classes:

Name	Description
RDFLibTermEncoder

RDFLibTermEncoder(lookup_preset=None)

Bases: TermEncoder

Methods:

Name	Description
same_term	Return whether two terms are equal, for repeated terms.
encode_spo	Encode s/p/o term based on its RDFLib object.
encode_graph	Encode graph name term based on its RDFLib object.

Source code in pyjelly/serialize/encode.py

def __init__(
    self,
    lookup_preset: options.LookupPreset | None = None,
) -> None:
    if lookup_preset is None:
        lookup_preset = options.LookupPreset()
    self.lookup_preset = lookup_preset
    self.names = LookupEncoder(lookup_size=lookup_preset.max_names)
    self.prefixes = LookupEncoder(lookup_size=lookup_preset.max_prefixes)
    self.datatypes = LookupEncoder(lookup_size=lookup_preset.max_datatypes)
    # Where rows for new lookup entries go in the row layout. A stream replaces
    # it with its own.
    self.flow: FrameFlow = ManualFrameFlow()
    self.iri_cache: dict[str, _CachedIri] = {}
    # Cleared when full, so that a long stream doesn't grow it without bound
    self.iri_cache_size = max(4 * lookup_preset.max_names, 1024)
    # The encoded subject, predicate, object and graph of the statement
    self.encoded = [EncodedTerm() for _ in range(4)]
    # Set by streams of Jelly-RDF 1.2. Then IRIs are encoded with the indices of
    # their entries, new entries are packed (see PackedEntries), and quoted
    # triples are encoded as triple terms.
    self.column_layout = False
    # The RDF version that a Jelly-RDF 1.2 stream declares, which limits its terms
    self.rdf_version: int = jelly.RDF_VERSION_UNSPECIFIED
    # The IRI sequence of the triple terms of a frame (only the object column can
    # have them): the indices of the last IRI. -1 makes the first IRI state its
    # prefix.
    self.triple_prefix_index = -1
    self.triple_name_index = 0
    # How deep the triple term being encoded is nested, and how many IRIs the
    # outermost one has so far
    self.triple_depth = 0
    self.triple_iris = 0

same_term(a, b)

Return whether two terms are equal, for repeated terms.

rdflib's eq is slow Python code, so the common cases are answered here by comparing the terms as strings:

• Two URIRefs or two BNodes are equal if their strings are.
• Two Literals with different lexical forms are never equal. With the same lexical form, language tag and datatype, they are equal. Otherwise rdflib decides (e.g., for language tags that differ only in case).
• Terms of different classes among these three are never equal.

Source code in pyjelly/integrations/rdflib/encode.py

def same_term(self, a: object, b: object) -> bool:
    """
    Return whether two terms are equal, for repeated terms.

    rdflib's __eq__ is slow Python code, so the common cases are answered here
    by comparing the terms as strings:

    - Two URIRefs or two BNodes are equal if their strings are.
    - Two Literals with different lexical forms are never equal. With the same
      lexical form, language tag and datatype, they are equal. Otherwise rdflib
      decides (e.g., for language tags that differ only in case).
    - Terms of different classes among these three are never equal.
    """
    if a is b:
        return True
    cls = type(a)
    other_cls = type(b)
    if cls is other_cls:
        if cls is URIRef or cls is BNode:
            a_str: str = cast("str", a)
            b_str: str = cast("str", b)
            return a_str == b_str
        if cls is Literal:
            a_lex: str = cast("str", a)
            b_lex: str = cast("str", b)
            if a_lex != b_lex:
                return False
            # Slots behind the language and datatype properties
            a_node: Any = a
            b_node: Any = b
            if _same_str(a_node._language, b_node._language) and _same_str(
                a_node._datatype, b_node._datatype
            ):
                return True
    elif (cls is URIRef or cls is BNode or cls is Literal) and (
        other_cls is URIRef or other_cls is BNode or other_cls is Literal
    ):
        return False
    return a == b

encode_spo(term, out)

Encode s/p/o term based on its RDFLib object.

Args: term (object): term to encode out (EncodedTerm): where to put the encoded term.

Source code in pyjelly/integrations/rdflib/encode.py

def encode_spo(self, term: object, out: EncodedTerm) -> None:
    """
    Encode s/p/o term based on its RDFLib object.

    Args:
        term (object): term to encode
        out (EncodedTerm): where to put the encoded term.

    """
    # Exact type checks first, as rdflib almost always makes terms of these
    # exact classes: they are cheaper than isinstance(), which is the fallback
    # for subclasses, and runs Python code for rdflib's abstract classes.
    # `node` is Any, because the type checks don't narrow the type when
    # rdflib isn't installed (as during the mypyc compilation).
    node: Any = term
    cls = type(node)
    if cls is URIRef or (
        cls is not Literal and cls is not BNode and isinstance(node, URIRef)
    ):
        # A plain str, so that the IRI cache never calls rdflib's __eq__
        self.encode_iri(str(node), out)
        return

    if cls is Literal or (cls is not BNode and isinstance(node, Literal)):
        # The slots behind the language and datatype properties, which are
        # Python functions
        datatype = node._datatype
        self.encode_literal(
            lex=str(node),
            language=node._language,
            # `datatype` is cast to `str` explicitly because
            # `URIRef.__eq__` overrides `str.__eq__` in an incompatible manner
            datatype=datatype and str(datatype),
            out=out,
        )
        return

    if isinstance(node, BNode):
        self.encode_bnode(str(node), out)
        return

    super().encode_spo(term, out)  # raises NotImplementedError

encode_graph(term, out)

Encode graph name term based on its RDFLib object.

Args: term (object): term to encode out (EncodedTerm): where to put the encoded graph name.

Source code in pyjelly/integrations/rdflib/encode.py

def encode_graph(self, term: object, out: EncodedTerm) -> None:
    """
    Encode graph name term based on its RDFLib object.

    Args:
        term (object): term to encode
        out (EncodedTerm): where to put the encoded graph name.

    """
    if self.same_term(term, DATASET_DEFAULT_GRAPH_ID):
        self.encode_default_graph(out)
        return

    # Exact type checks first, see encode_spo()
    cls = type(term)
    if cls is URIRef or (cls is not BNode and isinstance(term, URIRef)):
        self.encode_iri(str(term), out)
        return

    if isinstance(term, BNode):
        self.encode_bnode(str(term), out)
        return
    super().encode_graph(term, out)  # raises NotImplementedError

parse

Classes:

Name	Description
RDFLibJellyParser

Functions:

Name	Description
parse_triples_stream	Parse flat triple stream.
parse_quads_stream	Parse flat quads stream.
parse_jelly_grouped	Take jelly file and return generators based on the detected physical type.
parse_jelly_to_graph	Add statements from Generator to provided Graph/Dataset.
parse_jelly_flat	Parse jelly file with FLAT logical type into a Generator of stream events.

RDFLibJellyParser

Bases: Parser

Methods:

Name	Description
parse	Parse jelly file into provided RDFLib Graph.

parse(source, sink, *, validate_iris=False)

Parse jelly file into provided RDFLib Graph.

Args: source (InputSource): jelly file as buffered binary stream InputSource obj sink (Graph): RDFLib Graph validate_iris (bool): whether rdflib checks the IRIs, see parse_jelly_to_graph(). Pass it to Graph.parse().

Raises: TypeError: raises error if invalid input

Source code in pyjelly/integrations/rdflib/parse.py

def parse(
    self,
    source: InputSource,
    sink: Graph,
    *,
    validate_iris: bool = False,
) -> None:
    """
    Parse jelly file into provided RDFLib Graph.

    Args:
        source (InputSource): jelly file as buffered binary stream InputSource obj
        sink (Graph): RDFLib Graph
        validate_iris (bool): whether rdflib checks the IRIs, see
            parse_jelly_to_graph(). Pass it to Graph.parse().

    Raises:
        TypeError: raises error if invalid input

    """
    byte_stream = source.getByteStream()
    if byte_stream is None:
        msg = "expected source to be a stream of bytes"
        raise TypeError(msg)

    inp = cast(IO[bytes], byte_stream)
    if inp is None:
        msg = "expected source to be a stream of bytes"
        raise TypeError(msg)
    # The default graph of a stream of quads goes into the Graph that rdflib
    # parses into, as with TriG and N-Quads: for Dataset.parse(), that is the
    # Dataset's default graph
    target = Graph(store=sink.store, identifier=sink.identifier)
    parse_jelly_to_graph(
        inp,
        graph_factory=lambda: target,
        dataset_factory=lambda: Dataset(store=sink.store),
        validate_iris=validate_iris,
        default_graph=target,
    )

parse_triples_stream(frames, options, frame_metadata=None, *, validate_iris=False)

Parse flat triple stream.

Args: frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames options (ParserOptions): stream options frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable used for extracting frame metadata validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is still read, rdflib only logs a warning. Checking slows down parsing.

Yields: Generator[Iterable[Triple | Prefix]]: Generator of iterables of Triple or Prefix objects, one iterable per frame.

Source code in pyjelly/integrations/rdflib/parse.py

def parse_triples_stream(
    frames: Iterable[jelly.RdfStreamFrame],
    options: ParserOptions,
    frame_metadata: ContextVar[MutableMapping[str, bytes]] | None = None,
    *,
    validate_iris: bool = False,
) -> Generator[Iterable[Triple | Prefix]]:
    """
    Parse flat triple stream.

    Args:
        frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames
        options (ParserOptions): stream options
        frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable
            used for extracting frame metadata
        validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is
            still read, rdflib only logs a warning. Checking slows down parsing.

    Yields:
        Generator[Iterable[Triple | Prefix]]:
            Generator of iterables of Triple or Prefix objects,
            one iterable per frame.

    """
    adapter = RDFLibTriplesAdapter(options, validate_iris=validate_iris)
    decoder = Decoder(adapter=adapter)
    for frame in frames:
        if frame_metadata is not None:
            frame_metadata.set(
                frame.metadata
            ) if frame.metadata else frame_metadata.set({})
        yield decoder.iter_rows(frame)
    if options.params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES:
        yield decoder.end_of_stream()

parse_quads_stream(frames, options, frame_metadata=None, *, validate_iris=False)

Parse flat quads stream.

Args: frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames options (ParserOptions): stream options frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable used for extracting frame metadata validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is still read, rdflib only logs a warning. Checking slows down parsing.

Yields: Generator[Iterable[Quad | Prefix]]: Generator of iterables of Quad or Prefix objects, one iterable per frame.

Source code in pyjelly/integrations/rdflib/parse.py

def parse_quads_stream(
    frames: Iterable[jelly.RdfStreamFrame],
    options: ParserOptions,
    frame_metadata: ContextVar[MutableMapping[str, bytes]] | None = None,
    *,
    validate_iris: bool = False,
) -> Generator[Iterable[Quad | Prefix]]:
    """
    Parse flat quads stream.

    Args:
        frames (Iterable[jelly.RdfStreamFrame]): iterator over stream frames
        options (ParserOptions): stream options
        frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable
            used for extracting frame metadata
        validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is
            still read, rdflib only logs a warning. Checking slows down parsing.

    Yields:
        Generator[Iterable[Quad | Prefix]]:
            Generator of iterables of Quad or Prefix objects,
            one iterable per frame.

    """
    adapter_class: type[RDFLibQuadsBaseAdapter]
    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_QUADS:
        adapter_class = RDFLibQuadsAdapter
    else:
        adapter_class = RDFLibGraphsAdapter
    adapter = adapter_class(options=options, validate_iris=validate_iris)
    decoder = Decoder(adapter=adapter)
    for frame in frames:
        if frame_metadata is not None:
            frame_metadata.set(
                frame.metadata
            ) if frame.metadata else frame_metadata.set({})
        yield decoder.iter_rows(frame)
    if options.params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES:
        yield decoder.end_of_stream()

parse_jelly_grouped(inp, graph_factory=Graph, dataset_factory=Dataset, *, logical_type_strict=False, frame_metadata=None, validate_iris=False)

Take jelly file and return generators based on the detected physical type.

Yields one graph/dataset per frame.

Args: inp (IO[bytes]): input jelly buffered binary stream graph_factory (Callable): lambda to construct a Graph. By default creates an empty in-memory Graph, but you can pass something else here. dataset_factory (Callable): lambda to construct a Dataset. By default creates an empty in-memory Dataset, but you can pass something else here. logical_type_strict (bool): If True, validate the logical type in stream options and require a grouped logical type. Otherwise, only the physical type is used to route parsing. frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable used for extracting frame metadata validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is still read, rdflib only logs a warning. Checking slows down parsing.

Raises: NotImplementedError: is raised if a physical type is not implemented

Yields: Generator[Graph] | Generator[Dataset]: returns generators for graphs/datasets based on the type of input

Source code in pyjelly/integrations/rdflib/parse.py

def parse_jelly_grouped(  # noqa: PLR0913 keyword-only options
    inp: IO[bytes],
    graph_factory: Callable[[], Graph] = Graph,
    dataset_factory: Callable[[], Dataset] = Dataset,
    *,
    logical_type_strict: bool = False,
    frame_metadata: ContextVar[MutableMapping[str, bytes]] | None = None,
    validate_iris: bool = False,
) -> Generator[Graph] | Generator[Dataset]:
    """
    Take jelly file and return generators based on the detected physical type.

    Yields one graph/dataset per frame.

    Args:
        inp (IO[bytes]): input jelly buffered binary stream
        graph_factory (Callable): lambda to construct a Graph.
            By default creates an empty in-memory Graph,
            but you can pass something else here.
        dataset_factory (Callable): lambda to construct a Dataset.
            By default creates an empty in-memory Dataset,
            but you can pass something else here.
        logical_type_strict (bool): If True, validate the *logical* type in
            stream options and require a grouped logical type. Otherwise, only the
            physical type is used to route parsing.
        frame_metadata: (ContextVar[ScalarMap[str, bytes]]): context variable
            used for extracting frame metadata
        validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is
            still read, rdflib only logs a warning. Checking slows down parsing.

    Raises:
        NotImplementedError: is raised if a physical type is not implemented

    Yields:
        Generator[Graph] | Generator[Dataset]:
            returns generators for graphs/datasets based on the type of input

    """
    options, frames = get_options_and_frames(inp)

    st = getattr(options, "stream_types", None)
    if logical_type_strict and (
        st is None
        or st.logical_type == jelly.LOGICAL_STREAM_TYPE_UNSPECIFIED
        or st.flat
    ):
        lt_name = (
            "UNSPECIFIED"
            if st is None
            else jelly.LogicalStreamType.Name(st.logical_type)
        )

        msg = (
            "strict logical type check requires options.stream_types"
            if st is None
            else f"expected GROUPED logical type, got {lt_name}"
        )
        raise JellyConformanceError(msg)

    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_TRIPLES:
        for graph in grouped_rows(
            parse_triples_stream(
                frames=frames,
                options=options,
                frame_metadata=frame_metadata,
                validate_iris=validate_iris,
            ),
            options,
        ):
            sink = graph_factory()
            for graph_item in graph:
                if isinstance(graph_item, Prefix):
                    sink.bind(graph_item.prefix, graph_item.iri)
                else:
                    sink.add(graph_item)
            yield sink
        return
    elif options.stream_types.physical_type in (
        jelly.PHYSICAL_STREAM_TYPE_QUADS,
        jelly.PHYSICAL_STREAM_TYPE_GRAPHS,
    ):
        for dataset in grouped_rows(
            parse_quads_stream(
                frames=frames,
                options=options,
                frame_metadata=frame_metadata,
                validate_iris=validate_iris,
            ),
            options,
        ):
            sink = dataset_factory()
            for item in dataset:
                if isinstance(item, Prefix):
                    sink.bind(item.prefix, item.iri)
                else:
                    s, p, o, graph_name = item
                    context = sink.get_context(graph_name)
                    sink.add((s, p, o, context))
            yield sink
        return

    physical_type_name = jelly.PhysicalStreamType.Name(
        options.stream_types.physical_type
    )
    msg = f"the stream type {physical_type_name} is not supported "
    raise NotImplementedError(msg)

parse_jelly_to_graph(inp, graph_factory=Graph, dataset_factory=Dataset, *, validate_iris=False, default_graph=None)

Add statements from Generator to provided Graph/Dataset.

Args: inp (IO[bytes]): input jelly stream. graph_factory (Callable[[], Graph]): factory to create Graph. By default creates an empty in-memory Graph, but you can pass something else here. dataset_factory (Callable[[], Dataset]): factory to create Dataset. By default creates an empty in-memory Dataset, but you can pass something else here. validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is still read, rdflib only logs a warning. Checking slows down parsing. default_graph (Graph | None): graph for the quads in the default graph, in the same store as the Dataset. By default, the Dataset's default graph.

Returns: Dataset | Graph: Dataset or Graph with statements.

Source code in pyjelly/integrations/rdflib/parse.py

def parse_jelly_to_graph(
    inp: IO[bytes],
    graph_factory: Callable[[], Graph] = Graph,
    dataset_factory: Callable[[], Dataset] = Dataset,
    *,
    validate_iris: bool = False,
    default_graph: Graph | None = None,
) -> Graph | Dataset:
    """
    Add statements from Generator to provided Graph/Dataset.

    Args:
        inp (IO[bytes]): input jelly stream.
        graph_factory (Callable[[], Graph]): factory to create Graph.
            By default creates an empty in-memory Graph,
            but you can pass something else here.
        dataset_factory (Callable[[], Dataset]): factory to create Dataset.
            By default creates an empty in-memory Dataset,
            but you can pass something else here.
        validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is
            still read, rdflib only logs a warning. Checking slows down parsing.
        default_graph (Graph | None): graph for the quads in the default graph,
            in the same store as the Dataset. By default, the Dataset's default
            graph.

    Returns:
        Dataset | Graph: Dataset or Graph with statements.

    """
    options, frames = get_options_and_frames(inp)

    sink: Graph
    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_TRIPLES:
        sink = graph_factory()
        # Graph.add only checks the terms before giving them to the store
        to_store = type(sink).add is Graph.add
    else:
        # Other stream types fail in parse_jelly_flat
        sink = dataset_factory()
        # Dataset.add would also make a new context graph for every quad
        to_store = type(sink).add is ConjunctiveGraph.add

    items = parse_jelly_flat(
        inp=inp, frames=frames, options=options, validate_iris=validate_iris
    )
    if to_store:
        _add_to_store(sink, items, default_graph)
        return sink

    # The Graph or Dataset class changes add(), so it gets every statement
    for item in items:
        if isinstance(item, Prefix):
            sink.bind(item.prefix, item.iri)
        if isinstance(item, Triple):
            sink.add(item)
        if isinstance(item, Quad):
            s, p, o, graph_name = item
            if default_graph is not None and graph_name == DATASET_DEFAULT_GRAPH_ID:
                context = default_graph
            else:
                context = cast(Dataset, sink).get_context(graph_name)
            sink.add((s, p, o, context))  # type: ignore[arg-type]
    return sink

parse_jelly_flat(inp, frames=None, options=None, *, logical_type_strict=False, validate_iris=False)

Parse jelly file with FLAT logical type into a Generator of stream events.

Args: inp (IO[bytes]): input jelly buffered binary stream. frames (Iterable[jelly.RdfStreamFrame | None): jelly frames if read before. options (ParserOptions | None): stream options if read before. logical_type_strict (bool): If True, validate the logical type in stream options and require FLAT_(TRIPLES|QUADS). Otherwise, only the physical type is used to route parsing. validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is still read, rdflib only logs a warning. Checking slows down parsing.

Raises: NotImplementedError: if physical type is not supported

Yields: Generator[Statement | Prefix]: Generator of stream events

Source code in pyjelly/integrations/rdflib/parse.py

def parse_jelly_flat(
    inp: IO[bytes],
    frames: Iterable[jelly.RdfStreamFrame] | None = None,
    options: ParserOptions | None = None,
    *,
    logical_type_strict: bool = False,
    validate_iris: bool = False,
) -> Generator[Statement | Prefix]:
    """
    Parse jelly file with FLAT logical type into a Generator of stream events.

    Args:
        inp (IO[bytes]): input jelly buffered binary stream.
        frames (Iterable[jelly.RdfStreamFrame | None):
            jelly frames if read before.
        options (ParserOptions | None): stream options
            if read before.
        logical_type_strict (bool): If True, validate the *logical* type in
            stream options and require FLAT_(TRIPLES|QUADS). Otherwise, only the
            physical type is used to route parsing.
        validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is
            still read, rdflib only logs a warning. Checking slows down parsing.

    Raises:
        NotImplementedError: if physical type is not supported

    Yields:
        Generator[Statement | Prefix]: Generator of stream events

    """
    if frames is None or options is None:
        options, frames = get_options_and_frames(inp)

    st = getattr(options, "stream_types", None)
    if logical_type_strict and (st is None or not st.flat):
        lt_name = (
            "UNSPECIFIED"
            if st is None
            else jelly.LogicalStreamType.Name(st.logical_type)
        )
        msg = (
            "strict logical type check requires options.stream_types"
            if st is None
            else f"expected FLAT logical type (TRIPLES/QUADS), got {lt_name}"
        )
        raise JellyConformanceError(msg)

    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_TRIPLES:
        yield from flat_rows(
            parse_triples_stream(
                frames=frames, options=options, validate_iris=validate_iris
            ),
            options,
        )
        return
    if options.stream_types.physical_type in (
        jelly.PHYSICAL_STREAM_TYPE_QUADS,
        jelly.PHYSICAL_STREAM_TYPE_GRAPHS,
    ):
        yield from flat_rows(
            parse_quads_stream(frames=frames, options=options), options
        )
        return
    physical_type_name = jelly.PhysicalStreamType.Name(
        options.stream_types.physical_type
    )
    msg = f"the stream type {physical_type_name} is not supported "
    raise NotImplementedError(msg)

serialize

Classes:

Name	Description
RDFLibJellySerializer	RDFLib serializer for writing graphs in Jelly RDF stream format.

Functions:

Name	Description
triples_stream_frames	Serialize a Graph/Dataset into jelly frames.
quads_stream_frames	Serialize a Dataset into jelly frames.
guess_options	Guess the serializer options based on the store type.
guess_stream	Return an appropriate stream implementation for the given options.
grouped_stream_to_frames	Transform Graphs/Datasets into Jelly frames, one frame per Graph/Dataset.
grouped_stream_to_file	Write stream of Graphs/Datasets to a binary file.
flat_stream_to_frames	Serialize a stream of raw triples or quads into Jelly frames.
flat_stream_to_file	Write Triple or Quad events to a binary file in Jelly flat format.

RDFLibJellySerializer(store)

Bases: Serializer

RDFLib serializer for writing graphs in Jelly RDF stream format.

Handles streaming RDF terms into Jelly frames using internal encoders. Supports only graphs and datasets (not quoted graphs).

Methods:

Name	Description
serialize	Serialize self.store content to Jelly format.

Source code in pyjelly/integrations/rdflib/serialize.py

def __init__(self, store: Graph) -> None:
    if isinstance(store, QuotedGraph):
        msg = "N3 format is not supported"
        raise NotImplementedError(msg)
    super().__init__(store)

serialize(out, /, *, stream=None, options=None, **unused)

Serialize self.store content to Jelly format.

Args: out (IO[bytes]): output buffered writer stream (Stream | None, optional): Jelly stream object. Defaults to None. options (SerializerOptions | None, optional): Serializer options if defined beforehand, e.g., read from a separate file. Defaults to None. **unused(Any): unused args for RDFLib serialize

Source code in pyjelly/integrations/rdflib/serialize.py

@override
def serialize(  # type: ignore[override]
    self,
    out: IO[bytes],
    /,
    *,
    stream: Stream | None = None,
    options: SerializerOptions | None = None,
    **unused: Any,
) -> None:
    """
    Serialize self.store content to Jelly format.

    Args:
        out (IO[bytes]): output buffered writer
        stream (Stream | None, optional): Jelly stream object. Defaults to None.
        options (SerializerOptions | None, optional): Serializer options
            if defined beforehand, e.g., read from a separate file.
            Defaults to None.
        **unused(Any): unused args for RDFLib serialize

    """
    if options is None:
        options = guess_options(self.store)
    if stream is None:
        stream = guess_stream(options, self.store)
    frames = stream_frames(stream, self.store)
    if stream.options.params.delimited:
        for stream_frame in frames:
            write_delimited(stream_frame, out)
        return
    written = False
    for stream_frame in frames:
        if written:
            msg = (
                "the data does not fit in one frame. A non-delimited stream can "
                "only have one frame. Write a delimited stream, or use larger "
                "lookups."
            )
            raise JellyConformanceError(msg)
        write_single(stream_frame, out)
        written = True

triples_stream_frames(stream, data)

Serialize a Graph/Dataset into jelly frames.

Args: stream (TripleStream): stream that specifies triples processing data (Graph | Dataset | Iterable[Triple]): Graph/Dataset/Statements to serialize.

Notes: if Dataset (or ConjunctiveGraph) is given, its graphs are unpacked and iterated over. If flow is GraphsFrameFlow, emits a frame per graph.

Yields: Generator[jelly.RdfStreamFrame]: jelly frames.

Source code in pyjelly/integrations/rdflib/serialize.py

@stream_frames.register(TripleStream)
def triples_stream_frames(
    stream: TripleStream,
    data: Graph | Dataset | Iterable[Triple],
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a Graph/Dataset into jelly frames.

    Args:
        stream (TripleStream): stream that specifies triples processing
        data (Graph | Dataset | Iterable[Triple]):
            Graph/Dataset/Statements to serialize.

    Notes:
        if Dataset (or ConjunctiveGraph) is given, its graphs are unpacked and
        iterated over. If flow is GraphsFrameFlow, emits a frame per graph.

    Yields:
        Generator[jelly.RdfStreamFrame]: jelly frames.

    """
    stream.enroll()
    if isinstance(data, Graph) and stream.options.params.namespace_declarations:
        yield from namespace_declarations(data, stream)

    graphs: Iterable[Graph | Iterable[Triple]]
    if isinstance(data, Dataset):
        graphs = data.graphs()
    elif isinstance(data, ConjunctiveGraph):
        graphs = data.contexts()
    else:
        graphs = (data,)
    for graph in graphs:
        tree = spo_tree(graph) if isinstance(graph, Graph) else None
        if tree is not None:
            # An in-memory graph, grouped by subject and predicate
            yield from write_tree(stream, tree)
        else:
            # Any other store, or a generator of triples, in its own order
            yield from stream.triples(graph)
        if frame := stream.flow.frame_from_graph():
            yield frame
    if stream.stream_types.flat and (frame := stream.flow.to_stream_frame()):
        yield frame

quads_stream_frames(stream, data)

Serialize a Dataset into jelly frames.

Notes: Emits one frame per dataset if flow is of DatasetsFrameFlow. A ConjunctiveGraph is written as a Dataset, with its default context as the default graph.

Args: stream (QuadStream): stream that specifies quads processing data (Dataset | Iterable[Quad]): Dataset to serialize.

Yields: Generator[jelly.RdfStreamFrame]: jelly frames

Source code in pyjelly/integrations/rdflib/serialize.py

@stream_frames.register(QuadStream)
def quads_stream_frames(
    stream: QuadStream,
    data: Dataset | Iterable[Quad],
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a Dataset into jelly frames.

    Notes:
        Emits one frame per dataset if flow is of DatasetsFrameFlow.
        A ConjunctiveGraph is written as a Dataset, with its default context
        as the default graph.

    Args:
        stream (QuadStream): stream that specifies quads processing
        data (Dataset | Iterable[Quad]): Dataset to serialize.

    Yields:
        Generator[jelly.RdfStreamFrame]: jelly frames

    """
    stream.enroll()
    if stream.options.params.namespace_declarations:
        yield from namespace_declarations(data, stream)  # type: ignore[arg-type]
    trees = gspo_trees(data) if isinstance(data, ConjunctiveGraph) else None
    if trees is not None:
        # An in-memory dataset, grouped by graph, subject and predicate
        yield from write_trees(stream, trees)
    else:
        # Any other store, or a generator of quads, in its own order
        quads = dataset_quads(data) if isinstance(data, ConjunctiveGraph) else data
        yield from stream.quads(quads)
    if frame := stream.flow.frame_from_dataset():
        yield frame
    if stream.stream_types.flat and (frame := stream.flow.to_stream_frame()):
        yield frame

guess_options(sink, *, grouped=False)

Guess the serializer options based on the store type.

For a stream of graphs or datasets (grouped), each is an RDF Message in Jelly-RDF 1.2, and one frame in Jelly-RDF 1.1.

guess_options(Graph()).logical_type 1 guess_options(Dataset()).logical_type 2 guess_options(Dataset(), grouped=True).logical_type 4

A ConjunctiveGraph (e.g., the one rdfpipe reads into) holds quads too:

guess_options(ConjunctiveGraph()).logical_type 2

Source code in pyjelly/integrations/rdflib/serialize.py

def guess_options(sink: Graph | Dataset, *, grouped: bool = False) -> SerializerOptions:
    """
    Guess the serializer options based on the store type.

    For a stream of graphs or datasets (`grouped`), each is an RDF Message in
    Jelly-RDF 1.2, and one frame in Jelly-RDF 1.1.

    >>> guess_options(Graph()).logical_type
    1
    >>> guess_options(Dataset()).logical_type
    2
    >>> guess_options(Dataset(), grouped=True).logical_type
    4

    A ConjunctiveGraph (e.g., the one rdfpipe reads into) holds quads too:

    >>> guess_options(ConjunctiveGraph()).logical_type
    2
    """
    if isinstance(sink, ConjunctiveGraph):
        logical_type = (
            jelly.LOGICAL_STREAM_TYPE_DATASETS
            if grouped
            else jelly.LOGICAL_STREAM_TYPE_FLAT_QUADS
        )
    else:
        logical_type = (
            jelly.LOGICAL_STREAM_TYPE_GRAPHS
            if grouped
            else jelly.LOGICAL_STREAM_TYPE_FLAT_TRIPLES
        )
    # RDFLib doesn't support RDF-star and generalized statements by default
    # as it requires specific handling for quoted triples and non-standard RDF terms
    params = StreamParameters(generalized_statements=False, rdf_star=False)
    return SerializerOptions(logical_type=logical_type, params=params)

guess_stream(options, sink)

Return an appropriate stream implementation for the given options.

Notes: if base(!) logical type is GRAPHS and Dataset (or ConjunctiveGraph) is given, initializes TripleStream

graph_ser = RDFLibJellySerializer(Graph()) ds_ser = RDFLibJellySerializer(Dataset())

type(guess_stream(guess_options(graph_ser.store), graph_ser.store)) type(guess_stream(guess_options(ds_ser.store), ds_ser.store))

Source code in pyjelly/integrations/rdflib/serialize.py

def guess_stream(options: SerializerOptions, sink: Graph | Dataset) -> Stream:
    """
    Return an appropriate stream implementation for the given options.

    Notes: if base(!) logical type is GRAPHS and Dataset (or ConjunctiveGraph)
        is given, initializes TripleStream

    >>> graph_ser = RDFLibJellySerializer(Graph())
    >>> ds_ser = RDFLibJellySerializer(Dataset())

    >>> type(guess_stream(guess_options(graph_ser.store), graph_ser.store))
    <class 'pyjelly.serialize.streams.TripleStream'>
    >>> type(guess_stream(guess_options(ds_ser.store), ds_ser.store))
    <class 'pyjelly.serialize.streams.QuadStream'>
    """
    stream_cls: type[Stream]
    if (options.logical_type % 10) != jelly.LOGICAL_STREAM_TYPE_GRAPHS and isinstance(
        sink, ConjunctiveGraph
    ):
        stream_cls = QuadStream
    else:
        stream_cls = TripleStream
    return stream_cls.for_rdflib(options=options)

grouped_stream_to_frames(sink_generator, options=None)

Transform Graphs/Datasets into Jelly frames, one frame per Graph/Dataset.

Note: options are guessed if not provided.

Args: sink_generator (Generator[Graph] | Generator[Dataset]): Generator of Graphs/Dataset to transform. options (SerializerOptions | None, optional): stream options to use. Options are guessed based on the sink store type. Defaults to None.

Yields: Generator[jelly.RdfStreamFrame]: produced Jelly frames

Source code in pyjelly/integrations/rdflib/serialize.py

def grouped_stream_to_frames(
    sink_generator: Generator[Graph] | Generator[Dataset],
    options: SerializerOptions | None = None,
) -> Generator[jelly.RdfStreamFrame]:
    """
    Transform Graphs/Datasets into Jelly frames, one frame per Graph/Dataset.

    Note: options are guessed if not provided.

    Args:
        sink_generator (Generator[Graph] | Generator[Dataset]): Generator of
            Graphs/Dataset to transform.
        options (SerializerOptions | None, optional): stream options to use.
            Options are guessed based on the sink store type. Defaults to None.

    Yields:
        Generator[jelly.RdfStreamFrame]: produced Jelly frames

    """
    stream = None
    for sink in sink_generator:
        if not stream:
            if options is None:
                options = guess_options(sink, grouped=True)
            stream = guess_stream(options, sink)
        yield from stream_frames(stream, sink)

grouped_stream_to_file(stream, output_file, **kwargs)

Write stream of Graphs/Datasets to a binary file.

Args: stream (Generator[Graph] | Generator[Dataset]): Generator of Graphs/Dataset to transform. output_file (IO[bytes]): output buffered writer. **kwargs (Any): options to pass to stream.

Source code in pyjelly/integrations/rdflib/serialize.py

def grouped_stream_to_file(
    stream: Generator[Graph] | Generator[Dataset],
    output_file: IO[bytes],
    **kwargs: Any,
) -> None:
    """
    Write stream of Graphs/Datasets to a binary file.

    Args:
        stream (Generator[Graph] | Generator[Dataset]): Generator of
            Graphs/Dataset to transform.
        output_file (IO[bytes]): output buffered writer.
        **kwargs (Any): options to pass to stream.

    """
    for frame in grouped_stream_to_frames(stream, **kwargs):
        write_delimited(frame, output_file)

flat_stream_to_frames(statements, options=None)

Serialize a stream of raw triples or quads into Jelly frames.

Args: statements (Generator[Triple | Quad]): s/p/o triples or s/p/o/g quads to serialize. options (SerializerOptions | None, optional): if omitted, guessed based on the first tuple.

Yields: Generator[jelly.RdfStreamFrame]: generated frames.

Source code in pyjelly/integrations/rdflib/serialize.py

def flat_stream_to_frames(
    statements: Generator[Triple | Quad],
    options: SerializerOptions | None = None,
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a stream of raw triples or quads into Jelly frames.

    Args:
        statements (Generator[Triple | Quad]):
          s/p/o triples or s/p/o/g quads to serialize.
        options (SerializerOptions | None, optional):
            if omitted, guessed based on the first tuple.

    Yields:
        Generator[jelly.RdfStreamFrame]: generated frames.

    """
    first = next(statements, None)
    if first is None:
        return

    sink = Dataset() if len(first) == QUAD_ARITY else Graph()
    if options is None:
        options = guess_options(sink)
    stream = guess_stream(options, sink)

    yield from stream_frames(stream, chain([first], statements))

flat_stream_to_file(statements, output_file, options=None)

Write Triple or Quad events to a binary file in Jelly flat format.

Args: statements (Generator[Triple | Quad]): statements to serialize. output_file (IO[bytes]): output buffered writer. options (SerializerOptions | None, optional): stream options.

Source code in pyjelly/integrations/rdflib/serialize.py

def flat_stream_to_file(
    statements: Generator[Triple | Quad],
    output_file: IO[bytes],
    options: SerializerOptions | None = None,
) -> None:
    """
    Write Triple or Quad events to a binary file in Jelly flat format.

    Args:
        statements (Generator[Triple | Quad]): statements to serialize.
        output_file (IO[bytes]): output buffered writer.
        options (SerializerOptions | None, optional): stream options.

    """
    for frame in flat_stream_to_frames(statements, options):
        write_delimited(frame, output_file)

sparql

Jelly-SPARQL support for rdflib's SPARQL query results (rdflib.query.Result).

With pyjelly installed, rdflib reads and writes results in the "jelly" format:

result = graph.query("SELECT ?s ?o WHERE { ?s ?p ?o }")
result.serialize("results.jellys", format="jelly")
result = Result.parse(open("results.jellys", "rb"), format="jelly")

Classes:

Name	Description
JellySparqlResult	An rdflib Result read from Jelly-SPARQL, with a few extra attributes.
RDFLibJellySparqlParser	rdflib parser plugin for Jelly-SPARQL results, used by Result.parse().
RDFLibJellySparqlSerializer	rdflib serializer plugin for Jelly-SPARQL, used by Result.serialize().

Functions:

Name	Description
parse_jelly_sparql	Read SPARQL query results from Jelly-SPARQL.
parse_jelly_sparql_sequence	Read a sequence of SPARQL query results from one Jelly-SPARQL stream.
sparql_stream	Create a Jelly-SPARQL writer for rdflib terms.
serialize_jelly_sparql	Write SPARQL query results (of a SELECT or ASK query) in Jelly-SPARQL.
serialize_jelly_sparql_sequence	Write a sequence of SPARQL query results in one Jelly-SPARQL stream.

JellySparqlResult(type_)

Bases: Result

An rdflib Result read from Jelly-SPARQL, with a few extra attributes.

Attributes: links: link IRIs for the results, as in head.link of SPARQL JSON results. truncated: whether the stream ended without a trailer, so some results may be missing. error: error message from the writer, if the results are incomplete. Only set by parse_jelly_sparql_sequence(). parse_jelly_sparql() raises JellyIncompleteResultsError instead.

Source code in pyjelly/integrations/rdflib/sparql.py

def __init__(self, type_: str) -> None:
    super().__init__(type_)
    self.links: list[str] = []
    self.truncated = False
    self.error = ""

RDFLibJellySparqlParser

Bases: ResultParser

rdflib parser plugin for Jelly-SPARQL results, used by Result.parse().

Methods:

Name	Description
parse	Read the results.

parse(source, content_type=None, *, validate_iris=False, lazy=False)

Read the results.

Args: source (IO[bytes] | bytes | str | Path): a stream, bytes, or a file path. content_type (str | None): unused. validate_iris (bool): whether rdflib checks the IRIs, see parse_jelly_sparql(). lazy (bool): read the rows during iteration, see parse_jelly_sparql(). Only for streams, not paths.

Source code in pyjelly/integrations/rdflib/sparql.py

@override
def parse(  # type: ignore[override]
    self,
    source: IO[bytes] | bytes | str | Path,
    content_type: str | None = None,
    *,
    validate_iris: bool = False,
    lazy: bool = False,
) -> Result:
    """
    Read the results.

    Args:
        source (IO[bytes] | bytes | str | Path): a stream, bytes, or a file
            path.
        content_type (str | None): unused.
        validate_iris (bool): whether rdflib checks the IRIs, see
            parse_jelly_sparql().
        lazy (bool): read the rows during iteration, see parse_jelly_sparql().
            Only for streams, not paths.

    """
    if not isinstance(source, (str, Path)):
        return parse_jelly_sparql(source, validate_iris=validate_iris, lazy=lazy)
    if lazy:
        msg = "lazy reading needs a stream that stays open, not a path"
        raise ValueError(msg)
    with Path(source).open("rb") as inp:
        return parse_jelly_sparql(inp, validate_iris=validate_iris)

RDFLibJellySparqlSerializer

Bases: ResultSerializer

rdflib serializer plugin for Jelly-SPARQL, used by Result.serialize().

Methods:

Name	Description
serialize	Write the results.

serialize(stream, encoding='utf-8', *, options=None)

Write the results.

Args: stream (IO[bytes]): output stream. encoding (str): unused, as Jelly is binary. options (SparqlOptions | None): options of the stream.

Source code in pyjelly/integrations/rdflib/sparql.py

@override
def serialize(  # type: ignore[override]
    self,
    stream: IO[bytes],
    encoding: str = "utf-8",
    *,
    options: SparqlOptions | None = None,
) -> None:
    """
    Write the results.

    Args:
        stream (IO[bytes]): output stream.
        encoding (str): unused, as Jelly is binary.
        options (SparqlOptions | None): options of the stream.

    """
    serialize_jelly_sparql(self.result, stream, options)

parse_jelly_sparql(inp, *, validate_iris=False, lazy=False)

Read SPARQL query results from Jelly-SPARQL.

Args: inp (IO[bytes] | bytes): the stream, delimited or not. It must have one result set (the FLAT stream type, the default). validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is still read, rdflib only logs a warning. Checking slows down parsing. lazy (bool): read the rows during iteration instead of all at once. The stream must stay open until the result has been iterated, and the error of incomplete results is raised at the end of the iteration.

Returns: JellySparqlResult: the results, an rdflib Result.

Raises: JellyIncompleteResultsError: if the writer said that the results are incomplete, e.g., because the query failed midway. Its result holds the results read before the error.

Source code in pyjelly/integrations/rdflib/sparql.py

def parse_jelly_sparql(
    inp: IO[bytes] | bytes, *, validate_iris: bool = False, lazy: bool = False
) -> JellySparqlResult:
    """
    Read SPARQL query results from Jelly-SPARQL.

    Args:
        inp (IO[bytes] | bytes): the stream, delimited or not. It must have one
            result set (the FLAT stream type, the default).
        validate_iris (bool): whether rdflib checks the IRIs. An invalid IRI is
            still read, rdflib only logs a warning. Checking slows down parsing.
        lazy (bool): read the rows during iteration instead of all at once. The
            stream must stay open until the result has been iterated, and the
            error of incomplete results is raised at the end of the iteration.

    Returns:
        JellySparqlResult: the results, an rdflib Result.

    Raises:
        JellyIncompleteResultsError: if the writer said that the results are
            incomplete, e.g., because the query failed midway. Its `result` holds
            the results read before the error.

    """
    adapter_factory = _adapter_factory(validate_iris=validate_iris)
    binding_rows = BindingRows()
    try:
        result_set = read_result_set(inp, adapter_factory, row_factory=binding_rows)
    except JellyIncompleteResultsError as error:
        # An incomplete boolean result
        result_set = cast("SparqlResultSet", error.result)
        error.result = _to_result(
            result_set, binding_rows, lazy=False, raise_error=False
        )
        raise
    return _to_result(result_set, binding_rows, lazy=lazy, raise_error=True)

parse_jelly_sparql_sequence(inp, *, validate_iris=False)

Read a sequence of SPARQL query results from one Jelly-SPARQL stream.

This reads PUNCTUATED streams, which hold any number of results. A FLAT stream (the default) holds one.

An incomplete result does not stop the sequence. Instead, its error attribute is set, so check it. The results after it are still read.

Args: inp (IO[bytes] | bytes): the stream, delimited. validate_iris (bool): whether rdflib checks the IRIs, see parse_jelly_sparql().

Yields: JellySparqlResult: the results, an rdflib Result for each result set.

Source code in pyjelly/integrations/rdflib/sparql.py

def parse_jelly_sparql_sequence(
    inp: IO[bytes] | bytes, *, validate_iris: bool = False
) -> Iterator[JellySparqlResult]:
    """
    Read a sequence of SPARQL query results from one Jelly-SPARQL stream.

    This reads PUNCTUATED streams, which hold any number of results. A FLAT
    stream (the default) holds one.

    An incomplete result does not stop the sequence. Instead, its `error`
    attribute is set, so check it. The results after it are still read.

    Args:
        inp (IO[bytes] | bytes): the stream, delimited.
        validate_iris (bool): whether rdflib checks the IRIs, see
            parse_jelly_sparql().

    Yields:
        JellySparqlResult: the results, an rdflib Result for each result set.

    """
    adapter_factory = _adapter_factory(validate_iris=validate_iris)
    binding_rows = BindingRows()
    for result_set in read_result_sets(inp, adapter_factory, row_factory=binding_rows):
        yield _to_result(result_set, binding_rows, lazy=False, raise_error=False)

sparql_stream(options=None)

Create a Jelly-SPARQL writer for rdflib terms.

Source code in pyjelly/integrations/rdflib/sparql.py

def sparql_stream(options: SparqlOptions | None = None) -> SparqlStream:
    """Create a Jelly-SPARQL writer for rdflib terms."""
    if options is None:
        options = SparqlOptions()
    encoder = RDFLibTermEncoder(lookup_preset=options.lookup_preset)
    return SparqlStream(encoder, options)

serialize_jelly_sparql(result, out, options=None)

Write SPARQL query results (of a SELECT or ASK query) in Jelly-SPARQL.

If the results fail midway (e.g., the query fails), the stream is ended with the error so readers know the results are incomplete, and the exception is re-raised. Results read from Jelly-SPARQL that were incomplete or truncated are written as incomplete.

Args: result (Result): the results. out (IO[bytes]): output stream. options (SparqlOptions | None): options of the stream.

Source code in pyjelly/integrations/rdflib/sparql.py

def serialize_jelly_sparql(
    result: Result, out: IO[bytes], options: SparqlOptions | None = None
) -> None:
    """
    Write SPARQL query results (of a SELECT or ASK query) in Jelly-SPARQL.

    If the results fail midway (e.g., the query fails), the stream is ended with
    the error so readers know the results are incomplete, and the exception is
    re-raised. Results read from Jelly-SPARQL that were incomplete or truncated
    are written as incomplete.

    Args:
        result (Result): the results.
        out (IO[bytes]): output stream.
        options (SparqlOptions | None): options of the stream.

    """
    stream = sparql_stream(options)
    _write_result(stream, result, FrameWriter(out, delimited=stream.options.delimited))

serialize_jelly_sparql_sequence(results, out, options=None)

Write a sequence of SPARQL query results in one Jelly-SPARQL stream.

The stream is PUNCTUATED, with one result set per result.

Args: results (Iterable[Result]): the results. out (IO[bytes]): output stream. options (SparqlOptions | None): options of the stream, which must be punctuated. Defaults to SparqlOptions(punctuated=True).

Source code in pyjelly/integrations/rdflib/sparql.py

def serialize_jelly_sparql_sequence(
    results: Iterable[Result], out: IO[bytes], options: SparqlOptions | None = None
) -> None:
    """
    Write a sequence of SPARQL query results in one Jelly-SPARQL stream.

    The stream is PUNCTUATED, with one result set per result.

    Args:
        results (Iterable[Result]): the results.
        out (IO[bytes]): output stream.
        options (SparqlOptions | None): options of the stream, which must be
            punctuated. Defaults to SparqlOptions(punctuated=True).

    """
    if options is None:
        options = SparqlOptions(punctuated=True)
    if not options.punctuated:
        msg = "a sequence of results needs SparqlOptions(punctuated=True)"
        raise ValueError(msg)
    stream = sparql_stream(options)
    writer = FrameWriter(out, delimited=True)
    for result in results:
        _write_result(stream, result, writer)

statements

The statements and prefixes that the rdflib parser yields.

Not compiled with mypyc, which can't compile subclasses of tuple. The classes have empty __slots__, so their instances have no __dict__ and are as small as plain tuples.

Classes:

Name	Description
Triple	Describe RDFLib triple.
Quad	Describe RDFLib quad.
Prefix	Describe RDF Prefix(i.e, namespace declaration).

Triple

Bases: tuple[Node, Node, Node]

Describe RDFLib triple.

Args: tuple (Node, Node, Node): s/p/o tuple of RDFLib Nodes.

Returns: Triple: triple as tuple.

Attributes:

Name	Type	Description
s	Node	The subject.
p	Node	The predicate.
o	Node	The object.

s

The subject.

p

The predicate.

o

The object.

Quad

Bases: tuple[Node, Node, Node, GraphName]

Describe RDFLib quad.

Args: tuple (Node, Node, Node, GraphName): s/p/o/g as a tuple of RDFLib nodes and a GraphName,

Returns: Quad: quad as tuple.

Attributes:

Name	Type	Description
s	Node	The subject.
p	Node	The predicate.
o	Node	The object.
g	GraphName	The graph name.

s

The subject.

p

The predicate.

o

The object.

g

The graph name.

Prefix

Bases: tuple[str, URIRef]

Describe RDF Prefix(i.e, namespace declaration).

Args: tuple (str, rdflib.URIRef): expects prefix as a string, and full namespace URI as Rdflib.URIRef.

Returns: Prefix: prefix as tuple(prefix, iri).

Attributes:

Name	Type	Description
prefix	str	The prefix.
iri	URIRef	The namespace IRI.

prefix

The prefix.

iri

The namespace IRI.

walk

Walk rdflib graphs grouped by graph, subject and predicate (gspo order).

Jelly writes a term only when it differs from the same term in the previous statement, and keeps recently used IRIs in lookup tables. Both work much better when statements are sorted by their terms.

Functions:

Name	Description
spo_tree	Return the graph's triples as a subject -> predicate -> objects tree.
gspo_trees	Return the dataset's graphs, each as its name and its triples as a tree.
dataset_quads	Return the dataset's quads, with graph names rather than Graph objects.
write_tree	Write the statements of a subject -> predicate -> objects tree.
write_trees	Write the statements of graphs, see gspo_trees() and write_tree().

spo_tree(graph)

Return the graph's triples as a subject -> predicate -> objects tree.

Only for rdflib's in-memory store, whose triples are in memory anyway. For any other store, return None: such a store can be far too big to group in memory. Its triples are then written in the store's own order, which for stores with indexes (e.g., Oxigraph, BerkeleyDB) is usually already grouped by subject.

Source code in pyjelly/integrations/rdflib/walk.py

def spo_tree(graph: Graph) -> Tree | None:
    """
    Return the graph's triples as a subject -> predicate -> objects tree.

    Only for rdflib's in-memory store, whose triples are in memory anyway. For any
    other store, return None: such a store can be far too big to group in memory.
    Its triples are then written in the store's own order, which for stores with
    indexes (e.g., Oxigraph, BerkeleyDB) is usually already grouped by subject.
    """
    store = graph.store
    if type(store) is not Memory:
        return None
    indexes = _memory_indexes(store)
    if indexes is None:
        # The private fields are not there: group with the public API
        return _group(graph.triples((None, None, None)))
    contexts = indexes.context_triples
    context = indexes.context_key(graph)
    triples = contexts.get(context)
    if not triples:
        return {}
    if not any(key not in (None, context) and value for key, value in contexts.items()):
        # The graph is alone in its store, so the store's subject-predicate-object
        # index contains its triples, already grouped and in the order they
        # were added
        return indexes.spo
    return _group(triples)

gspo_trees(dataset)

Return the dataset's graphs, each as its name and its triples as a tree.

Only for rdflib's in-memory store, see spo_tree(). Each graph is grouped only when it's its turn, so at most one graph's tree is in memory at a time. The default graph is named DATASET_DEFAULT_GRAPH_ID, see dataset_quads().

Source code in pyjelly/integrations/rdflib/walk.py

def gspo_trees(dataset: ConjunctiveGraph) -> Iterator[tuple[Any, Tree]] | None:
    """
    Return the dataset's graphs, each as its name and its triples as a tree.

    Only for rdflib's in-memory store, see spo_tree(). Each graph is grouped only
    when it's its turn, so at most one graph's tree is in memory at a time. The
    default graph is named DATASET_DEFAULT_GRAPH_ID, see dataset_quads().
    """
    store = dataset.store
    if type(store) is not Memory:
        return None
    indexes = _memory_indexes(store)
    if indexes is None:
        # The private fields are not there: group with the public API
        return _group_quads(dataset_quads(dataset))
    return _memory_gspo_trees(indexes, _default_graph_name(dataset))

dataset_quads(dataset)

Return the dataset's quads, with graph names rather than Graph objects.

The default graph is named DATASET_DEFAULT_GRAPH_ID. A Dataset already names it so, but a ConjunctiveGraph (e.g., the one rdfpipe reads into) names it with a random blank node. The quads of that graph go into the default graph, as with rdflib's TriG serializer.

Source code in pyjelly/integrations/rdflib/walk.py

def dataset_quads(dataset: ConjunctiveGraph) -> Iterator[tuple[Any, Any, Any, Any]]:
    """
    Return the dataset's quads, with graph names rather than Graph objects.

    The default graph is named DATASET_DEFAULT_GRAPH_ID. A Dataset already names
    it so, but a ConjunctiveGraph (e.g., the one rdfpipe reads into) names it
    with a random blank node. The quads of that graph go into the default graph,
    as with rdflib's TriG serializer.
    """
    default = _default_graph_name(dataset)
    for s, p, o, g in dataset.quads((None, None, None, None)):
        name = getattr(g, "identifier", g)
        if name is None or name == default:
            name = DATASET_DEFAULT_GRAPH_ID
        yield s, p, o, name

write_tree(stream, tree)

Write the statements of a subject -> predicate -> objects tree.

Uses the stream's cursor API, so the stream doesn't compare terms. In a tree, each subject differs from the one before it, and so do the predicates of one subject and the objects of one predicate. Only the first subject of the tree and the first predicate of a subject can be equal to the one before it (e.g., rdf:type as the first predicate of every subject), so only those are compared. The first object of a predicate is rarely equal to the object before it, and most predicates have only one object, so objects are not compared: an equal one is just written again.

Yields: jelly.RdfStreamFrame: full frames.

Source code in pyjelly/integrations/rdflib/walk.py

def write_tree(stream: Stream, tree: Tree) -> Iterator[jelly.RdfStreamFrame]:
    """
    Write the statements of a subject -> predicate -> objects tree.

    Uses the stream's cursor API, so the stream doesn't compare terms. In a tree,
    each subject differs from the one before it, and so do the predicates of one
    subject and the objects of one predicate. Only the first subject of the tree
    and the first predicate of a subject can be equal to the one before it (e.g.,
    rdf:type as the first predicate of every subject), so only those are
    compared. The first object of a predicate is rarely equal to the object
    before it, and most predicates have only one object, so objects are not
    compared: an equal one is just written again.

    Yields:
        jelly.RdfStreamFrame: full frames.

    """
    encoder = stream.encoder
    terms = stream.repeated_terms
    first_subject = True
    for s, predicates in tree.items():
        if not first_subject or not encoder.same_term(terms[0], s):
            stream.set_subject(s)
        first_subject = False
        first_predicate = True
        for p, objects in predicates.items():
            if not first_predicate or not encoder.same_term(terms[1], p):
                stream.set_predicate(p)
            first_predicate = False
            for o in objects:
                stream.set_object(o)
                frame = stream.emit()
                if frame is not None:
                    yield frame

write_trees(stream, trees)

Write the statements of graphs, see gspo_trees() and write_tree().

Source code in pyjelly/integrations/rdflib/walk.py

def write_trees(
    stream: QuadStream, trees: Iterable[tuple[Any, Tree]]
) -> Iterator[jelly.RdfStreamFrame]:
    """Write the statements of graphs, see gspo_trees() and write_tree()."""
    for g, tree in trees:
        if not stream.encoder.same_term(stream.repeated_terms[3], g):
            stream.set_graph(g)
        yield from write_tree(stream, tree)

jelly

Modules:

Name	Description
rdf_pb2	Generated protocol buffer code.
sparql_pb2	Generated protocol buffer code.

rdf_pb2

Generated protocol buffer code.

sparql_pb2

Generated protocol buffer code.

options

Classes:

Name	Description
LookupPreset	Sizes of the lookups of a stream.
StreamTypes
StreamParameters	Parameters of a stream, other than its types and lookup sizes.
SparqlOptions	Options for writing a Jelly-SPARQL stream.

Functions:

Name	Description
is_row_layout	Return whether streams of a protocol version use the row layout.
register_mimetypes	Associate files that have Jelly extension with Jelly MIME types.
stream_type_for	Return the Jelly-RDF 1.2 stream type of a stream that is written.

Attributes:

Name	Type	Description
INTEGRATION_SIDE_EFFECTS	bool	Whether to allow integration module imports to trigger side effects.

INTEGRATION_SIDE_EFFECTS = True

Whether to allow integration module imports to trigger side effects.

These side effects are cheap and may include populating some registries for guessing the defaults for external integrations that work with Jelly.

LookupPreset(max_names=DEFAULT_NAME_LOOKUP_SIZE, max_prefixes=DEFAULT_PREFIX_LOOKUP_SIZE, max_datatypes=DEFAULT_DATATYPE_LOOKUP_SIZE)

Sizes of the lookups of a stream.

The defaults are those of Jelly-RDF. Jelly-SPARQL streams use larger lookups by default, see SparqlOptions.

Attributes: max_names: size of the name lookup, at least 8. max_prefixes: size of the prefix lookup, 0 to not use prefixes. max_datatypes: size of the datatype lookup.

Methods:

Name	Description
small	Return small lookups: 128 names, 32 prefixes and 32 datatypes.

small()

Return small lookups: 128 names, 32 prefixes and 32 datatypes.

Source code in pyjelly/options.py

@classmethod
def small(cls) -> Self:
    """Return small lookups: 128 names, 32 prefixes and 32 datatypes."""
    return cls(max_names=128, max_prefixes=32, max_datatypes=32)

StreamTypes(physical_type=jelly.PHYSICAL_STREAM_TYPE_UNSPECIFIED, logical_type=jelly.LOGICAL_STREAM_TYPE_UNSPECIFIED)

Methods:

Name	Description
__repr__	Return the representation of StreamTypes.

__repr__()

Return the representation of StreamTypes.

repr(StreamTypes(9999, 8888)) 'StreamTypes(9999, 8888)'

Source code in pyjelly/options.py

def __repr__(self) -> str:
    """
    Return the representation of StreamTypes.

    >>> repr(StreamTypes(9999, 8888))
    'StreamTypes(9999, 8888)'
    """
    with suppress(ValueError):
        physical_type_name = jelly.PhysicalStreamType.Name(self.physical_type)
        logical_type_name = jelly.LogicalStreamType.Name(self.logical_type)
        return f"StreamTypes({physical_type_name}, {logical_type_name})"
    return f"StreamTypes({self.physical_type}, {self.logical_type})"

StreamParameters(generalized_statements=False, rdf_star=False, version=MAX_VERSION, delimited=True, namespace_declarations=False, stream_name='', stream_type=jelly.RDF_STREAM_TYPE_FLAT, rdf_version=jelly.RDF_VERSION_UNSPECIFIED)

Parameters of a stream, other than its types and lookup sizes.

By default, streams are written in Jelly-RDF 1.2. Set version to 1 or 2 to write Jelly-RDF 1.0 or 1.1 instead, which older parsers can read.

Attributes: generalized_statements: Jelly-RDF 1.0 and 1.1 only: whether the stream may have generalized statements. rdf_star: Jelly-RDF 1.0 and 1.1 only: whether the stream may have RDF-star statements. version: protocol version. delimited: whether the frames are delimited. namespace_declarations: whether the stream has namespace declarations. Not available in Jelly-RDF 1.0. stream_name: name of the stream. stream_type: Jelly-RDF 1.2 only: whether the stream is one graph or dataset (FLAT), or a sequence of RDF Messages (MESSAGES). rdf_version: Jelly-RDF 1.2 only: the version of RDF whose terms may occur in the stream. RDF 1.1 does not allow base directions or triple terms, RDF 1.2 Basic does not allow triple terms.

SparqlOptions(lookup_preset=SPARQL_LOOKUP_PRESET, rdf_version=jelly.RDF_VERSION_UNSPECIFIED, punctuated=False, frame_values=DEFAULT_SPARQL_FRAME_VALUES, delimited=True, stream_name='')

Options for writing a Jelly-SPARQL stream.

Attributes: lookup_preset: lookup sizes. The name lookup needs at least 128 entries. rdf_version: RDF version of the terms in the stream. RDF 1.1 allows no base directions or triple terms, RDF 1.2 Basic no triple terms. punctuated: write a sequence of result sets (the PUNCTUATED stream type) instead of one result set. frame_values: maximum values (rows times variables) per frame. A frame can end earlier if the lookups fill up. delimited: write delimited frames, which files and HTTP need. A non-delimited stream is a single frame, e.g. one message for a message broker. All terms of the results must then fit in the lookups. stream_name: name of the stream.

Methods:

Name	Description
rows_per_frame	Compute the maximum rows per frame for this number of variables.

rows_per_frame(variables)

Compute the maximum rows per frame for this number of variables.

SparqlOptions().rows_per_frame(4) 1024 SparqlOptions(frame_values=10).rows_per_frame(0) 10

Source code in pyjelly/options.py

def rows_per_frame(self, variables: int) -> int:
    """
    Compute the maximum rows per frame for this number of variables.

    >>> SparqlOptions().rows_per_frame(4)
    1024
    >>> SparqlOptions(frame_values=10).rows_per_frame(0)
    10
    """
    return max(1, self.frame_values // max(1, variables))

is_row_layout(version)

Return whether streams of a protocol version use the row layout.

is_row_layout(VERSION_1_1) True is_row_layout(VERSION_1_2) False

Source code in pyjelly/options.py

def is_row_layout(version: int) -> bool:
    """
    Return whether streams of a protocol version use the row layout.

    >>> is_row_layout(VERSION_1_1)
    True
    >>> is_row_layout(VERSION_1_2)
    False
    """
    return version < VERSION_1_2

register_mimetypes(extension='.jelly')

Associate files that have Jelly extension with Jelly MIME types.

register_mimetypes() mimetypes.guess_type("out.jelly") ('application/x-jelly-rdf', None)

Source code in pyjelly/options.py

def register_mimetypes(extension: str = ".jelly") -> None:
    """
    Associate files that have Jelly extension with Jelly MIME types.

    >>> register_mimetypes()
    >>> mimetypes.guess_type("out.jelly")
    ('application/x-jelly-rdf', None)
    """
    for mimetype in MIMETYPES:
        mimetypes.add_type(mimetype, extension)

stream_type_for(params, logical_type)

Return the Jelly-RDF 1.2 stream type of a stream that is written.

As in Jelly-JVM, a stream of graphs or datasets (the logical types GRAPHS and DATASETS, and their subtypes) is a stream of RDF Messages, one per graph or dataset.

stream_type_for(StreamParameters(), jelly.LOGICAL_STREAM_TYPE_FLAT_TRIPLES) 0 stream_type_for(StreamParameters(), jelly.LOGICAL_STREAM_TYPE_NAMED_GRAPHS) 1

Source code in pyjelly/options.py

def stream_type_for(
    params: StreamParameters, logical_type: jelly.LogicalStreamType
) -> jelly.RdfStreamType:
    """
    Return the Jelly-RDF 1.2 stream type of a stream that is written.

    As in Jelly-JVM, a stream of graphs or datasets (the logical types GRAPHS and
    DATASETS, and their subtypes) is a stream of RDF Messages, one per graph or
    dataset.

    >>> stream_type_for(StreamParameters(), jelly.LOGICAL_STREAM_TYPE_FLAT_TRIPLES)
    0
    >>> stream_type_for(StreamParameters(), jelly.LOGICAL_STREAM_TYPE_NAMED_GRAPHS)
    1
    """
    if params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES or logical_type % 10 in (
        jelly.LOGICAL_STREAM_TYPE_GRAPHS,
        jelly.LOGICAL_STREAM_TYPE_DATASETS,
    ):
        return jelly.RDF_STREAM_TYPE_MESSAGES
    return jelly.RDF_STREAM_TYPE_FLAT

parse

Modules:

Name	Description
columns	Decoding of the column layout of Jelly-RDF 1.2.
decode
ioutils
lookup
sparql	Jelly-SPARQL reader: SPARQL query results in frames, with one column per variable.

columns

Decoding of the column layout of Jelly-RDF 1.2.

The terms are decoded once per run, and the runs are only expanded into statements at the end, so the memory needed does not depend on the row count declared by the frame.

Classes:

Name	Description
MessageEnd	The end of an RDF Message, among the rows of Decoder.iter_rows().
IriSequence	IRIs written as RdfIri messages, in the order they are read.
ColumnDecoder	Decodes the frames of a Jelly-RDF 1.2 stream for a Decoder.

MessageEnd

The end of an RDF Message, among the rows of Decoder.iter_rows().

IriSequence(decoder)

IRIs written as RdfIri messages, in the order they are read.

A prefix index of 0 means the same prefix as the IRI before (none in the first IRI), and a name index of 0 means the name after the one of the IRI before.

Source code in pyjelly/parse/columns.py

def __init__(self, decoder: Decoder) -> None:
    self.decoder = decoder
    self.prefix_index = 0
    self.name_index = 0

ColumnDecoder(decoder)

Decodes the frames of a Jelly-RDF 1.2 stream for a Decoder.

Methods:

Name	Description
iter_frame	Return the namespace declarations and statements of a frame.
restate_options	Read restated stream options, which empty the lookups.
end_of_stream	Yield the end of the last RDF Message, if it has statements.
iter_batch	Return the namespace declarations and statements of a batch.
message_lengths	Return the lengths of the RDF Messages that end in a batch.
apply_entries	Apply a frame's packed lookup entries.
statements	Yield the statements, counts at a time, with MESSAGE_END in between.
runs	Decode a column into its runs, whose cells add up to the row count.
cells	Decode a Jelly-SPARQL column into its cells, in order.
run_values	Decode the run values of a column, in order.
mixed_values	Decode the run values of a column whose kinds say which type each is.
literal_kind	Resolve a literal kind: its language tag, datatype and base direction.
direction	Return a base direction read from the stream.

Source code in pyjelly/parse/columns.py

def __init__(self, decoder: Decoder) -> None:
    self.decoder = decoder
    options = decoder.options
    self.quads = options.stream_types.physical_type == (
        jelly.PHYSICAL_STREAM_TYPE_QUADS
    )
    self.messages = options.params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES
    # Whether the stream options were read: those the decoder was made with
    self.options_read = False
    # Statements of the current RDF Message so far, in this and earlier frames
    self.rows_in_message = 0
    #: Prefix for blank node labels, so that equal labels in different result
    #: sets give different blank nodes (see pyjelly.parse.sparql)
    self.bnode_scope = ""

iter_frame(frame)

Return the namespace declarations and statements of a frame.

The stream options and the lookup entries of the frame are read right away, the statements as they are iterated.

Source code in pyjelly/parse/columns.py

def iter_frame(self, frame: jelly.RdfStreamFrame) -> Iterator[Any]:
    """
    Return the namespace declarations and statements of a frame.

    The stream options and the lookup entries of the frame are read right away,
    the statements as they are iterated.
    """
    message_ended = False
    rows = frame.rows
    if rows:
        if len(rows) > 1 or _ROW_WHICH_ONEOF(rows[0], "row") != "options":
            msg = (
                "a frame of a Jelly-RDF 1.2 stream can only have the stream "
                "options in its rows, in one row"
            )
            raise JellyConformanceError(msg)
        if self.options_read:
            message_ended = self.restate_options(rows[0].options)
        else:
            self.options_read = True
    if not _FRAME_HAS_FIELD(frame, "columns"):
        return iter((MESSAGE_END,) if message_ended else ())
    if not self.options_read:
        msg = "the stream has statements before the stream options"
        raise JellyConformanceError(msg)
    batch_rows = self.iter_batch(frame.columns)
    if message_ended:
        return chain((MESSAGE_END,), batch_rows)
    return batch_rows

restate_options(options)

Read restated stream options, which empty the lookups.

Returns: bool: whether an RDF Message ended, because it had statements.

Source code in pyjelly/parse/columns.py

def restate_options(self, options: jelly.RdfStreamOptions) -> bool:
    """
    Read restated stream options, which empty the lookups.

    Returns:
        bool: whether an RDF Message ended, because it had statements.

    """
    current = self.decoder.options
    if (
        options.version != current.params.version
        or options.physical_type != current.stream_types.physical_type
        or options.stream_type != current.params.stream_type
    ):
        msg = (
            "the protocol version, physical type and stream type must be the same "
            "in all stream options of a stream"
        )
        raise JellyConformanceError(msg)
    self.decoder.restate_options(options)
    if self.messages and self.rows_in_message:
        self.rows_in_message = 0
        return True
    return False

end_of_stream()

Yield the end of the last RDF Message, if it has statements.

Source code in pyjelly/parse/columns.py

def end_of_stream(self) -> Iterator[Any]:
    """Yield the end of the last RDF Message, if it has statements."""
    if self.messages and self.rows_in_message:
        self.rows_in_message = 0
        yield MESSAGE_END

iter_batch(batch)

Return the namespace declarations and statements of a batch.

Source code in pyjelly/parse/columns.py

def iter_batch(self, batch: jelly.RdfColumnBatch) -> Iterator[Any]:
    """Return the namespace declarations and statements of a batch."""
    rows = batch.row_count
    if rows > MAX_ROWS:
        msg = f"a frame can have at most {MAX_ROWS} statements, not {rows}"
        raise JellyConformanceError(msg)
    lengths = self.message_lengths(batch, rows)
    # All lookup entries come first, so that the columns can refer to them
    self.ingest_entries(batch)
    namespaces = self.namespaces(batch) if batch.namespaces else []
    has_graphs = _BATCH_HAS_FIELD(batch, "graphs")
    if has_graphs and not self.quads:
        msg = "a stream of triples cannot have a graphs column"
        raise JellyConformanceError(msg)
    subjects = self.runs(batch.subjects, rows, _RESOURCES, "subject")
    predicates = self.runs(batch.predicates, rows, _IRIS, "predicate")
    objects = self.runs(batch.objects, rows, _ANY, "object")
    graphs = None
    if self.quads:
        if has_graphs:
            graphs = self.runs(batch.graphs, rows, _RESOURCES, "graph")
        else:
            graphs = _Runs([], rows, "graph")
            graphs.add(None, rows)
        graphs.set_unbound(self.decoder.adapter.default_graph())

    # The statements of each RDF Message that ends in the batch, and the rest
    counts = [*lengths, rows - sum(lengths)]
    if lengths:
        self.rows_in_message = counts[-1]
    else:
        self.rows_in_message += rows
    statements = self.statements(counts, subjects, predicates, objects, graphs)
    if namespaces:
        return chain(namespaces, statements)
    return statements

message_lengths(batch, rows)

Return the lengths of the RDF Messages that end in a batch.

Source code in pyjelly/parse/columns.py

def message_lengths(self, batch: jelly.RdfColumnBatch, rows: int) -> list[int]:
    """Return the lengths of the RDF Messages that end in a batch."""
    lengths = batch.message_lengths[:]
    if lengths:
        if not self.messages:
            msg = "message boundaries are only valid in streams of RDF Messages"
            raise JellyConformanceError(msg)
        if sum(lengths) > rows:
            msg = (
                f"the message lengths add up to {sum(lengths)}, more than the "
                f"{rows} statements of the frame"
            )
            raise JellyConformanceError(msg)
    return lengths

apply_entries(names, prefixes, datatypes)

Apply a frame's packed lookup entries.

Source code in pyjelly/parse/columns.py

def apply_entries(self, names: Any, prefixes: Any, datatypes: Any) -> None:
    """Apply a frame's packed lookup entries."""
    decoder = self.decoder
    _apply_entries(names, decoder.names)
    _apply_entries(prefixes, decoder.prefixes)
    for entry in datatypes:
        for value in entry.values:
            if value in (LANG_STRING_DATATYPE_IRI, DIR_LANG_STRING_DATATYPE_IRI):
                msg = f"the datatype lookup cannot have {value}"
                raise JellyConformanceError(msg)
    _apply_entries(datatypes, decoder.datatypes)

statements(counts, subjects, predicates, objects, graphs)

Yield the statements, counts at a time, with MESSAGE_END in between.

Source code in pyjelly/parse/columns.py

def statements(
    self,
    counts: list[int],
    subjects: _Runs,
    predicates: _Runs,
    objects: _Runs,
    graphs: _Runs | None,
) -> Iterator[Any]:
    """Yield the statements, `counts` at a time, with MESSAGE_END in between."""
    adapter = self.decoder.adapter
    last = len(counts) - 1
    for i, count in enumerate(counts):
        if graphs is None:
            for _ in range(count):
                yield adapter.triple(
                    (subjects.next(), predicates.next(), objects.next())
                )
        else:
            for _ in range(count):
                yield adapter.quad(
                    (
                        subjects.next(),
                        predicates.next(),
                        objects.next(),
                        graphs.next(),
                    )
                )
        if i < last:
            yield MESSAGE_END

runs(column, rows, allowed, position)

Decode a column into its runs, whose cells add up to the row count.

Source code in pyjelly/parse/columns.py

def runs(
    self, column: jelly.RdfColumn, rows: int, allowed: int, position: str
) -> _Runs:
    """Decode a column into its runs, whose cells add up to the row count."""
    values = self.run_values(column, allowed, position)
    unbound_allowed = position == "graph"
    layouts: list[int] = column.layouts[:]
    runs = _Runs(values, rows, position)
    index = 0
    k = 0
    while k < len(layouts):
        token = layouts[k]
        k += 1
        length = token & _LENGTH_MASK
        if length == _LENGTH_MASK:
            if k == len(layouts):
                msg = (
                    f"corrupt layout of the {position} column: a length is cut off"
                )
                raise JellyConformanceError(msg)
            length += layouts[k]
            k += 1
        skip = token >> _SKIP_SHIFT
        if index + skip > len(values):
            msg = (
                f"corrupt layout of the {position} column: it skips past its values"
            )
            raise JellyConformanceError(msg)
        runs.add_values(index, skip)
        index += skip
        if token & _UNBOUND:
            if not unbound_allowed:
                raise _unbound(position)
            runs.add(None, length + 1)
        elif index < len(values):
            runs.add(values[index], length + 2)
            index += 1
        else:
            msg = (
                f"corrupt layout of the {position} column: a repeated value is "
                "past its values"
            )
            raise JellyConformanceError(msg)
    runs.add_values(index, len(values) - index)
    if runs.cells < rows:
        if not unbound_allowed:
            raise _unbound(position)
        runs.add(None, rows - runs.cells)
    return runs

cells(column, rows, position)

Decode a Jelly-SPARQL column into its cells, in order.

The column may hold terms of any type. Unbound cells are None.

Source code in pyjelly/parse/columns.py

def cells(self, column: jelly.RdfColumn, rows: int, position: str) -> list[Any]:
    """
    Decode a Jelly-SPARQL column into its cells, in order.

    The column may hold terms of any type. Unbound cells are None.
    """
    values = self.run_values(column, _ANY, position)
    layouts: list[int] = column.layouts[:]
    # Without layout tokens, each value fills one cell
    cells = _layout_cells(values, layouts, rows, position) if layouts else values
    if len(cells) > rows:
        raise _too_many_cells(position)
    if len(cells) < rows:
        # Pad with unbound cells
        cells += [None] * (rows - len(cells))
    return cells

run_values(column, allowed, position)

Decode the run values of a column, in order.

Source code in pyjelly/parse/columns.py

def run_values(
    self, column: jelly.RdfColumn, allowed: int, position: str
) -> list[Any]:
    """Decode the run values of a column, in order."""
    counts = (
        len(column.name_ids),
        len(column.lex_values),
        len(column.bnodes),
        len(column.triple_terms),
    )
    for kind in range(4):
        if counts[kind] and not allowed & (1 << kind):
            msg = f"the {position} column cannot have {_KIND_NAMES[kind]}"
            raise JellyConformanceError(msg)
    if column.kinds:
        return self.mixed_values(column, counts, position)
    if sum(1 for count in counts if count) > 1:
        msg = f"the {position} column has terms of more than one type, but no kinds"
        raise JellyConformanceError(msg)
    if counts[_KIND_LITERAL]:
        return self.literals(column)
    if counts[_KIND_BNODE]:
        return self.bnodes(column)
    if counts[_KIND_TRIPLE]:
        return self.triple_terms(column)
    return self.iris(column)

mixed_values(column, counts, position)

Decode the run values of a column whose kinds say which type each is.

Source code in pyjelly/parse/columns.py

def mixed_values(
    self, column: jelly.RdfColumn, counts: tuple[int, ...], position: str
) -> list[Any]:
    """Decode the run values of a column whose kinds say which type each is."""
    kinds: bytes = column.kinds
    value_count = sum(counts)
    _check_kinds(kinds, value_count, position)
    iris = self.iris(column)
    literals = self.literals(column)
    bnodes = self.bnodes(column)
    triple_terms = self.triple_terms(column)
    # How many values of each type were taken. If no type runs out of values,
    # all are used up at the end, as there are value_count of them
    iri = literal = bnode = triple = 0
    values: list[Any] = []
    for i in range(value_count):
        kind = (kinds[i >> 2] >> ((i & 3) << 1)) & 3
        if kind == _KIND_IRI:
            if iri == len(iris):
                raise _kinds_mismatch(position, kind)
            values.append(iris[iri])
            iri += 1
        elif kind == _KIND_LITERAL:
            if literal == len(literals):
                raise _kinds_mismatch(position, kind)
            values.append(literals[literal])
            literal += 1
        elif kind == _KIND_BNODE:
            if bnode == len(bnodes):
                raise _kinds_mismatch(position, kind)
            values.append(bnodes[bnode])
            bnode += 1
        else:
            if triple == len(triple_terms):
                raise _kinds_mismatch(position, kind)
            values.append(triple_terms[triple])
            triple += 1
    return values

literal_kind(kind, langtags, directions)

Resolve a literal kind: its language tag, datatype and base direction.

Source code in pyjelly/parse/columns.py

def literal_kind(self, kind: int, langtags: Any, directions: Any) -> _LiteralKind:
    """Resolve a literal kind: its language tag, datatype and base direction."""
    decoder = self.decoder
    if not kind:
        return _LiteralKind(decoder, None, None, None)
    if kind & 1:
        index = (kind + 1) >> 1
        if index > decoder.datatypes.lookup_size:
            msg = (
                f"a literal kind refers to datatype {index}, which is outside the "
                "datatype lookup"
            )
            raise JellyConformanceError(msg)
        datatype = decoder.datatypes.at(index)
        return _LiteralKind(decoder, None, datatype, None)
    index = (kind >> 1) - 1
    if index >= len(langtags):
        msg = f"a literal kind refers to language tag {index}, which is not there"
        raise JellyConformanceError(msg)
    direction = None
    if directions and directions[index]:
        direction = self.direction(directions[index])
    return _LiteralKind(decoder, langtags[index] or None, None, direction)

direction(value)

Return a base direction read from the stream.

Source code in pyjelly/parse/columns.py

def direction(self, value: int) -> str:
    """Return a base direction read from the stream."""
    direction = _DIRECTIONS.get(value)
    if direction is None:
        msg = f"unknown base direction: {value}"
        raise JellyConformanceError(msg)
    if self.decoder.options.params.rdf_version == jelly.RDF_VERSION_1_1:
        msg = "the stream declares RDF 1.1, but has a literal with a base direction"
        raise JellyConformanceError(msg)
    return direction

decode

Classes:

Name	Description
ParsingMode	Specifies how jelly frames should be treated.
Adapter
Decoder

Functions:

Name	Description
check_stream_options	Check that the parser can read a stream with these options.
parser_options	Return the parser options of a stream, after checking them.
options_from_frame	Fill stream options based on the options row.
messages	Group the rows of a stream of RDF Messages by message.
grouped_rows	Return the rows of each frame, or of each message in a stream of RDF Messages.
flat_rows	Return the rows of all frames, without the ends of RDF Messages.

ParsingMode

Bases: Enum

Specifies how jelly frames should be treated.

Modes: FLAT Yield all frames as one Graph or Dataset. GROUPED Yield one Graph/Dataset per frame (grouped parsing).

Adapter(options, parsing_mode=ParsingMode.FLAT)

Methods:

Name	Description
iri	Return an IRI term. The decoder reuses it, so it must be immutable.
literal	Return a literal term. The decoder reuses it, so it must be immutable.
directional_literal	Return a language-tagged string with a base direction, "ltr" or "rtl".
namespace_declaration	Return a namespace declaration. iri is a term made by iri().

Source code in pyjelly/parse/decode.py

def __init__(
    self, options: ParserOptions, parsing_mode: ParsingMode = ParsingMode.FLAT
) -> None:
    self.options = options
    self.parsing_mode = parsing_mode

iri(iri)

Return an IRI term. The decoder reuses it, so it must be immutable.

Source code in pyjelly/parse/decode.py

@abstractmethod
def iri(self, iri: str) -> Any:
    """Return an IRI term. The decoder reuses it, so it must be immutable."""
    raise NotImplementedError

literal(lex, language=None, datatype=None)

Return a literal term. The decoder reuses it, so it must be immutable.

Source code in pyjelly/parse/decode.py

@abstractmethod
def literal(
    self,
    lex: str,
    language: str | None = None,
    datatype: str | None = None,
) -> Any:
    """Return a literal term. The decoder reuses it, so it must be immutable."""
    raise NotImplementedError

directional_literal(lex, language, direction)

Return a language-tagged string with a base direction, "ltr" or "rtl".

Source code in pyjelly/parse/decode.py

def directional_literal(self, lex: str, language: str, direction: str) -> Any:  # noqa: ARG002
    """Return a language-tagged string with a base direction, "ltr" or "rtl"."""
    _adapter_missing(
        "decoding literals with a base direction",
        stream_types=self.options.stream_types,
    )

namespace_declaration(name, iri)

Return a namespace declaration. iri is a term made by iri().

Source code in pyjelly/parse/decode.py

def namespace_declaration(self, name: str, iri: Any) -> Any:  # noqa: ARG002
    """Return a namespace declaration. `iri` is a term made by iri()."""
    _adapter_missing(
        "decoding namespace declarations",
        stream_types=self.options.stream_types,
    )

Decoder(adapter)

Initializes decoder with lookup tables of preset sizes and an integration-dependent adapter.

Args: adapter (Adapter): integration-dependent adapter that specifies terms conversion to specific objects, framing, namespace declarations, and graphs/datasets forming.

Methods:

Name	Description
new_lookups	Make empty lookups, of the sizes in the options.
restate_options	Use restated options of a Jelly-RDF 1.2 stream. This empties the lookups.
end_of_stream	Yield what ends with the stream: the last RDF Message, if any.
iter_rows	Iterate through rows in the frame.
iter_row_layout	Yield the decoded rows of a frame of a Jelly-RDF 1.0 or 1.1 stream.
decode_other_row	Decode a row that gives nothing to yield: graph start or end, options.
validate_stream_options	Check that repeated stream options are the same as the first ones.
ingest_prefix_entry	Update prefix lookup table based on the table entry.
ingest_name_entry	Update name lookup table based on the table entry.
ingest_datatype_entry	Update datatype lookup table based on the table entry.
decode_term	Decode a term based on its type: IRI/literal/BN/default graph.
decode_iri	Decode RdfIri message to IRI using a custom adapter.
iri_term	Return the IRI made of the entries at the indices (prefix 0 is none).
new_iri	Make an IRI term, and cache it for its name entry.
decode_bnode	Decode string message to blank node (BN) using a custom adapter.
decode_literal	Decode RdfLiteral to literal based on custom adapter implementation.
literal_term	Return a literal, reusing a recently made one if it's the same.
literal_cache	Return recent literals of a language tag and datatype, by lexical form.
new_literal	Make a literal term, and keep it in its literal_cache().
decode_statement	Decode the terms of a triple or quad message.

Source code in pyjelly/parse/decode.py

def __init__(self, adapter: Adapter) -> None:
    """
    Initialize decoder.

    Initializes decoder with lookup tables of preset sizes and an
    integration-dependent adapter.

    Args:
        adapter (Adapter): integration-dependent adapter that specifies terms
        conversion to specific objects, framing,
        namespace declarations, and graphs/datasets forming.

    """
    self.adapter = adapter
    self.new_lookups()
    # The last s, p, o and g terms, for repeated terms
    self.repeated_terms: list[object] = [_MISSING] * 4
    # Literals often repeat (e.g., the same values in many entities), and
    # making one is often slow (rdflib parses the value). By language tag and
    # datatype, then by lexical form, see literal_cache()
    self.literal_caches: dict[tuple[str | None, str | None], dict[str, Any]] = {}
    self.literal_cache_size = 0

    self.term_handlers: dict[type[Any], TermHandler] = {
        t: getattr(self, name) for t, name in self._TERM_HANDLER_NAMES.items()
    }
    #: Decodes the frames of Jelly-RDF 1.2 streams, None for older ones
    self.column_decoder: ColumnDecoder | None = None
    if not is_row_layout(self.options.params.version):
        self.column_decoder = ColumnDecoder(self)

new_lookups()

Make empty lookups, of the sizes in the options.

Source code in pyjelly/parse/decode.py

def new_lookups(self) -> None:
    """Make empty lookups, of the sizes in the options."""
    lookup_preset = self.options.lookup_preset
    self.names = LookupDecoder(lookup_size=lookup_preset.max_names)
    self.prefixes = LookupDecoder(lookup_size=lookup_preset.max_prefixes)
    self.datatypes = LookupDecoder(lookup_size=lookup_preset.max_datatypes)
    # One IRI term per name table entry: an IRI is often
    # used with the same prefix again, so its term can be reused. Index 0 is
    # unused, as table indices start from 1
    self.iri_cache: list[_CachedIri | None] = [None] * (lookup_preset.max_names + 1)

restate_options(options)

Use restated options of a Jelly-RDF 1.2 stream. This empties the lookups.

Source code in pyjelly/parse/decode.py

def restate_options(self, options: jelly.RdfStreamOptions) -> None:
    """Use restated options of a Jelly-RDF 1.2 stream. This empties the lookups."""
    self.adapter.options = parser_options(
        options, delimited=self.options.params.delimited
    )
    self.new_lookups()

end_of_stream()

Yield what ends with the stream: the last RDF Message, if any.

Source code in pyjelly/parse/decode.py

def end_of_stream(self) -> Iterator[Any]:
    """Yield what ends with the stream: the last RDF Message, if any."""
    column_decoder = self.column_decoder
    if column_decoder is not None:
        yield from column_decoder.end_of_stream()

iter_rows(frame)

Iterate through rows in the frame.

Args: frame (jelly.RdfStreamFrame): jelly frame

Returns: Iterator[Any]: decoded rows, and in streams of RDF Messages, MESSAGE_END at the end of each message

Source code in pyjelly/parse/decode.py

def iter_rows(self, frame: jelly.RdfStreamFrame) -> Iterator[Any]:
    """
    Iterate through rows in the frame.

    Args:
        frame (jelly.RdfStreamFrame): jelly frame

    Returns:
        Iterator[Any]: decoded rows, and in streams of RDF Messages, MESSAGE_END
            at the end of each message

    """
    column_decoder = self.column_decoder
    if column_decoder is not None:
        return column_decoder.iter_frame(frame)
    return self.iter_row_layout(frame)

iter_row_layout(frame)

Yield the decoded rows of a frame of a Jelly-RDF 1.0 or 1.1 stream.

Source code in pyjelly/parse/decode.py

def iter_row_layout(self, frame: jelly.RdfStreamFrame) -> Iterator[Any]:
    """Yield the decoded rows of a frame of a Jelly-RDF 1.0 or 1.1 stream."""
    # Rows are told apart by the name of their field, most common ones first
    for row_owner in frame.rows:
        kind: str | None = _ROW_WHICH_ONEOF(row_owner, "row")
        if kind == "triple":
            yield self.decode_triple(row_owner.triple)
        elif kind == "quad":
            yield self.decode_quad(row_owner.quad)
        elif kind == "name":
            self.ingest_name_entry(row_owner.name)
        elif kind == "prefix":
            self.ingest_prefix_entry(row_owner.prefix)
        elif kind == "datatype":
            self.ingest_datatype_entry(row_owner.datatype)
        elif kind == "namespace":
            yield self.decode_namespace_declaration(row_owner.namespace)
        else:
            self.decode_other_row(row_owner, kind)

decode_other_row(row_owner, kind)

Decode a row that gives nothing to yield: graph start or end, options.

Source code in pyjelly/parse/decode.py

def decode_other_row(self, row_owner: jelly.RdfStreamRow, kind: str | None) -> None:
    """Decode a row that gives nothing to yield: graph start or end, options."""
    if kind == "graph_start":
        self.decode_graph_start(row_owner.graph_start)
    elif kind == "graph_end":
        self.decode_graph_end(row_owner.graph_end)
    elif kind == "options":
        self.validate_stream_options(row_owner.options)
    elif kind is None:
        msg = "a row of the stream is empty"
        raise JellyConformanceError(msg)
    else:
        msg = f"decoder not implemented for row {kind}"
        raise TypeError(msg)

validate_stream_options(options)

Check that repeated stream options are the same as the first ones.

Source code in pyjelly/parse/decode.py

def validate_stream_options(self, options: jelly.RdfStreamOptions) -> None:
    """Check that repeated stream options are the same as the first ones."""
    stream_types, lookup_preset, params = self.options
    if (
        stream_types.physical_type != options.physical_type
        or stream_types.logical_type != options.logical_type
        or params.stream_name != options.stream_name
        or params.version < options.version
        or lookup_preset.max_prefixes != options.max_prefix_table_size
        or lookup_preset.max_datatypes != options.max_datatype_table_size
        or lookup_preset.max_names != options.max_name_table_size
    ):
        msg = "the repeated stream options are not the same as the first ones"
        raise JellyConformanceError(msg)

ingest_prefix_entry(entry)

Update prefix lookup table based on the table entry.

Args: entry (jelly.RdfPrefixEntry): prefix message, containing id and value

Source code in pyjelly/parse/decode.py

def ingest_prefix_entry(self, entry: jelly.RdfPrefixEntry) -> None:
    """
    Update prefix lookup table based on the table entry.

    Args:
        entry (jelly.RdfPrefixEntry): prefix message, containing id and value

    """
    self.prefixes.assign_entry(index=entry.id, value=entry.value)

ingest_name_entry(entry)

Update name lookup table based on the table entry.

Args: entry (jelly.RdfNameEntry): name message, containing id and value

Source code in pyjelly/parse/decode.py

def ingest_name_entry(self, entry: jelly.RdfNameEntry) -> None:
    """
    Update name lookup table based on the table entry.

    Args:
        entry (jelly.RdfNameEntry): name message, containing id and value

    """
    self.names.assign_entry(index=entry.id, value=entry.value)

ingest_datatype_entry(entry)

Update datatype lookup table based on the table entry.

Args: entry (jelly.RdfDatatypeEntry): name message, containing id and value

Source code in pyjelly/parse/decode.py

def ingest_datatype_entry(self, entry: jelly.RdfDatatypeEntry) -> None:
    """
    Update datatype lookup table based on the table entry.

    Args:
        entry (jelly.RdfDatatypeEntry): name message, containing id and value

    """
    self.datatypes.assign_entry(index=entry.id, value=entry.value)

decode_term(term)

Decode a term based on its type: IRI/literal/BN/default graph.

Notes: requires a custom adapter with implemented methods for terms decoding.

Args: term (Any): IRI/literal/BN(string)/Default graph message

Raises: TypeError: raises error if no handler for the term is found

Returns: Any: decoded term (currently, rdflib objects, e.g., rdflib.term.URIRef)

Source code in pyjelly/parse/decode.py

def decode_term(self, term: Any) -> Any:
    """
    Decode a term based on its type: IRI/literal/BN/default graph.

    Notes: requires a custom adapter with implemented methods for terms decoding.

    Args:
        term (Any): IRI/literal/BN(string)/Default graph message

    Raises:
        TypeError: raises error if no handler for the term is found

    Returns:
        Any: decoded term (currently, rdflib objects, e.g., rdflib.term.URIRef)

    """
    decode_term = self.term_handlers.get(type(term))
    if decode_term is None:
        msg = f"decoder not implemented for {type(term)}"
        raise TypeError(msg) from None
    return decode_term(term)

decode_iri(iri)

Decode RdfIri message to IRI using a custom adapter.

The term is reused if the same name and prefix entries were decoded before, so the adapter's IRI terms must be immutable.

Args: iri (jelly.RdfIri): RdfIri message

Returns: Any: IRI, based on adapter implementation, e.g., rdflib.term.URIRef

Source code in pyjelly/parse/decode.py

def decode_iri(self, iri: jelly.RdfIri) -> Any:
    """
    Decode RdfIri message to IRI using a custom adapter.

    The term is reused if the same name and prefix entries were decoded
    before, so the adapter's IRI terms must be immutable.

    Args:
        iri (jelly.RdfIri): RdfIri message

    Returns:
        Any: IRI, based on adapter implementation, e.g., rdflib.term.URIRef

    """
    names = self.names
    name_index = iri.name_id or names.last_reused_index + 1
    name = names.decode_name_term_index(name_index)
    prefix = self.prefixes.decode_prefix_term_index(iri.prefix_id)
    # The lookup tables keep the strings they were given, so the same objects
    # mean the same entries (or at least equal strings, which is just as good)
    cached = self.iri_cache[name_index]
    if cached is not None and cached.name is name and cached.prefix is prefix:
        return cached.term
    return self.new_iri(name_index, name, prefix)

iri_term(prefix_index, name_index)

Return the IRI made of the entries at the indices (prefix 0 is none).

Source code in pyjelly/parse/decode.py

def iri_term(self, prefix_index: int, name_index: int) -> Any:
    """Return the IRI made of the entries at the indices (prefix 0 is none)."""
    name = self.names.at(name_index)
    prefix = self.prefixes.at(prefix_index) if prefix_index else ""
    cached = self.iri_cache[name_index]
    if cached is not None and cached.name is name and cached.prefix is prefix:
        return cached.term
    return self.new_iri(name_index, name, prefix)

new_iri(name_index, name, prefix)

Make an IRI term, and cache it for its name entry.

Source code in pyjelly/parse/decode.py

def new_iri(self, name_index: int, name: str, prefix: str) -> Any:
    """Make an IRI term, and cache it for its name entry."""
    term = self.adapter.iri(iri=prefix + name)
    self.iri_cache[name_index] = _CachedIri(name, prefix, term)
    return term

decode_bnode(bnode)

Decode string message to blank node (BN) using a custom adapter.

Args: bnode (str): blank node id

Returns: Any: blank node object from the custom adapter

Source code in pyjelly/parse/decode.py

def decode_bnode(self, bnode: str) -> Any:
    """
    Decode string message to blank node (BN) using a custom adapter.

    Args:
        bnode (str): blank node id

    Returns:
        Any: blank node object from the custom adapter

    """
    return self.adapter.bnode(bnode)

decode_literal(literal)

Decode RdfLiteral to literal based on custom adapter implementation.

Recently decoded literals are reused, so the adapter's literal terms must be immutable.

Args: literal (jelly.RdfLiteral): RdfLiteral message

Returns: Any: literal returned by the custom adapter

Source code in pyjelly/parse/decode.py

def decode_literal(self, literal: jelly.RdfLiteral) -> Any:
    """
    Decode RdfLiteral to literal based on custom adapter implementation.

    Recently decoded literals are reused, so the adapter's literal terms must
    be immutable.

    Args:
        literal (jelly.RdfLiteral): RdfLiteral message

    Returns:
        Any: literal returned by the custom adapter

    """
    language = datatype = None
    kind = _LITERAL_WHICH_ONEOF(literal, "literalKind")
    if kind == "langtag":
        language = literal.langtag or None
    elif kind == "datatype":
        datatype = self.datatypes.decode_datatype_term_index(literal.datatype)
    lex = literal.lex
    cache = self.literal_cache(language, datatype)
    term = cache.get(lex)
    if term is None:
        term = self.new_literal(lex, language, datatype, cache)
    return term

literal_term(lex, language, datatype, direction)

Return a literal, reusing a recently made one if it's the same.

Source code in pyjelly/parse/decode.py

def literal_term(
    self,
    lex: str,
    language: str | None,
    datatype: str | None,
    direction: str | None,
) -> Any:
    """Return a literal, reusing a recently made one if it's the same."""
    if direction:
        assert language
        return self.adapter.directional_literal(lex, language, direction)
    cache = self.literal_cache(language, datatype)
    term = cache.get(lex)
    if term is None:
        term = self.new_literal(lex, language, datatype, cache)
    return term

literal_cache(language, datatype)

Return recent literals of a language tag and datatype, by lexical form.

Source code in pyjelly/parse/decode.py

def literal_cache(
    self, language: str | None, datatype: str | None
) -> dict[str, Any]:
    """Return recent literals of a language tag and datatype, by lexical form."""
    key = (language, datatype)
    cache = self.literal_caches.get(key)
    if cache is None:
        cache = self.literal_caches[key] = {}
    return cache

new_literal(lex, language, datatype, cache)

Make a literal term, and keep it in its literal_cache().

Source code in pyjelly/parse/decode.py

def new_literal(
    self,
    lex: str,
    language: str | None,
    datatype: str | None,
    cache: dict[str, Any],
) -> Any:
    """Make a literal term, and keep it in its literal_cache()."""
    term = self.adapter.literal(lex=lex, language=language, datatype=datatype)
    if self.literal_cache_size >= LITERAL_CACHE_SIZE:
        for full in self.literal_caches.values():
            full.clear()
        self.literal_cache_size = 0
    cache[lex] = term
    self.literal_cache_size += 1
    return term

decode_statement(fields, size)

Decode the terms of a triple or quad message.

Notes: also updates the repeated terms

Args: fields (list[Any]): the message's fields, from ListFields() size (int): 3 for a triple, 4 for a quad

Raises: JellyConformanceError: if a missing repeated term is encountered

Returns: Any: a list of decoded terms

Source code in pyjelly/parse/decode.py

def decode_statement(self, fields: list[Any], size: int) -> list[Any]:
    """
    Decode the terms of a triple or quad message.

    Notes: also updates the repeated terms

    Args:
        fields (list[Any]): the message's fields, from ListFields()
        size (int): 3 for a triple, 4 for a quad

    Raises:
        JellyConformanceError: if a missing repeated term is encountered

    Returns:
        Any: a list of decoded terms

    """
    repeated = self.repeated_terms
    terms = repeated[:size]
    # The fields come in the order of their numbers, so the terms are decoded
    # in s, p, o, g order. Each term has four fields, numbered from 1:
    # IRI, blank node, literal and quoted triple (s_iri = 1, ..., o_iri = 9),
    # but the graph's are IRI, blank node, default graph and literal
    for field, value in fields:
        number: int = field.number - 1
        position = number >> 2
        kind = number & 3
        if kind == _IRI:
            term = self.decode_iri(value)
        elif kind == _BNODE:
            term = self.adapter.bnode(value)
        elif kind == _LITERAL_OR_DEFAULT_GRAPH:
            if position == _GRAPH:
                term = self.adapter.default_graph()
            else:
                term = self.decode_literal(value)
        elif position == _GRAPH:
            term = self.decode_literal(value)
        else:
            term = self.decode_quoted_triple(value)
        terms[position] = term
        repeated[position] = term
    for i in range(size):
        if terms[i] is _MISSING:
            msg = f"missing repeated term {_POSITIONS[i]}"
            raise JellyConformanceError(msg)
    return terms

check_stream_options(options)

Check that the parser can read a stream with these options.

The fields of one layout must not be set in a stream of the other. The lookups must not be larger than the parser allows (see the MAX_*_LOOKUP_SIZE constants).

Raises: JellyConformanceError: if it can't.

Source code in pyjelly/parse/decode.py

def check_stream_options(options: jelly.RdfStreamOptions) -> None:
    """
    Check that the parser can read a stream with these options.

    The fields of one layout must not be set in a stream of the
    other. The lookups must not be larger than the parser allows (see the
    MAX_*_LOOKUP_SIZE constants).

    Raises:
        JellyConformanceError: if it can't.

    """
    version = options.version
    if not MIN_VERSION <= version <= MAX_VERSION:
        msg = (
            f"unsupported protocol version {version}, "
            f"expected {MIN_VERSION} to {MAX_VERSION}"
        )
        raise JellyConformanceError(msg)
    physical_type = options.physical_type
    if physical_type not in _PHYSICAL_TYPES:
        msg = f"unknown physical stream type: {physical_type}"
        raise JellyConformanceError(msg)
    if is_row_layout(version):
        if options.stream_type or options.rdf_version:
            msg = (
                "the stream type and RDF version are only valid in Jelly-RDF 1.2 "
                "streams (protocol version 3)"
            )
            raise JellyConformanceError(msg)
        if options.logical_type not in _LOGICAL_TYPES:
            msg = f"unknown logical stream type: {options.logical_type}"
            raise JellyConformanceError(msg)
        min_names = MIN_NAME_LOOKUP_SIZE
    else:
        if options.generalized_statements or options.rdf_star or options.logical_type:
            msg = (
                "generalized statements, RDF-star and logical stream types are only "
                "valid in Jelly-RDF 1.0 and 1.1 streams (protocol versions 1 and 2)"
            )
            raise JellyConformanceError(msg)
        if physical_type == jelly.PHYSICAL_STREAM_TYPE_GRAPHS:
            msg = "Jelly-RDF 1.2 streams cannot be of the GRAPHS physical type"
            raise JellyConformanceError(msg)
        if options.stream_type not in (
            jelly.RDF_STREAM_TYPE_FLAT,
            jelly.RDF_STREAM_TYPE_MESSAGES,
        ):
            msg = f"unknown stream type: {options.stream_type}"
            raise JellyConformanceError(msg)
        if options.rdf_version not in _RDF_VERSIONS:
            msg = f"unknown RDF version: {options.rdf_version}"
            raise JellyConformanceError(msg)
        min_names = MIN_COLUMN_NAME_LOOKUP_SIZE
    _check_size("name", options.max_name_table_size, min_names, MAX_NAME_LOOKUP_SIZE)
    _check_size("prefix", options.max_prefix_table_size, 0, MAX_PREFIX_LOOKUP_SIZE)
    _check_size(
        "datatype", options.max_datatype_table_size, 0, MAX_DATATYPE_LOOKUP_SIZE
    )

parser_options(options, *, delimited)

Return the parser options of a stream, after checking them.

Args: options (jelly.RdfStreamOptions): the stream options delimited (bool): whether the stream is delimited

Returns: ParserOptions: types, lookup sizes and other parameters of the stream

Raises: JellyConformanceError: if the parser can't read the stream.

Source code in pyjelly/parse/decode.py

def parser_options(
    options: jelly.RdfStreamOptions, *, delimited: bool
) -> ParserOptions:
    """
    Return the parser options of a stream, after checking them.

    Args:
        options (jelly.RdfStreamOptions): the stream options
        delimited (bool): whether the stream is delimited

    Returns:
        ParserOptions: types, lookup sizes and other parameters of the stream

    Raises:
        JellyConformanceError: if the parser can't read the stream.

    """
    check_stream_options(options)
    return ParserOptions(
        stream_types=StreamTypes(
            physical_type=options.physical_type,
            logical_type=_logical_type(options),
        ),
        lookup_preset=LookupPreset(
            max_names=options.max_name_table_size,
            max_prefixes=options.max_prefix_table_size,
            max_datatypes=options.max_datatype_table_size,
        ),
        params=StreamParameters(
            stream_name=options.stream_name,
            generalized_statements=options.generalized_statements,
            rdf_star=options.rdf_star,
            version=options.version,
            delimited=delimited,
            namespace_declarations=options.version >= VERSION_1_1,
            stream_type=options.stream_type,
            rdf_version=options.rdf_version,
        ),
    )

options_from_frame(frame, *, delimited)

Fill stream options based on the options row.

Args: frame (jelly.RdfStreamFrame): first non-empty frame from the stream delimited (bool): derived delimited flag

Returns: ParserOptions: filled options with types/lookups/stream parameters information

Raises: JellyConformanceError: if the parser can't read the stream.

Source code in pyjelly/parse/decode.py

def options_from_frame(
    frame: jelly.RdfStreamFrame,
    *,
    delimited: bool,
) -> ParserOptions:
    """
    Fill stream options based on the options row.

    Args:
        frame (jelly.RdfStreamFrame): first non-empty frame from the stream
        delimited (bool): derived delimited flag

    Returns:
        ParserOptions: filled options with types/lookups/stream parameters information

    Raises:
        JellyConformanceError: if the parser can't read the stream.

    """
    row = frame.rows[0]
    if _ROW_WHICH_ONEOF(row, "row") != "options":
        msg = "the stream does not start with the stream options"
        raise JellyConformanceError(msg)
    return parser_options(row.options, delimited=delimited)

messages(rows)

Group the rows of a stream of RDF Messages by message.

Args: rows (Iterable[Iterable[Any]]): the rows that Decoder.iter_rows() returns for each frame, and Decoder.end_of_stream()

Yields: list[Any]: the namespace declarations and statements of each message

Source code in pyjelly/parse/decode.py

def messages(rows: Iterable[Iterable[Any]]) -> Iterator[list[Any]]:
    """
    Group the rows of a stream of RDF Messages by message.

    Args:
        rows (Iterable[Iterable[Any]]): the rows that Decoder.iter_rows() returns
            for each frame, and Decoder.end_of_stream()

    Yields:
        list[Any]: the namespace declarations and statements of each message

    """
    message: list[Any] = []
    for frame_rows in rows:
        for row in frame_rows:
            if row is MESSAGE_END:
                yield message
                message = []
            else:
                message.append(row)

grouped_rows(rows, options)

Return the rows of each frame, or of each message in a stream of RDF Messages.

Args: rows (Iterable[Iterable[Any]]): the rows that Decoder.iter_rows() returns for each frame, and Decoder.end_of_stream() options (ParserOptions): the stream's options

Source code in pyjelly/parse/decode.py

def grouped_rows(
    rows: Iterable[Iterable[Any]], options: ParserOptions
) -> Iterable[Iterable[Any]]:
    """
    Return the rows of each frame, or of each message in a stream of RDF Messages.

    Args:
        rows (Iterable[Iterable[Any]]): the rows that Decoder.iter_rows() returns
            for each frame, and Decoder.end_of_stream()
        options (ParserOptions): the stream's options

    """
    if options.params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES:
        return messages(rows)
    return rows

flat_rows(rows, options)

Return the rows of all frames, without the ends of RDF Messages.

Args: rows (Iterable[Iterable[Any]]): the rows that Decoder.iter_rows() returns for each frame, and Decoder.end_of_stream() options (ParserOptions): the stream's options

Source code in pyjelly/parse/decode.py

def flat_rows(rows: Iterable[Iterable[Any]], options: ParserOptions) -> Iterator[Any]:
    """
    Return the rows of all frames, without the ends of RDF Messages.

    Args:
        rows (Iterable[Iterable[Any]]): the rows that Decoder.iter_rows() returns
            for each frame, and Decoder.end_of_stream()
        options (ParserOptions): the stream's options

    """
    # Not a generator: the rows of the frames reach the caller without passing
    # through one more Python frame
    all_rows = chain.from_iterable(rows)
    if options.params.stream_type == jelly.RDF_STREAM_TYPE_MESSAGES:
        return filter(partial(is_not, MESSAGE_END), all_rows)
    return all_rows

ioutils

Functions:

Name	Description
single_byte_tags	Return the single-byte tags that a message's fields can have.
hint_bytes_needed	Return how many bytes delimited_jelly_hint needs to decide.
delimited_jelly_hint	Detect whether a Jelly stream is delimited, from its first bytes.
read_hint_header	Read just enough of the start of the stream to detect delimiting.
detect_delimited	Detect from the first bytes whether a stream is delimited.
parse_delimited	Read the next length-prefixed message, or None at the end of the input.
get_options_and_frames	Return stream options and frames from the buffered binary stream.

Attributes:

Name	Type	Description
FRAME_TAGS	Final	Tag bytes a non-delimited frame can start with (currently 0x0A, 0x12, 0x7A).
SPARQL_FRAME_TAGS	Final	Tag bytes that a non-delimited Jelly-SPARQL frame can start with.

FRAME_TAGS = single_byte_tags(cast('Descriptor', jelly.RdfStreamFrame.DESCRIPTOR))

Tag bytes a non-delimited frame can start with (currently 0x0A, 0x12, 0x7A).

SPARQL_FRAME_TAGS = single_byte_tags(cast('Descriptor', jelly.SparqlResultsFrame.DESCRIPTOR))

Tag bytes that a non-delimited Jelly-SPARQL frame can start with.

single_byte_tags(descriptor)

Return the single-byte tags that a message's fields can have.

Only fields numbered 1 to 15 have a one-byte tag. A repeated scalar field may be written packed or not, so both of its tags are returned.

Args: descriptor (Descriptor): descriptor of the message

Returns: frozenset[int]: tag bytes of the message's fields

Source code in pyjelly/parse/ioutils.py

def single_byte_tags(descriptor: Descriptor) -> frozenset[int]:
    """
    Return the single-byte tags that a message's fields can have.

    Only fields numbered 1 to 15 have a one-byte tag. A repeated scalar field
    may be written packed or not, so both of its tags are returned.

    Args:
        descriptor (Descriptor): descriptor of the message

    Returns:
        frozenset[int]: tag bytes of the message's fields

    """
    tags = set()
    for field in descriptor.fields:
        if field.number > _MAX_SINGLE_BYTE_FIELD:
            continue
        tags.add((field.number << 3) | _WIRE_TYPES.get(field.type, _WIRE_VARINT))
        if field.is_repeated:
            tags.add((field.number << 3) | _WIRE_LENGTH_DELIMITED)
    return frozenset(tags)

hint_bytes_needed(first_byte, tags=FRAME_TAGS)

Return how many bytes delimited_jelly_hint needs to decide.

Args: first_byte (int): first byte of the stream tags (frozenset[int]): tag bytes a frame can start with

Returns: int: number of bytes to read from the start of the stream, at most 128

hint_bytes_needed(0x11) 1

hint_bytes_needed(0x0A) 11

Source code in pyjelly/parse/ioutils.py

def hint_bytes_needed(first_byte: int, tags: frozenset[int] = FRAME_TAGS) -> int:
    """
    Return how many bytes `delimited_jelly_hint` needs to decide.

    Args:
        first_byte (int): first byte of the stream
        tags (frozenset[int]): tag bytes a frame can start with

    Returns:
        int: number of bytes to read from the start of the stream, at most 128

    >>> hint_bytes_needed(0x11)
    1

    >>> hint_bytes_needed(0x0A)
    11

    """
    if first_byte in tags:
        return first_byte + 1
    return 1

delimited_jelly_hint(header, tags=FRAME_TAGS)

Detect whether a Jelly stream is delimited, from its first bytes.

A non-delimited stream is a single frame, so it starts with the tag of one of the frame's fields -- any of them, because a writer may put the fields in any order. A delimited stream starts with the size of its first frame. So, if the first byte is not a frame tag, the stream is delimited. Otherwise it may still be the size of a small first frame, of at most 127 bytes: the stream is then delimited if the bytes after it are exactly that many bytes of frame fields.

Pass as much of the start of the stream as you have, up to 128 bytes -- hint_bytes_needed tells you how much is actually needed once you have the first byte. Given fewer bytes than the check needs, the stream is reported as non-delimited.

Args: header (bytes): first bytes of the stream tags (frozenset[int]): tag bytes a frame can start with, FRAME_TAGS for Jelly-RDF and SPARQL_FRAME_TAGS for Jelly-SPARQL

Returns: bool: True if the stream is delimited

An empty stream is an empty non-delimited frame:

delimited_jelly_hint(b"") False

0x11 is not a frame tag, so it is the size of the first frame:

delimited_jelly_hint(bytes([0x11, 0x0A, 0x0F, 0x0A, 0x0D])) True

A 10-byte frame holding one name entry, written non-delimited. It starts with the tag of the rows field, and the 9 bytes after it are not a whole frame of 10 bytes:

frame = bytes([0x0A, 0x08, 0x4A, 0x06, 0x12, 0x04, 0x6E, 0x61, 0x6D, 0x65]) delimited_jelly_hint(frame) False

The same frame written delimited. Its size, 10, is also the tag of the rows field, but the 10 bytes after it are exactly one frame's worth of fields:

delimited_jelly_hint(bytes([0x0A]) + frame) True

A stream of empty delimited frames is delimited too -- 0 is never a tag:

delimited_jelly_hint(bytes([0x00, 0x00, 0x00])) True

Two bytes of garbage are not enough to tell, so they count as a frame:

delimited_jelly_hint(bytes([0x12, 0x34])) False

Source code in pyjelly/parse/ioutils.py

def delimited_jelly_hint(header: bytes, tags: frozenset[int] = FRAME_TAGS) -> bool:
    """
    Detect whether a Jelly stream is delimited, from its first bytes.

    A non-delimited stream is a single frame, so it starts with the tag of one
    of the frame's fields -- any of them, because a writer may put the fields
    in any order. A delimited stream starts with the size of its first frame.
    So, if the first byte is not a frame tag, the stream is delimited.
    Otherwise it may still be the size of a small first frame, of at most 127
    bytes: the stream is then delimited if the bytes after it are exactly that
    many bytes of frame fields.

    Pass as much of the start of the stream as you have, up to 128 bytes --
    `hint_bytes_needed` tells you how much is actually needed once you have the
    first byte. Given fewer bytes than the check needs, the stream is reported
    as non-delimited.

    Args:
        header (bytes): first bytes of the stream
        tags (frozenset[int]): tag bytes a frame can start with, FRAME_TAGS for
            Jelly-RDF and SPARQL_FRAME_TAGS for Jelly-SPARQL

    Returns:
        bool: True if the stream is delimited

    An empty stream is an empty non-delimited frame:

    >>> delimited_jelly_hint(b"")
    False

    0x11 is not a frame tag, so it is the size of the first frame:

    >>> delimited_jelly_hint(bytes([0x11, 0x0A, 0x0F, 0x0A, 0x0D]))
    True

    A 10-byte frame holding one name entry, written non-delimited. It starts
    with the tag of the rows field, and the 9 bytes after it are not a whole
    frame of 10 bytes:

    >>> frame = bytes([0x0A, 0x08, 0x4A, 0x06, 0x12, 0x04, 0x6E, 0x61, 0x6D, 0x65])
    >>> delimited_jelly_hint(frame)
    False

    The same frame written delimited. Its size, 10, is also the tag of the rows
    field, but the 10 bytes after it are exactly one frame's worth of fields:

    >>> delimited_jelly_hint(bytes([0x0A]) + frame)
    True

    A stream of empty delimited frames is delimited too -- 0 is never a tag:

    >>> delimited_jelly_hint(bytes([0x00, 0x00, 0x00]))
    True

    Two bytes of garbage are not enough to tell, so they count as a frame:

    >>> delimited_jelly_hint(bytes([0x12, 0x34]))
    False

    """
    if not header:
        return False
    first = header[0]
    if first not in tags:
        # Not a frame tag, so it is the size of the first frame. This includes
        # 0, an empty frame: a stream of only empty frames is delimited too.
        return True
    if len(header) < first + 1:
        return False
    return _is_frame_fields(header[1 : first + 1], tags)

read_hint_header(inp, tags=FRAME_TAGS)

Read just enough of the start of the stream to detect delimiting.

Reads one byte, and at most 128. Pass the result to delimited_jelly_hint.

Args: inp (IO[bytes]): stream to read from tags (frozenset[int]): tag bytes a frame can start with

Returns: bytes: bytes read from the start of the stream

Source code in pyjelly/parse/ioutils.py

def read_hint_header(inp: IO[bytes], tags: frozenset[int] = FRAME_TAGS) -> bytes:
    """
    Read just enough of the start of the stream to detect delimiting.

    Reads one byte, and at most 128. Pass the result to `delimited_jelly_hint`.

    Args:
        inp (IO[bytes]): stream to read from
        tags (frozenset[int]): tag bytes a frame can start with

    Returns:
        bytes: bytes read from the start of the stream

    """
    header = inp.read(1)
    if not header:
        return header
    remaining = hint_bytes_needed(header[0], tags) - 1
    if remaining:
        header += _read_exactly(inp, remaining)
    return header

detect_delimited(inp, tags=FRAME_TAGS)

Detect from the first bytes whether a stream is delimited.

Args: inp (IO[bytes]): the stream tags (frozenset[int]): tag bytes a frame can start with

Returns: tuple[IO[bytes], bool]: a stream positioned at the start again, and whether it is delimited

Source code in pyjelly/parse/ioutils.py

def detect_delimited(
    inp: IO[bytes], tags: frozenset[int] = FRAME_TAGS
) -> tuple[IO[bytes], bool]:
    """
    Detect from the first bytes whether a stream is delimited.

    Args:
        inp (IO[bytes]): the stream
        tags (frozenset[int]): tag bytes a frame can start with

    Returns:
        tuple[IO[bytes], bool]: a stream positioned at the start again, and
            whether it is delimited

    """
    if inp.seekable():
        header = read_hint_header(inp, tags)
        inp.seek(-len(header), os.SEEK_CUR)
    else:
        # Input may not be seekable (e.g. a network stream). Then the bytes we read
        # are put back in front of it. Only an unbuffered stream gets a buffer: a
        # second buffer around a buffered stream (e.g. from socket.makefile())
        # waits for the inner one to fill up, which blocks streaming.
        # See also: https://github.com/Jelly-RDF/pyjelly/issues/298
        buffered = inp
        if not isinstance(inp, io.BufferedIOBase):
            buffered = io.BufferedReader(inp)  # type: ignore[arg-type, type-var, unused-ignore]
        header = read_hint_header(buffered, tags)
        inp = _StreamWithHeader(header, buffered)  # type: ignore[assignment]
    return inp, delimited_jelly_hint(header, tags)

parse_delimited(message_class, inp)

Read the next length-prefixed message, or None at the end of the input.

Raises: DecodeError: if the bytes are not a message of the class, or the input ends in the middle of one. Or UnicodeDecodeError, see DECODE_ERRORS. ValueError: if the input is closed.

Source code in pyjelly/parse/ioutils.py

def parse_delimited(message_class: type[MessageT], inp: IO[bytes]) -> MessageT | None:
    """
    Read the next length-prefixed message, or None at the end of the input.

    Raises:
        DecodeError: if the bytes are not a message of the class, or the input
            ends in the middle of one. Or UnicodeDecodeError, see DECODE_ERRORS.
        ValueError: if the input is closed.

    """
    try:
        # protobuf annotates `input_bytes` as io.BytesIO, but any IO[bytes] works.
        return parse_length_prefixed(message_class, inp)  # type: ignore[arg-type]
    except DECODE_ERRORS:
        raise
    except (ValueError, OverflowError) as error:
        if getattr(inp, "closed", False):
            raise
        # The length prefix is cut short, or too large to read
        msg = f"invalid length prefix ({error})"
        raise DecodeError(msg) from None

get_options_and_frames(inp)

Return stream options and frames from the buffered binary stream.

The options are in the first frame that has rows. A stream with no such frame (e.g., an empty file) is an empty graph, with the options of an empty Jelly-RDF 1.2 stream of triples.

Args: inp (IO[bytes]): jelly buffered binary stream

Raises: JellyConformanceError: if the parser can't read the stream

Returns: tuple[ParserOptions, Iterator[jelly.RdfStreamFrame]]: ParserOptions holds: stream types, lookup presets and other stream options

Source code in pyjelly/parse/ioutils.py

def get_options_and_frames(
    inp: IO[bytes],
) -> tuple[ParserOptions, Iterator[jelly.RdfStreamFrame]]:
    """
    Return stream options and frames from the buffered binary stream.

    The options are in the first frame that has rows. A stream with no such frame
    (e.g., an empty file) is an empty graph, with the options of an empty
    Jelly-RDF 1.2 stream of triples.

    Args:
        inp (IO[bytes]): jelly buffered binary stream

    Raises:
        JellyConformanceError: if the parser can't read the stream

    Returns:
        tuple[ParserOptions, Iterator[jelly.RdfStreamFrame]]: ParserOptions holds:
            stream types, lookup presets and other stream options

    """
    inp, is_delimited = detect_delimited(inp)

    if is_delimited:
        first_frame = None
        skipped_frames = []
        frames = frame_iterator(inp)
        for frame in frames:
            if not frame.rows:
                skipped_frames.append(frame)
            else:
                first_frame = frame
                break
        if first_frame is None:
            return _empty_stream_options(delimited=True), iter(skipped_frames)

        options = options_from_frame(first_frame, delimited=True)
        return options, chain(skipped_frames, (first_frame,), frames)

    try:
        frame = parse(jelly.RdfStreamFrame, inp.read())
    except DECODE_ERRORS as error:
        raise _not_jelly(error) from None

    if not frame.rows:
        return _empty_stream_options(delimited=False), iter((frame,))

    options = options_from_frame(frame, delimited=False)
    return options, iter((frame,))

lookup

Classes:

Name	Description
LookupDecoder	Shared base for RDF lookup encoders using Jelly compression.

LookupDecoder(*, lookup_size)

Shared base for RDF lookup encoders using Jelly compression.

Tracks the last assigned and last reused index.

Parameters:

Name	Type	Description	Default
lookup_size	int	Maximum lookup size.	required

Source code in pyjelly/parse/lookup.py

def __init__(self, *, lookup_size: int) -> None:
    self.lookup_size = lookup_size
    # A list, not a deque: indexing a deque walks through its blocks
    self.data: list[str | None] = [None] * lookup_size
    self.last_assigned_index = 0
    self.last_reused_index = 0

sparql

Jelly-SPARQL reader: SPARQL query results in frames, with one column per variable.

SparqlDecoder decodes frames one at a time. read_result_sets() groups the rows of the frames into result sets.

Classes:

Name	Description
DecodedFrame	Contents of one decoded Jelly-SPARQL frame.
SparqlDecoder	Decoder of the frames of a Jelly-SPARQL stream, in stream order.
ResultSetReader	Reads and decodes frames one at a time, for the result sets.
SparqlResultSet	One result set of a Jelly-SPARQL stream: a solution sequence or a boolean result.

Functions:

Name	Description
tuple_rows	Build the rows of a frame as tuples, with one value per variable.
check_sparql_options	Check that this parser supports a Jelly-SPARQL stream with these options.
sparql_parser_options	Build the adapter options for a Jelly-SPARQL stream.
decode_links	Split the "link" metadata value into IRIs. Return None if it isn't UTF-8.
sparql_frames	Iterate the frames of a Jelly-SPARQL stream, delimited or not.
read_result_sets	Read the result sets of a Jelly-SPARQL stream.
read_result_set	Read the single result set of a FLAT Jelly-SPARQL stream.

DecodedFrame()

Contents of one decoded Jelly-SPARQL frame.

Source code in pyjelly/parse/sparql.py

def __init__(self) -> None:
    #: Whether this is the first frame of a result set
    self.starts = False
    #: Variables of the result set, if this frame starts a solution sequence
    self.variables: list[str] = []
    #: Boolean result, if the frame has one
    self.boolean: bool | None = None
    #: Rows of the frame, built by the decoder's row factory
    self.rows: list[Any] = []
    #: Link IRIs for the result set, if the frame has them
    self.links: list[str] | None = None
    #: Trailer: None if the frame has none, "" if the result set is complete,
    #: otherwise the error message
    self.trailer: str | None = None

SparqlDecoder(adapter_factory, *, row_factory=tuple_rows, max_rows=MAX_SPARQL_FRAME_ROWS, max_values=MAX_SPARQL_FRAME_VALUES)

Decoder of the frames of a Jelly-SPARQL stream, in stream order.

Args: adapter_factory (AdapterFactory): builds the term adapter from the stream options. row_factory (RowFactory): builds the rows of a frame. Tuples by default. max_rows (int): maximum rows per frame. max_values (int): maximum values (rows times variables) per frame. A frame is decoded all at once, so this caps its memory use.

Methods:

Name	Description
decode	Decode a frame.
read_options	Read the stream options. Repeated options also empty the lookups.
start_result_set	Reset the per-result-set state, for the next result set.
read_variables	Read the frame's header, if any, and return the current variables.

Source code in pyjelly/parse/sparql.py

def __init__(
    self,
    adapter_factory: AdapterFactory,
    *,
    row_factory: RowFactory = tuple_rows,
    max_rows: int = MAX_SPARQL_FRAME_ROWS,
    max_values: int = MAX_SPARQL_FRAME_VALUES,
) -> None:
    self.adapter_factory = adapter_factory
    self.row_factory = row_factory
    self.max_rows = min(max_rows, MAX_ROWS)
    self.max_values = max_values
    #: Stream options, once read
    self.options: jelly.SparqlResultsOptions | None = None
    self.decoder: Decoder | None = None
    self.columns: ColumnDecoder | None = None
    #: Whether the stream is a sequence of result sets
    self.punctuated = False
    # Variables declared by the first header of the result set, and whether a
    # header is in effect. Repeated stream options cancel the header.
    self.variables: list[str] | None = None
    self.header_in_effect = False
    # Whether the next frame starts a result set, whether the current result
    # set is a boolean result, and whether a trailer came after the options
    self.result_set_start = True
    self.boolean_read = False
    self.trailer_read = False
    # Number of result sets started so far
    self.result_sets = 0

decode(frame)

Decode a frame.

Source code in pyjelly/parse/sparql.py

def decode(self, frame: jelly.SparqlResultsFrame) -> DecodedFrame:
    """Decode a frame."""
    out = DecodedFrame()
    has_options = _FRAME_HAS_FIELD(frame, "options")
    if has_options:
        self.read_options(frame.options)
    elif self.options is None:
        msg = "the stream does not start with the stream options"
        raise JellyConformanceError(msg)
    elif self.trailer_read and not self.punctuated:
        msg = (
            "a frame after the trailer must have the stream options: the trailer "
            "ends the result set"
        )
        raise JellyConformanceError(msg)
    if self.result_set_start:
        self.start_result_set()
        out.starts = True
    if self.boolean_read:
        msg = "no frame of the result set can follow its boolean result"
        raise JellyConformanceError(msg)
    if _FRAME_HAS_FIELD(frame, "ask_result"):
        self.read_boolean(frame, out)
    else:
        self.read_solutions(frame, has_options=has_options, out=out)
    links = frame.metadata.get(LINK_KEY)
    if links is not None:
        out.links = decode_links(links)
    if _FRAME_HAS_FIELD(frame, "trailer"):
        out.trailer = frame.trailer.error
        self.trailer_read = True
        self.result_set_start = self.punctuated
    else:
        self.result_set_start = False
    return out

read_options(options)

Read the stream options. Repeated options also empty the lookups.

Source code in pyjelly/parse/sparql.py

def read_options(self, options: jelly.SparqlResultsOptions) -> None:
    """Read the stream options. Repeated options also empty the lookups."""
    check_sparql_options(options)
    punctuated = options.stream_type == jelly.SPARQL_STREAM_TYPE_PUNCTUATED
    parser_options = sparql_parser_options(options)
    current = self.options
    if current is None:
        decoder = Decoder(self.adapter_factory(parser_options))
        self.decoder = decoder
        self.columns = decoder.column_decoder
    else:
        if options.stream_type != current.stream_type:
            msg = "the stream type must be the same in all stream options"
            raise JellyConformanceError(msg)
        if punctuated and not self.result_set_start:
            msg = (
                "in a PUNCTUATED stream, the stream options can only be repeated "
                "in the first frame of a result set"
            )
            raise JellyConformanceError(msg)
        current_decoder = self.decoder
        assert current_decoder is not None
        current_decoder.adapter.options = parser_options
        current_decoder.new_lookups()
        self.header_in_effect = False
    self.options = options
    self.punctuated = punctuated
    if not punctuated:
        self.trailer_read = False

start_result_set()

Reset the per-result-set state, for the next result set.

Source code in pyjelly/parse/sparql.py

def start_result_set(self) -> None:
    """Reset the per-result-set state, for the next result set."""
    self.variables = None
    self.header_in_effect = False
    self.boolean_read = False
    self.trailer_read = False
    if self.result_sets:
        # Each result set has its own blank node labels
        columns = self.columns
        assert columns is not None
        columns.bnode_scope = f"r{self.result_sets}-"
    self.result_sets += 1

read_variables(frame, *, has_options)

Read the frame's header, if any, and return the current variables.

Source code in pyjelly/parse/sparql.py

def read_variables(
    self, frame: jelly.SparqlResultsFrame, *, has_options: bool
) -> list[str]:
    """Read the frame's header, if any, and return the current variables."""
    if frame.variables:
        if not has_options and not self.result_set_start:
            msg = (
                "the variables can only be declared in the first frame of a "
                "result set, or in a frame with the stream options"
            )
            raise JellyConformanceError(msg)
        self.read_header(list(frame.variables))
    elif not self.header_in_effect:
        # A missing header next to the options, or at the start of a result
        # set, means zero variables
        if self.variables:
            msg = (
                "the frame repeats the stream options, so it must repeat the "
                "variables too"
            )
            raise JellyConformanceError(msg)
        self.read_header([])
    variables = self.variables
    assert variables is not None
    return variables

ResultSetReader(frames, decoder)

Reads and decodes frames one at a time, for the result sets.

Methods:

Name	Description
next	Return the next frame, or None at the end of the stream.

Source code in pyjelly/parse/sparql.py

def __init__(
    self, frames: Iterator[jelly.SparqlResultsFrame], decoder: SparqlDecoder
) -> None:
    self.frames = frames
    self.decoder = decoder
    # A frame read ahead that belongs to the next result set
    self.pending: DecodedFrame | None = None

next()

Return the next frame, or None at the end of the stream.

Source code in pyjelly/parse/sparql.py

def next(self) -> DecodedFrame | None:
    """Return the next frame, or None at the end of the stream."""
    frame = self.pending
    if frame is not None:
        self.pending = None
        return frame
    message = next(self.frames, None)
    if message is None:
        return None
    return self.decoder.decode(message)

SparqlResultSet(reader, first)

One result set of a Jelly-SPARQL stream: a solution sequence or a boolean result.

Rows are read from the stream lazily, during iteration. Each row is a tuple with one value per variable, None if the variable is unbound. A loop that stops early can be resumed later. Once all rows have been read, iterating again raises an error. If the next result set is read first, the unread rows are kept in memory.

If the producer ended the result set with an error, iterating raises JellyIncompleteResultsError after the last row, and so does boolean. If the stream ends without a trailer, truncated is set and a JellyTruncatedResultsWarning is issued.

Methods:

Name	Description
collect	Append all remaining rows to a list. Faster than iterating.
keep	Read the remaining rows into memory, for iteration later.

Attributes:

Name	Type	Description
boolean	bool | None	Boolean result of an ASK query, or None for a solution sequence.
is_boolean	bool	Whether this is a boolean result (of an ASK query).

Source code in pyjelly/parse/sparql.py

def __init__(self, reader: ResultSetReader, first: DecodedFrame) -> None:
    self.reader = reader
    self.first: DecodedFrame | None = first
    #: Variables in projection order. Empty for a boolean result.
    self.variables: list[str] = first.variables
    self._boolean = first.boolean
    #: Link IRIs for the result set (head.link in SPARQL JSON)
    self.links: list[str] = first.links or []
    #: Error message from the producer, if the result set is incomplete
    self.error = ""
    #: Whether the stream ended without a trailer, so rows may be missing.
    #: Only known after all rows have been read.
    self.truncated = False
    # Row lists of the coming frames, and the unread rows of the current one
    self._frames = self._read()
    self._rows: Iterator[Any] = iter(())
    # Whether an iteration has read all rows
    self._done = False
    if first.boolean is not None:
        # A boolean result is a single frame, so its trailer is already known
        if first.trailer:
            self.error = first.trailer
        elif first.trailer is None:
            self._truncated()

boolean

Boolean result of an ASK query, or None for a solution sequence.

Raises: JellyIncompleteResultsError: if the producer reported an error.

is_boolean

Whether this is a boolean result (of an ASK query).

collect(rows)

Append all remaining rows to a list. Faster than iterating.

Raises: JellyIncompleteResultsError: if the result set is incomplete. The rows are appended first.

Source code in pyjelly/parse/sparql.py

def collect(self, rows: list[Any]) -> None:
    """
    Append all remaining rows to a list. Faster than iterating.

    Raises:
        JellyIncompleteResultsError: if the result set is incomplete. The rows
            are appended first.

    """
    self._check_not_done()
    rows.extend(self._rows)
    for frame_rows in self._frames:
        rows.extend(frame_rows)
    self._end()

keep()

Read the remaining rows into memory, for iteration later.

Source code in pyjelly/parse/sparql.py

def keep(self) -> None:
    """Read the remaining rows into memory, for iteration later."""
    if not self._done:
        kept = list(self._rows)
        for rows in self._frames:
            kept += rows
        self._rows = iter(kept)

tuple_rows(variables, cells, rows)

Build the rows of a frame as tuples, with one value per variable.

Source code in pyjelly/parse/sparql.py

def tuple_rows(variables: list[str], cells: list[list[Any]], rows: int) -> list[Any]:
    """Build the rows of a frame as tuples, with one value per variable."""
    if not variables:
        return [()] * rows
    return list(zip(*cells, strict=True))

check_sparql_options(options)

Check that this parser supports a Jelly-SPARQL stream with these options.

Raises: JellyConformanceError: if it can't.

Source code in pyjelly/parse/sparql.py

def check_sparql_options(options: jelly.SparqlResultsOptions) -> None:
    """
    Check that this parser supports a Jelly-SPARQL stream with these options.

    Raises:
        JellyConformanceError: if it can't.

    """
    version = options.version
    if not 1 <= version <= SPARQL_MAX_VERSION:
        msg = (
            f"unsupported Jelly-SPARQL version {version}, "
            f"expected 1 to {SPARQL_MAX_VERSION}"
        )
        raise JellyConformanceError(msg)
    if options.stream_type not in _STREAM_TYPES:
        msg = f"unknown stream type: {options.stream_type}"
        raise JellyConformanceError(msg)
    if options.rdf_version not in _RDF_VERSIONS:
        msg = f"unknown RDF version: {options.rdf_version}"
        raise JellyConformanceError(msg)
    _check_size(
        "name",
        options.max_name_table_size,
        MIN_COLUMN_NAME_LOOKUP_SIZE,
        MAX_NAME_LOOKUP_SIZE,
    )
    _check_size("prefix", options.max_prefix_table_size, 0, MAX_PREFIX_LOOKUP_SIZE)
    _check_size(
        "datatype", options.max_datatype_table_size, 0, MAX_DATATYPE_LOOKUP_SIZE
    )

sparql_parser_options(options)

Build the adapter options for a Jelly-SPARQL stream.

Terms are decoded like in a Jelly-RDF 1.2 stream with the same lookup sizes and RDF version. The stream types are left unspecified.

Source code in pyjelly/parse/sparql.py

def sparql_parser_options(options: jelly.SparqlResultsOptions) -> ParserOptions:
    """
    Build the adapter options for a Jelly-SPARQL stream.

    Terms are decoded like in a Jelly-RDF 1.2 stream with the same lookup sizes and
    RDF version. The stream types are left unspecified.
    """
    return ParserOptions(
        stream_types=StreamTypes(),
        lookup_preset=LookupPreset(
            max_names=options.max_name_table_size,
            max_prefixes=options.max_prefix_table_size,
            max_datatypes=options.max_datatype_table_size,
        ),
        params=StreamParameters(
            version=VERSION_1_2,
            stream_name=options.stream_name,
            rdf_version=options.rdf_version,
        ),
    )

decode_links(value)

Split the "link" metadata value into IRIs. Return None if it isn't UTF-8.

decode_links(b"http://example.org/a\nhttp://example.org/b") ['http://example.org/a', 'http://example.org/b'] decode_links(b"") [] decode_links(b"\xff") is None True

Source code in pyjelly/parse/sparql.py

def decode_links(value: bytes) -> list[str] | None:
    r"""
    Split the "link" metadata value into IRIs. Return None if it isn't UTF-8.

    >>> decode_links(b"http://example.org/a\nhttp://example.org/b")
    ['http://example.org/a', 'http://example.org/b']
    >>> decode_links(b"")
    []
    >>> decode_links(b"\xff") is None
    True
    """
    try:
        text = value.decode()
    except UnicodeDecodeError:
        return None
    return text.split("\n") if text else []

sparql_frames(inp)

Iterate the frames of a Jelly-SPARQL stream, delimited or not.

Raises: JellyConformanceError: if the stream is empty, or isn't Jelly-SPARQL.

Source code in pyjelly/parse/sparql.py

def sparql_frames(inp: IO[bytes] | bytes) -> Iterator[jelly.SparqlResultsFrame]:
    """
    Iterate the frames of a Jelly-SPARQL stream, delimited or not.

    Raises:
        JellyConformanceError: if the stream is empty, or isn't Jelly-SPARQL.

    """
    if isinstance(inp, (bytes, bytearray, memoryview)):
        inp = io.BytesIO(inp)
    inp, delimited = detect_delimited(inp, SPARQL_FRAME_TAGS)
    if not delimited:
        data = inp.read()
        if not data:
            msg = "the stream is empty, but it must have at least one frame"
            raise JellyConformanceError(msg)
        try:
            return iter((parse(jelly.SparqlResultsFrame, data),))
        except DECODE_ERRORS as error:
            raise _not_sparql(error) from None
    return _delimited_frames(inp)

read_result_sets(inp, adapter_factory, *, row_factory=tuple_rows, max_rows=MAX_SPARQL_FRAME_ROWS, max_values=MAX_SPARQL_FRAME_VALUES)

Read the result sets of a Jelly-SPARQL stream.

A FLAT stream (the default) has one result set, a PUNCTUATED stream any number. Moving on to the next result set reads the rest of the current one and keeps its unread rows in memory.

Args: inp (IO[bytes] | bytes): the stream, delimited or not. adapter_factory (AdapterFactory): builds the term adapter from the stream options. row_factory (RowFactory): builds the rows. Tuples by default. max_rows (int): maximum rows per frame. max_values (int): maximum values (rows times variables) per frame.

Yields: SparqlResultSet: the result sets, in order.

Source code in pyjelly/parse/sparql.py

def read_result_sets(
    inp: IO[bytes] | bytes,
    adapter_factory: AdapterFactory,
    *,
    row_factory: RowFactory = tuple_rows,
    max_rows: int = MAX_SPARQL_FRAME_ROWS,
    max_values: int = MAX_SPARQL_FRAME_VALUES,
) -> Iterator[SparqlResultSet]:
    """
    Read the result sets of a Jelly-SPARQL stream.

    A FLAT stream (the default) has one result set, a PUNCTUATED stream any
    number. Moving on to the next result set reads the rest of the current one
    and keeps its unread rows in memory.

    Args:
        inp (IO[bytes] | bytes): the stream, delimited or not.
        adapter_factory (AdapterFactory): builds the term adapter from the stream
            options.
        row_factory (RowFactory): builds the rows. Tuples by default.
        max_rows (int): maximum rows per frame.
        max_values (int): maximum values (rows times variables) per frame.

    Yields:
        SparqlResultSet: the result sets, in order.

    """
    decoder = SparqlDecoder(
        adapter_factory,
        row_factory=row_factory,
        max_rows=max_rows,
        max_values=max_values,
    )
    reader = ResultSetReader(sparql_frames(inp), decoder)
    while (first := reader.next()) is not None:
        result_set = SparqlResultSet(reader, first)
        yield result_set
        result_set.keep()

read_result_set(inp, adapter_factory, *, row_factory=tuple_rows, max_rows=MAX_SPARQL_FRAME_ROWS, max_values=MAX_SPARQL_FRAME_VALUES)

Read the single result set of a FLAT Jelly-SPARQL stream.

The first frame is read now, the rows during iteration. For PUNCTUATED streams, use read_result_sets().

Raises: JellyConformanceError: if the stream is PUNCTUATED. JellyIncompleteResultsError: for a boolean result that the producer could not complete.

Source code in pyjelly/parse/sparql.py

def read_result_set(
    inp: IO[bytes] | bytes,
    adapter_factory: AdapterFactory,
    *,
    row_factory: RowFactory = tuple_rows,
    max_rows: int = MAX_SPARQL_FRAME_ROWS,
    max_values: int = MAX_SPARQL_FRAME_VALUES,
) -> SparqlResultSet:
    """
    Read the single result set of a FLAT Jelly-SPARQL stream.

    The first frame is read now, the rows during iteration. For PUNCTUATED
    streams, use read_result_sets().

    Raises:
        JellyConformanceError: if the stream is PUNCTUATED.
        JellyIncompleteResultsError: for a boolean result that the producer
            could not complete.

    """
    result_sets = read_result_sets(
        inp,
        adapter_factory,
        row_factory=row_factory,
        max_rows=max_rows,
        max_values=max_values,
    )
    result_set = next(result_sets)
    if result_set.reader.decoder.punctuated:
        msg = (
            "the stream is a sequence of result sets (PUNCTUATED). Read it with "
            "parse_jelly_sparql_sequence() of pyjelly.integrations.rdflib.sparql "
            "or pyjelly.integrations.generic.sparql"
        )
        raise JellyConformanceError(msg)
    if result_set.is_boolean and result_set.error:
        raise _incomplete(result_set.error, result_set)
    return result_set

serialize

Modules:

Name	Description
columns	The column layout of Jelly-RDF 1.2: a frame's statements, column by column.
encode
flows
ioutils
lookup
sparql	Jelly-SPARQL writer: SPARQL query results in frames.
streams

columns

The column layout of Jelly-RDF 1.2: a frame's statements, column by column.

Classes:

Name	Description
Column	The cells of one statement position in the frame being built.
ColumnBatch	The column batch of the frame being built.

Column(allowed, position)

The cells of one statement position in the frame being built.

Methods:

Name	Description
add_value	Add a run value: a term that is not the same as the one before.
end_run	End the current run, adding a layout token if it isn't one cell long.
write	Write the column into its message, and empty it for the next frame.

Source code in pyjelly/serialize/columns.py

def __init__(self, allowed: int, position: str) -> None:
    # Which types of terms the column allows (bits 1 << KIND_*), and its name
    # in error messages
    self.allowed = allowed
    self.position = position
    # The current run: its term, and how many cells it has. A run of the
    # default graph is a run of unbound cells
    self.run_term: object = None
    self.run_length = 0
    self.run_unbound = False
    # Run values that took one cell each since the last layout token
    self.skip = 0
    self.layouts: list[int] = []
    self.value_count = 0
    # The types of the run values (bits 1 << KIND_*), and once there is more
    # than one, the type of each value, packed as written: 2 bits each, four
    # in a byte, from the lowest bits
    self.kinds_used = 0
    self.kinds: list[int] | None = None
    # IRIs: name indices (0 if the previous one + 1) and prefix indices (0 if
    # the same as the previous one). If all prefix indices are the same, the
    # first is written alone.
    self.name_ids: list[int] = []
    self.prefix_ids: list[int] = []
    self.last_name_id = 0
    self.last_prefix_id = -1
    self.prefix_ids_differ = False
    # Literals: lexical forms and literal kinds, and the language tags they
    # refer to, with their base directions. If all literal kinds are the same,
    # the first is written alone.
    self.lex_values: list[str] = []
    self.literal_kinds: list[int] = []
    self.last_literal_kind = -1
    self.literal_kinds_differ = False
    self.langtags: list[str] = []
    self.langtag_directions: list[int] = []
    # Index of each language tag in langtags, one dict per base direction
    self.langtag_indices: list[dict[str, int]] = [{}, {}, {}]
    self.bnodes: list[str] = []
    self.triple_terms: list[jelly.RdfTripleTerm] = []

add_value(term)

Add a run value: a term that is not the same as the one before.

Source code in pyjelly/serialize/columns.py

def add_value(self, term: EncodedTerm) -> None:
    """Add a run value: a term that is not the same as the one before."""
    kind = _kind_of(term)
    bit = 1 << kind
    used = self.kinds_used
    if used != bit:
        if not used & bit:
            self._new_kind(kind, used)
        kinds = self.kinds
        if kinds is not None:
            shift = (self.value_count & 3) << 1
            if shift:
                kinds[-1] |= kind << shift
            else:
                kinds.append(kind)
    self.value_count += 1
    if kind == KIND_IRI:
        self._add_iri(term.first, term.second)
    elif kind == KIND_LITERAL:
        self._add_literal(term)
    elif kind == KIND_BNODE:
        self.bnodes.append(term.text)
    else:
        assert term.triple_term is not None
        self.triple_terms.append(term.triple_term)

end_run()

End the current run, adding a layout token if it isn't one cell long.

Source code in pyjelly/serialize/columns.py

def end_run(self) -> None:
    """End the current run, adding a layout token if it isn't one cell long."""
    length = self.run_length
    if not length:
        return
    if self.run_unbound:
        self._token(TOKEN_UNBOUND, length - 1)
        self.run_unbound = False
    elif length > 1:
        self._token(0, length - 2)
    else:
        self.skip += 1
    self.run_length = 0

write(column)

Write the column into its message, and empty it for the next frame.

Source code in pyjelly/serialize/columns.py

def write(self, column: jelly.RdfColumn) -> None:
    """Write the column into its message, and empty it for the next frame."""
    if self.run_unbound:
        # Unbound cells at the end are skipped: the parser pads the column
        self.run_length = 0
        self.run_unbound = False
    else:
        self.end_run()
    if self.layouts:
        column.layouts.extend(self.layouts)
    if self.name_ids:
        self._write_iris(column)
    if self.lex_values:
        self._write_literals(column)
    if self.bnodes:
        column.bnodes.extend(self.bnodes)
    if self.triple_terms:
        column.triple_terms.extend(self.triple_terms)
    if self.kinds is not None:
        column.kinds = bytes(self.kinds)
    self.clear()

ColumnBatch(encoder, *, quads, messages)

The column batch of the frame being built.

The terms are encoded by the stream's TermEncoder. The frame flow decides when the frame ends (see FrameFlow.to_stream_frame()).

Args: encoder (TermEncoder): encoder of the stream's terms. quads (bool): whether the statements are quads. messages (bool): whether the stream is a stream of RDF Messages.

Methods:

Name	Description
has_room	Return whether one more statement fits in the frame.
add	Add a subject, predicate or object to its column.
add_changed	Add a term of the cursor API, which tells if it changed.
add_graph	Add a graph name, an unbound cell for the default graph.
add_graph_changed	Add a graph name of the cursor API, which tells if it changed.
add_triple	Add a triple, and keep its terms in repeated_terms, as the row layout.
add_quad	Add a quad, and keep its terms in repeated_terms, as the row layout.
add_namespace	Add a namespace declaration, which comes before the frame's statements.
end_message	End the current RDF Message, which may have started in an earlier frame.
write	Write the frame's statements into its batch, and start the next frame.

Source code in pyjelly/serialize/columns.py

def __init__(self, encoder: TermEncoder, *, quads: bool, messages: bool) -> None:
    self.encoder = encoder
    self.messages = messages
    self.subjects = Column(_RESOURCES, "subject")
    self.predicates = Column(_IRIS, "predicate")
    self.objects = Column(_ANY, "object")
    self.graphs: Column | None = Column(_RESOURCES, "graph") if quads else None
    self.encoded = EncodedTerm()
    self.row_count = 0
    # The statements of the frame that are in messages that ended
    self.rows_in_messages = 0
    self.message_lengths: list[int] = []
    # Namespace declarations: name, and the indices of the IRI as written
    self.namespaces: list[tuple[str, int, int]] = []
    self.namespace_prefix_index = 0
    self.namespace_name_index = 0
    # The most IRIs a triple term had, so far in the stream: a statement can
    # need that many entries in the lookups for its object
    self.max_triple_term_iris = 1
    self.new_frame()

has_room()

Return whether one more statement fits in the frame.

Source code in pyjelly/serialize/columns.py

def has_room(self) -> bool:
    """Return whether one more statement fits in the frame."""
    return self.row_count < MAX_ROWS and self.encoder.has_room()

add(column, term)

Add a subject, predicate or object to its column.

Source code in pyjelly/serialize/columns.py

def add(self, column: Column, term: object) -> None:
    """Add a subject, predicate or object to its column."""
    if column.run_length and self.encoder.same_term(column.run_term, term):
        column.run_length += 1
        return
    column.end_run()
    encoded = self.encoded
    self.encoder.encode_spo(term, encoded)
    column.add_value(encoded)
    column.run_term = term
    column.run_length = 1
    if encoded.triple_term is not None:
        self._triple_term_added()

add_changed(column, term, *, changed)

Add a term of the cursor API, which tells if it changed.

Source code in pyjelly/serialize/columns.py

def add_changed(self, column: Column, term: object, *, changed: bool) -> None:
    """Add a term of the cursor API, which tells if it changed."""
    if not changed and column.run_length:
        column.run_length += 1
    else:
        self.add(column, term)

add_graph(term)

Add a graph name, an unbound cell for the default graph.

Source code in pyjelly/serialize/columns.py

def add_graph(self, term: object) -> None:
    """Add a graph name, an unbound cell for the default graph."""
    column = self.graphs
    assert column is not None
    if column.run_length and self.encoder.same_term(column.run_term, term):
        column.run_length += 1
        return
    encoded = self.encoded
    self.encoder.encode_graph(term, encoded)
    if encoded.kind == TERM_DEFAULT_GRAPH:
        if not column.run_unbound:
            column.end_run()
            column.run_unbound = True
    else:
        column.end_run()
        column.add_value(encoded)
    column.run_term = term
    column.run_length += 1

add_graph_changed(term, *, changed)

Add a graph name of the cursor API, which tells if it changed.

Source code in pyjelly/serialize/columns.py

def add_graph_changed(self, term: object, *, changed: bool) -> None:
    """Add a graph name of the cursor API, which tells if it changed."""
    column = self.graphs
    assert column is not None
    if not changed and column.run_length:
        column.run_length += 1
    else:
        self.add_graph(term)

add_triple(terms, repeated_terms)

Add a triple, and keep its terms in repeated_terms, as the row layout.

Source code in pyjelly/serialize/columns.py

def add_triple(
    self, terms: Iterable[object], repeated_terms: list[object | None]
) -> None:
    """Add a triple, and keep its terms in `repeated_terms`, as the row layout."""
    # Indexing a tuple is cheaper than unpacking any iterable
    if isinstance(terms, tuple) and len(terms) == _TRIPLE_SIZE:
        s = terms[0]
        p = terms[1]
        o = terms[2]
    else:
        s, p, o = terms
    self.add(self.subjects, s)
    self.add(self.predicates, p)
    self.add(self.objects, o)
    repeated_terms[0] = s
    repeated_terms[1] = p
    repeated_terms[2] = o
    self.row_count += 1

add_quad(terms, repeated_terms)

Add a quad, and keep its terms in repeated_terms, as the row layout.

Source code in pyjelly/serialize/columns.py

def add_quad(
    self, terms: Iterable[object], repeated_terms: list[object | None]
) -> None:
    """Add a quad, and keep its terms in `repeated_terms`, as the row layout."""
    # See add_triple()
    if isinstance(terms, tuple) and len(terms) == _QUAD_SIZE:
        s = terms[0]
        p = terms[1]
        o = terms[2]
        g = terms[3]
    else:
        s, p, o, g = terms
    self.add(self.subjects, s)
    self.add(self.predicates, p)
    self.add(self.objects, o)
    self.add_graph(g)
    repeated_terms[0] = s
    repeated_terms[1] = p
    repeated_terms[2] = o
    repeated_terms[3] = g
    self.row_count += 1

add_namespace(name, iri)

Add a namespace declaration, which comes before the frame's statements.

Source code in pyjelly/serialize/columns.py

def add_namespace(self, name: str, iri: str) -> None:
    """Add a namespace declaration, which comes before the frame's statements."""
    prefix_index, name_index = self.encoder.iri_indices(iri)
    # The IRIs of the declarations are a sequence of their own
    written_prefix = (
        0 if prefix_index == self.namespace_prefix_index else prefix_index
    )
    written_name = 0 if name_index == self.namespace_name_index + 1 else name_index
    self.namespace_prefix_index = prefix_index
    self.namespace_name_index = name_index
    self.namespaces.append((name, written_prefix, written_name))

end_message()

End the current RDF Message, which may have started in an earlier frame.

Source code in pyjelly/serialize/columns.py

def end_message(self) -> None:
    """End the current RDF Message, which may have started in an earlier frame."""
    self.message_lengths.append(self.row_count - self.rows_in_messages)
    self.rows_in_messages = self.row_count

write(batch)

Write the frame's statements into its batch, and start the next frame.

Source code in pyjelly/serialize/columns.py

def write(self, batch: jelly.RdfColumnBatch) -> None:
    """Write the frame's statements into its batch, and start the next frame."""
    encoder = self.encoder
    batch.row_count = self.row_count
    encoder.names.packed.write(batch.names)
    encoder.prefixes.packed.write(batch.prefixes)
    encoder.datatypes.packed.write(batch.datatypes)
    for name, prefix_index, name_index in self.namespaces:
        declaration = batch.namespaces.add()
        declaration.name = name
        declaration.value.prefix_id = prefix_index
        declaration.value.name_id = name_index
    if self.row_count:
        self.subjects.write(batch.subjects)
        self.predicates.write(batch.predicates)
        self.objects.write(batch.objects)
        graphs = self.graphs
        if graphs is not None:
            if graphs.value_count:
                graphs.write(batch.graphs)
            else:
                # All statements are in the default graph: no column at all
                graphs.clear()
    if self.message_lengths:
        batch.message_lengths.extend(self.message_lengths)
    self.row_count = 0
    self.rows_in_messages = 0
    self.message_lengths = []
    self.namespaces = []
    self.new_frame()

encode

Classes:

Name	Description
EncodedTerm	A term encoded for a statement, before it is written into the statement.
TermEncoder

Functions:

Name	Description
split_iri	Split iri into prefix and name.
write_term	Write an encoded subject (slot 0), predicate (1) or object (2).
write_graph	Write an encoded graph name.
encode_triple	Encode one triple, adding its rows to the encoder's flow.
encode_quad	Encode one quad, adding its rows to the encoder's flow.
encode_namespace_declaration	Encode namespace declaration, adding its rows to the encoder's flow.
encode_options	Encode stream options to ProtoBuf message.

EncodedTerm()

A term encoded for a statement, before it is written into the statement.

A statement is written in two steps: first its terms are encoded (adding new lookup entries to the frame), then the terms are written into the statement's row (row layout) or columns (column layout). This way the row is filled in place, and never copied.

Source code in pyjelly/serialize/encode.py

def __init__(self) -> None:
    #: One of the TERM_* constants
    self.kind = 0
    #: IRI: prefix index; literal: datatype index
    self.first = 0
    #: IRI: name index
    self.second = 0
    #: Literal: lexical form; blank node: label
    self.text = ""
    #: Literal: language tag, empty if none
    self.lang = ""
    #: Literal: base direction, as an RdfBaseDirection value, 0 if none
    self.direction = 0
    #: Quoted triple, in the row layout
    self.triple: jelly.RdfTriple | None = None
    #: Triple term, in the column layout
    self.triple_term: jelly.RdfTripleTerm | None = None

TermEncoder(lookup_preset=None)

Methods:

Name	Description
same_term	Return whether two terms are equal, for repeated terms.
new_epoch	Start encoding a new statement (or other row with terms) in the row layout.
new_frame	Start encoding a new frame in the column layout.
has_room	Return whether one more statement fits in the frame, see new_frame().
reserve_iris	Take IRIs that later statements may need out of the frame's budget.
iri_indices	Return the indices of an IRI's prefix and name entries, adding new entries.
encode_iri_indices	Encode lookup indices for IRI, as written in the row layout.
encode_iri	Encode iri into out, adding new lookup entries.
encode_bnode	Encode a blank node into out.
encode_default_graph	Encode default graph into out.
encode_literal	Encode literal into out, adding a new datatype entry.
encode_direction	Return the RdfBaseDirection value of a base direction.
encode_datatype	Return the index of a datatype, adding an entry if it's new.
encode_quoted_triple	Encode a quoted triple into out, adding new lookup entries.
encode_triple_term	Encode a triple term of the column layout, adding new lookup entries.
encode_spo	Encode a subject, predicate or object into out.
encode_graph	Encode a graph name into out.

Source code in pyjelly/serialize/encode.py

def __init__(
    self,
    lookup_preset: options.LookupPreset | None = None,
) -> None:
    if lookup_preset is None:
        lookup_preset = options.LookupPreset()
    self.lookup_preset = lookup_preset
    self.names = LookupEncoder(lookup_size=lookup_preset.max_names)
    self.prefixes = LookupEncoder(lookup_size=lookup_preset.max_prefixes)
    self.datatypes = LookupEncoder(lookup_size=lookup_preset.max_datatypes)
    # Where rows for new lookup entries go in the row layout. A stream replaces
    # it with its own.
    self.flow: FrameFlow = ManualFrameFlow()
    self.iri_cache: dict[str, _CachedIri] = {}
    # Cleared when full, so that a long stream doesn't grow it without bound
    self.iri_cache_size = max(4 * lookup_preset.max_names, 1024)
    # The encoded subject, predicate, object and graph of the statement
    self.encoded = [EncodedTerm() for _ in range(4)]
    # Set by streams of Jelly-RDF 1.2. Then IRIs are encoded with the indices of
    # their entries, new entries are packed (see PackedEntries), and quoted
    # triples are encoded as triple terms.
    self.column_layout = False
    # The RDF version that a Jelly-RDF 1.2 stream declares, which limits its terms
    self.rdf_version: int = jelly.RDF_VERSION_UNSPECIFIED
    # The IRI sequence of the triple terms of a frame (only the object column can
    # have them): the indices of the last IRI. -1 makes the first IRI state its
    # prefix.
    self.triple_prefix_index = -1
    self.triple_name_index = 0
    # How deep the triple term being encoded is nested, and how many IRIs the
    # outermost one has so far
    self.triple_depth = 0
    self.triple_iris = 0

same_term(a, b)

Return whether two terms are equal, for repeated terms.

Integrations can make it faster for their term classes.

Source code in pyjelly/serialize/encode.py

def same_term(self, a: object, b: object) -> bool:
    """
    Return whether two terms are equal, for repeated terms.

    Integrations can make it faster for their term classes.
    """
    return a is b or a == b

new_epoch()

Start encoding a new statement (or other row with terms) in the row layout.

Until the next epoch, the lookup entries it uses are not evicted: the row still refers to them.

Source code in pyjelly/serialize/encode.py

def new_epoch(self) -> None:
    """
    Start encoding a new statement (or other row with terms) in the row layout.

    Until the next epoch, the lookup entries it uses are not evicted: the row
    still refers to them.
    """
    self.names.lookup.new_epoch()
    self.prefixes.lookup.new_epoch()
    self.datatypes.lookup.new_epoch()

new_frame(iris, literals=1)

Start encoding a new frame in the column layout.

The frame is one epoch of the lookups: all its entries must fit in them, because they are all applied before its statements are read. So each lookup gets a budget, which leaves room for one more statement (see has_room()).

Args: iris (int): the most IRIs that one statement can have. literals (int): maximum number of literals in one statement or row. In a statement, only the object can be a literal.

Source code in pyjelly/serialize/encode.py

def new_frame(self, iris: int, literals: int = 1) -> None:
    """
    Start encoding a new frame in the column layout.

    The frame is one epoch of the lookups: all its entries must fit in them,
    because they are all applied before its statements are read. So each lookup
    gets a budget, which leaves room for one more statement (see has_room()).

    Args:
        iris (int): the most IRIs that one statement can have.
        literals (int): maximum number of literals in one statement or row. In a
            statement, only the object can be a literal.

    """
    self.names.lookup.new_epoch(_budget(self.names, iris))
    self.prefixes.lookup.new_epoch(_budget(self.prefixes, iris))
    self.datatypes.lookup.new_epoch(_budget(self.datatypes, literals))
    self.triple_prefix_index = -1
    self.triple_name_index = 0

has_room()

Return whether one more statement fits in the frame, see new_frame().

Source code in pyjelly/serialize/encode.py

def has_room(self) -> bool:
    """Return whether one more statement fits in the frame, see new_frame()."""
    return (
        self.names.lookup.remaining >= 0
        and self.prefixes.lookup.remaining >= 0
        and self.datatypes.lookup.remaining >= 0
    )

reserve_iris(count)

Take IRIs that later statements may need out of the frame's budget.

Source code in pyjelly/serialize/encode.py

def reserve_iris(self, count: int) -> None:
    """Take IRIs that later statements may need out of the frame's budget."""
    self.names.lookup.remaining -= count
    self.prefixes.lookup.remaining -= count

iri_indices(iri_string)

Return the indices of an IRI's prefix and name entries, adding new entries.

Most IRIs repeat, so how an IRI splits into prefix and name, and the indices of their entries, are cached (see _CachedIri).

Args: iri_string (str): full iri in string format.

Returns: tuple[int, int]: indices in prefix and name tables.

Source code in pyjelly/serialize/encode.py

def iri_indices(self, iri_string: str) -> tuple[int, int]:
    """
    Return the indices of an IRI's prefix and name entries, adding new entries.

    Most IRIs repeat, so how an IRI splits into prefix and name, and the
    indices of their entries, are cached (see _CachedIri).

    Args:
        iri_string (str): full iri in string format.

    Returns:
        tuple[int, int]: indices in prefix and name tables.

    """
    cached = self.iri_cache.get(iri_string)
    if cached is None:
        if len(self.iri_cache) >= self.iri_cache_size:
            self.iri_cache.clear()
        if self.prefixes.lookup.max_size:
            prefix, name = split_iri(iri_string)
        else:
            prefix, name = "", iri_string
        cached = _CachedIri(prefix, name)
        self.iri_cache[iri_string] = cached

    prefix_index = 0
    prefixes = self.prefixes
    if prefixes.lookup.max_size:
        prefix_index = cached.prefix_index
        if prefix_index and prefixes.lookup.keys[prefix_index] is cached.prefix:
            prefixes.lookup.touch(prefix_index)
        else:
            prefix_index = self._new_prefix_index(cached)

    names = self.names
    name_index = cached.name_index
    if name_index and names.lookup.keys[name_index] is cached.name:
        names.lookup.touch(name_index)
    else:
        name_index = self._new_name_index(cached)
    return prefix_index, name_index

encode_iri_indices(iri_string)

Encode lookup indices for IRI, as written in the row layout.

The indices are relative to the IRI before (see RdfIri), so IRIs must be encoded in the order they are written.

Args: iri_string (str): full iri in string format.

Returns: tuple[int, int]: indices in prefix and name tables.

Source code in pyjelly/serialize/encode.py

def encode_iri_indices(self, iri_string: str) -> tuple[int, int]:
    """
    Encode lookup indices for IRI, as written in the row layout.

    The indices are relative to the IRI before (see RdfIri), so IRIs must be
    encoded in the order they are written.

    Args:
        iri_string (str): full iri in string format.

    Returns:
        tuple[int, int]: indices in prefix and name tables.

    """
    prefix_index, name_index = self.iri_indices(iri_string)
    prefixes = self.prefixes
    if prefixes.lookup.max_size:
        prefix_index = prefixes.prefix_index_to_write(
            prefix_index, empty=not prefixes.lookup.keys[prefix_index]
        )
    return prefix_index, self.names.name_index_to_write(name_index)

encode_iri(iri_string, out)

Encode iri into out, adding new lookup entries.

Source code in pyjelly/serialize/encode.py

def encode_iri(self, iri_string: str, out: EncodedTerm) -> None:
    """Encode iri into `out`, adding new lookup entries."""
    if self.column_layout:
        prefix_index, name_index = self.iri_indices(iri_string)
    else:
        prefix_index, name_index = self.encode_iri_indices(iri_string)
    out.kind = TERM_IRI
    out.first = prefix_index
    out.second = name_index

encode_bnode(label, out)

Encode a blank node into out.

Source code in pyjelly/serialize/encode.py

def encode_bnode(self, label: str, out: EncodedTerm) -> None:
    """Encode a blank node into `out`."""
    out.kind = TERM_BNODE
    out.text = label

encode_default_graph(out)

Encode default graph into out.

Source code in pyjelly/serialize/encode.py

def encode_default_graph(self, out: EncodedTerm) -> None:
    """Encode default graph into `out`."""
    out.kind = TERM_DEFAULT_GRAPH

encode_literal(*, lex, language=None, datatype=None, direction=None, out)

Encode literal into out, adding a new datatype entry.

Args: lex (str): lexical form/literal value language (str | None, optional): langtag. Defaults to None. datatype (str | None, optional): data type if it is a typed literal. Defaults to None. direction (str | None, optional): base direction of a language-tagged string, "ltr" or "rtl". Defaults to None. out (EncodedTerm): where to put the encoded literal.

Raises: JellyConformanceError: if datatype specified while datatable is not used, or if the stream can't have the base direction.

Source code in pyjelly/serialize/encode.py

def encode_literal(
    self,
    *,
    lex: str,
    language: str | None = None,
    datatype: str | None = None,
    direction: str | None = None,
    out: EncodedTerm,
) -> None:
    """
    Encode literal into `out`, adding a new datatype entry.

    Args:
        lex (str): lexical form/literal value
        language (str | None, optional): langtag. Defaults to None.
        datatype (str | None, optional): data type if
        it is a typed literal. Defaults to None.
        direction (str | None, optional): base direction of a language-tagged
            string, "ltr" or "rtl". Defaults to None.
        out (EncodedTerm): where to put the encoded literal.

    Raises:
        JellyConformanceError: if datatype specified while
            datatable is not used, or if the stream can't have the base
            direction.

    """
    datatype_index = 0
    direction_value = 0
    if language:
        if direction:
            direction_value = self.encode_direction(direction)
    elif datatype and datatype != options.STRING_DATATYPE_IRI:
        datatype_index = self.encode_datatype(datatype)
    out.kind = TERM_LITERAL
    out.text = lex
    out.lang = language or ""
    out.first = datatype_index
    out.direction = direction_value

encode_direction(direction)

Return the RdfBaseDirection value of a base direction.

Source code in pyjelly/serialize/encode.py

def encode_direction(self, direction: str) -> int:
    """Return the RdfBaseDirection value of a base direction."""
    value = _DIRECTIONS.get(direction, 0)
    if not value:
        msg = f"unknown base direction: {direction!r}, expected 'ltr' or 'rtl'"
        raise JellyConformanceError(msg)
    if not self.column_layout:
        msg = (
            "literals with a base direction can only be written in Jelly-RDF 1.2 "
            "(protocol version 3)"
        )
        raise JellyConformanceError(msg)
    if self.rdf_version == jelly.RDF_VERSION_1_1:
        msg = "the stream declares RDF 1.1, which has no base directions"
        raise JellyConformanceError(msg)
    return value

encode_datatype(datatype)

Return the index of a datatype, adding an entry if it's new.

Source code in pyjelly/serialize/encode.py

def encode_datatype(self, datatype: str) -> int:
    """Return the index of a datatype, adding an entry if it's new."""
    datatypes = self.datatypes
    if datatypes.lookup.max_size == 0:
        msg = (
            f"can't encode literal with type {datatype}: "
            "datatype lookup cannot be used if disabled "
            "(its size was set to 0)"
        )
        raise JellyConformanceError(msg)
    index, set_index = datatypes.encode_entry(datatype)
    if set_index >= 0:
        if not self.column_layout:
            entry = self.flow.new_row().datatype
            entry.id = set_index
            entry.value = datatype
        elif datatype in (
            options.LANG_STRING_DATATYPE_IRI,
            options.DIR_LANG_STRING_DATATYPE_IRI,
        ):
            msg = f"a literal with the datatype {datatype} must have a language tag"
            raise JellyConformanceError(msg)
        else:
            datatypes.packed.add(index, datatype)
    return index

encode_quoted_triple(terms, out)

Encode a quoted triple into out, adding new lookup entries.

In the column layout, it is a triple term, see encode_triple_term().

Notes: Although a triple, it is treated as a part of a statement. Repeated terms are not used when encoding quoted triples.

Args: terms (Iterable[object]): triple terms to encode. out (EncodedTerm): where to put the encoded triple.

Source code in pyjelly/serialize/encode.py

def encode_quoted_triple(self, terms: Iterable[object], out: EncodedTerm) -> None:
    """
    Encode a quoted triple into `out`, adding new lookup entries.

    In the column layout, it is a triple term, see encode_triple_term().

    Notes:
        Although a triple, it is treated as a part of a statement.
        Repeated terms are not used when encoding quoted triples.

    Args:
        terms (Iterable[object]): triple terms to encode.
        out (EncodedTerm): where to put the encoded triple.

    """
    if self.column_layout:
        out.triple_term = self.encode_triple_term(terms)
        out.kind = TERM_TRIPLE
        return
    terms = iter(terms)
    triple = jelly.RdfTriple()
    for slot in range(3):
        term = EncodedTerm()
        self.encode_spo(next(terms), term)
        write_term(term, triple, slot)
    out.kind = TERM_TRIPLE
    out.triple = triple

encode_triple_term(terms)

Encode a triple term of the column layout, adding new lookup entries.

Its IRIs are in the IRI sequence of the frame's triple terms, in the order subject, predicate, object. triple_iris counts the IRIs of the outermost triple term, nested ones included.

Args: terms (Iterable[object]): the subject, predicate and object.

Returns: jelly.RdfTripleTerm: the triple term.

Source code in pyjelly/serialize/encode.py

def encode_triple_term(self, terms: Iterable[object]) -> jelly.RdfTripleTerm:
    """
    Encode a triple term of the column layout, adding new lookup entries.

    Its IRIs are in the IRI sequence of the frame's triple terms, in the order
    subject, predicate, object. `triple_iris` counts the IRIs of the outermost
    triple term, nested ones included.

    Args:
        terms (Iterable[object]): the subject, predicate and object.

    Returns:
        jelly.RdfTripleTerm: the triple term.

    """
    depth = self.triple_depth
    if not depth:
        rdf_version = self.rdf_version
        if rdf_version in (jelly.RDF_VERSION_1_1, jelly.RDF_VERSION_1_2_BASIC):
            name = (
                "RDF 1.1"
                if rdf_version == jelly.RDF_VERSION_1_1
                else "RDF 1.2 Basic"
            )
            msg = f"the stream declares {name}, which has no triple terms"
            raise JellyConformanceError(msg)
        self.triple_iris = 0
    elif depth >= MAX_TRIPLE_TERM_DEPTH:
        msg = (
            f"triple terms nested deeper than {MAX_TRIPLE_TERM_DEPTH} levels "
            "are not supported"
        )
        raise JellyConformanceError(msg)
    self.triple_depth = depth + 1
    try:
        triple = jelly.RdfTripleTerm()
        terms = iter(terms)
        term = EncodedTerm()
        self.encode_spo(next(terms), term)
        if term.kind == TERM_IRI:
            self._triple_term_iri(term, triple.s_iri)
        elif term.kind == TERM_BNODE:
            triple.s_bnode = term.text
        else:
            msg = "the subject of a triple term must be an IRI or a blank node"
            raise JellyConformanceError(msg)
        self.encode_spo(next(terms), term)
        if term.kind != TERM_IRI:
            msg = "the predicate of a triple term must be an IRI"
            raise JellyConformanceError(msg)
        self._triple_term_iri(term, triple.p_iri)
        self.encode_spo(next(terms), term)
        if term.kind == TERM_IRI:
            self._triple_term_iri(term, triple.o_iri)
        elif term.kind == TERM_BNODE:
            triple.o_bnode = term.text
        elif term.kind == TERM_LITERAL:
            _write_literal(term, triple.o_literal)
        else:
            assert term.triple_term is not None
            triple.o_triple_term.CopyFrom(term.triple_term)
    finally:
        self.triple_depth = depth
    return triple

encode_spo(term, out)

Encode a subject, predicate or object into out.

Source code in pyjelly/serialize/encode.py

def encode_spo(self, term: object, out: EncodedTerm) -> None:
    """Encode a subject, predicate or object into `out`."""
    msg = f"unsupported term type: {type(term)}"
    raise NotImplementedError(msg)

encode_graph(term, out)

Encode a graph name into out.

Source code in pyjelly/serialize/encode.py

def encode_graph(self, term: object, out: EncodedTerm) -> None:
    """Encode a graph name into `out`."""
    msg = f"unsupported term type: {type(term)}"
    raise NotImplementedError(msg)

split_iri(iri_string)

Split iri into prefix and name.

Args: iri_string (str): full iri string.

Returns: tuple[str, str]: iri's prefix and name.

Source code in pyjelly/serialize/encode.py

def split_iri(iri_string: str) -> tuple[str, str]:
    """
    Split iri into prefix and name.

    Args:
        iri_string (str): full iri string.

    Returns:
        tuple[str, str]: iri's prefix and name.

    """
    # After the last "#", or else the last "/". Without either, all is the name.
    i = iri_string.rfind("#")
    if i < 0:
        i = iri_string.rfind("/")
    return iri_string[: i + 1], iri_string[i + 1 :]

write_term(term, statement, slot)

Write an encoded subject (slot 0), predicate (1) or object (2).

Source code in pyjelly/serialize/encode.py

def write_term(term: EncodedTerm, statement: Statement, slot: int) -> None:
    """Write an encoded subject (slot 0), predicate (1) or object (2)."""
    kind = term.kind
    if kind == TERM_IRI:
        iri = _iri_field(statement, slot)
        iri.prefix_id = term.first
        iri.name_id = term.second
    elif kind == TERM_LITERAL:
        _write_literal(term, _literal_field(statement, slot))
    elif kind == TERM_BNODE:
        _set_bnode(statement, slot, term.text)
    elif kind == TERM_TRIPLE:
        assert term.triple is not None
        _triple_field(statement, slot).CopyFrom(term.triple)

write_graph(term, quad)

Write an encoded graph name.

Source code in pyjelly/serialize/encode.py

def write_graph(term: EncodedTerm, quad: jelly.RdfQuad) -> None:
    """Write an encoded graph name."""
    kind = term.kind
    if kind == TERM_IRI:
        iri = quad.g_iri
        iri.prefix_id = term.first
        iri.name_id = term.second
    elif kind == TERM_DEFAULT_GRAPH:
        quad.g_default_graph.SetInParent()
    elif kind == TERM_BNODE:
        quad.g_bnode = term.text
    elif kind == TERM_LITERAL:
        _write_literal(term, quad.g_literal)

encode_triple(terms, term_encoder, repeated_terms)

Encode one triple, adding its rows to the encoder's flow.

Args: terms (Iterable[object]): original terms to encode term_encoder (TermEncoder): current encoder with lookup tables repeated_terms (list[object | None]): list of repeated terms.

Source code in pyjelly/serialize/encode.py

def encode_triple(
    terms: Iterable[object],
    term_encoder: TermEncoder,
    repeated_terms: list[object | None],
) -> None:
    """
    Encode one triple, adding its rows to the encoder's flow.

    Args:
        terms (Iterable[object]): original terms to encode
        term_encoder (TermEncoder): current encoder with lookup tables
        repeated_terms (list[object | None]): list of repeated terms.

    """
    term_encoder.new_epoch()
    encoded = term_encoder.encoded
    terms = iter(terms)
    # First encode the terms that changed, which adds rows for new lookup
    # entries, then add the triple's row after them and write the terms
    changed = 0
    for slot in range(3):
        term = next(terms)
        if not term_encoder.same_term(repeated_terms[slot], term):
            term_encoder.encode_spo(term, encoded[slot])
            repeated_terms[slot] = term
            changed |= 1 << slot
    triple = term_encoder.flow.new_row().triple
    if not changed:
        # Every term is repeated: the row is an empty triple
        triple.SetInParent()
    for slot in range(3):
        if changed & (1 << slot):
            write_term(encoded[slot], triple, slot)

encode_quad(terms, term_encoder, repeated_terms)

Encode one quad, adding its rows to the encoder's flow.

Args: terms (Iterable[object]): original terms to encode term_encoder (TermEncoder): current encoder with lookup tables repeated_terms (list[object | None]): list of repeated terms.

Source code in pyjelly/serialize/encode.py

def encode_quad(
    terms: Iterable[object],
    term_encoder: TermEncoder,
    repeated_terms: list[object | None],
) -> None:
    """
    Encode one quad, adding its rows to the encoder's flow.

    Args:
        terms (Iterable[object]): original terms to encode
        term_encoder (TermEncoder): current encoder with lookup tables
        repeated_terms (list[object | None]): list of repeated terms.

    """
    term_encoder.new_epoch()
    encoded = term_encoder.encoded
    terms = iter(terms)
    # See encode_triple()
    changed = 0
    for slot in range(3):
        term = next(terms)
        if not term_encoder.same_term(repeated_terms[slot], term):
            term_encoder.encode_spo(term, encoded[slot])
            repeated_terms[slot] = term
            changed |= 1 << slot
    g = next(terms)
    if not term_encoder.same_term(repeated_terms[3], g):
        term_encoder.encode_graph(g, encoded[3])
        repeated_terms[3] = g
        changed |= 1 << 3
    quad = term_encoder.flow.new_row().quad
    if not changed:
        quad.SetInParent()
    for slot in range(3):
        if changed & (1 << slot):
            write_term(encoded[slot], quad, slot)
    if changed & (1 << 3):
        write_graph(encoded[3], quad)

encode_namespace_declaration(name, value, term_encoder)

Encode namespace declaration, adding its rows to the encoder's flow.

Args: name (str): namespace prefix label value (str): namespace iri term_encoder (TermEncoder): current encoder

Source code in pyjelly/serialize/encode.py

def encode_namespace_declaration(
    name: str,
    value: str,
    term_encoder: TermEncoder,
) -> None:
    """
    Encode namespace declaration, adding its rows to the encoder's flow.

    Args:
        name (str): namespace prefix label
        value (str): namespace iri
        term_encoder (TermEncoder): current encoder

    """
    term_encoder.new_epoch()
    # The rows for new lookup entries go first
    prefix_index, name_index = term_encoder.encode_iri_indices(value)
    declaration = term_encoder.flow.new_row().namespace
    declaration.name = name
    declaration.value.prefix_id = prefix_index
    declaration.value.name_id = name_index

encode_options(lookup_preset, stream_types, params)

Encode stream options to ProtoBuf message.

Args: lookup_preset (options.LookupPreset): lookup tables options stream_types (options.StreamTypes): physical and logical types params (options.StreamParameters): other params.

Returns: jelly.RdfStreamOptions: encoded stream options

Source code in pyjelly/serialize/encode.py

def encode_options(
    lookup_preset: options.LookupPreset,
    stream_types: options.StreamTypes,
    params: options.StreamParameters,
) -> jelly.RdfStreamOptions:
    """
    Encode stream options to ProtoBuf message.

    Args:
        lookup_preset (options.LookupPreset): lookup tables options
        stream_types (options.StreamTypes): physical and logical types
        params (options.StreamParameters): other params.

    Returns:
        jelly.RdfStreamOptions: encoded stream options

    """
    if options.is_row_layout(params.version):
        return jelly.RdfStreamOptions(
            stream_name=params.stream_name,
            physical_type=stream_types.physical_type,
            generalized_statements=params.generalized_statements,
            rdf_star=params.rdf_star,
            max_name_table_size=lookup_preset.max_names,
            max_prefix_table_size=lookup_preset.max_prefixes,
            max_datatype_table_size=lookup_preset.max_datatypes,
            logical_type=stream_types.logical_type,
            version=params.version,
        )
    return jelly.RdfStreamOptions(
        stream_name=params.stream_name,
        physical_type=stream_types.physical_type,
        stream_type=options.stream_type_for(params, stream_types.logical_type),
        rdf_version=params.rdf_version,
        max_name_table_size=lookup_preset.max_names,
        max_prefix_table_size=lookup_preset.max_prefixes,
        max_datatype_table_size=lookup_preset.max_datatypes,
        version=params.version,
    )

flows

Classes:

Name	Description
FrameFlow	Abstract base class for producing Jelly frames from RDF stream rows.
ManualFrameFlow	Produces frames only when manually requested (never automatically).
BoundedFrameFlow	Produce frames automatically when a fixed number of rows is reached.
GraphsFrameFlow
DatasetsFrameFlow

Functions:

Name	Description
flow_for_type	Return flow based on logical type requested.

FrameFlow(*, logical_type=None)

Abstract base class for producing Jelly frames from RDF stream rows.

Rows are added straight into the frame being built (see new_row()), so they never have to be copied: protobuf copies a message by serializing it and parsing it again.

In Jelly-RDF 1.2 streams, the frame's statements are in its column batch (see batch), and the rows only store the stream options. A frame may then also end because the lookups are full.

Allows for passing LogicalStreamType, required for logical subtypes and non-delimited streams.

Methods:

Name	Description
new_row	Add an empty row at the end of the current frame, and return it.
__len__	Return how many rows (in Jelly-RDF 1.2, statements) the frame has.
frame_from_graph	End a graph: in a stream of RDF Messages, end its message.
frame_from_dataset	End a dataset: in a stream of RDF Messages, end its message.
end_message	End the current RDF Message, in a stream of RDF Messages.
frame_from_bounds	Produce a frame if no more statements fit in it.
to_stream_frame	Return the current frame, and start a new one.

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
) -> None:
    self.logical_type = logical_type or self.default_logical_type
    self.frame = jelly.RdfStreamFrame()
    self.rows = self.frame.rows
    #: Set by streams of Jelly-RDF 1.2: the statements of the frame being built
    self.batch: ColumnBatch | None = None

new_row()

Add an empty row at the end of the current frame, and return it.

Source code in pyjelly/serialize/flows.py

def new_row(self) -> jelly.RdfStreamRow:
    """Add an empty row at the end of the current frame, and return it."""
    return self.rows.add()

__len__()

Return how many rows (in Jelly-RDF 1.2, statements) the frame has.

Source code in pyjelly/serialize/flows.py

def __len__(self) -> int:
    """Return how many rows (in Jelly-RDF 1.2, statements) the frame has."""
    batch = self.batch
    if batch is not None:
        return batch.row_count
    return len(self.rows)

frame_from_graph()

End a graph: in a stream of RDF Messages, end its message.

Default implementation returns None: the frame goes on.

Source code in pyjelly/serialize/flows.py

def frame_from_graph(self) -> jelly.RdfStreamFrame | None:
    """
    End a graph: in a stream of RDF Messages, end its message.

    Default implementation returns None: the frame goes on.
    """
    self.end_message()
    return None

frame_from_dataset()

End a dataset: in a stream of RDF Messages, end its message.

Default implementation returns None: the frame goes on.

Source code in pyjelly/serialize/flows.py

def frame_from_dataset(self) -> jelly.RdfStreamFrame | None:
    """
    End a dataset: in a stream of RDF Messages, end its message.

    Default implementation returns None: the frame goes on.
    """
    self.end_message()
    return None

end_message()

End the current RDF Message, in a stream of RDF Messages.

Source code in pyjelly/serialize/flows.py

def end_message(self) -> None:
    """End the current RDF Message, in a stream of RDF Messages."""
    batch = self.batch
    if batch is not None and batch.messages:
        batch.end_message()

frame_from_bounds()

Produce a frame if no more statements fit in it.

Only in Jelly-RDF 1.2, where the lookups must fit all the frame's entries.

Source code in pyjelly/serialize/flows.py

def frame_from_bounds(self) -> jelly.RdfStreamFrame | None:
    """
    Produce a frame if no more statements fit in it.

    Only in Jelly-RDF 1.2, where the lookups must fit all the frame's entries.
    """
    batch = self.batch
    if batch is not None and not batch.has_room():
        return self.to_stream_frame()
    return None

to_stream_frame()

Return the current frame, and start a new one.

Returns: jelly.RdfStreamFrame | None: stream frame, None if it is empty

Source code in pyjelly/serialize/flows.py

def to_stream_frame(self) -> jelly.RdfStreamFrame | None:
    """
    Return the current frame, and start a new one.

    Returns:
        jelly.RdfStreamFrame | None: stream frame, None if it is empty

    """
    batch = self.batch
    if batch is not None and batch.has_content():
        batch.write(self.frame.columns)
    elif not len(self.rows):
        return None
    frame = self.frame
    self.frame = jelly.RdfStreamFrame()
    self.rows = self.frame.rows
    return frame

ManualFrameFlow(*, logical_type=None)

Bases: FrameFlow

Produces frames only when manually requested (never automatically).

Warning

All stream rows are kept in memory until to_stream_frame() is called. This may lead to high memory usage for large streams.

Used for non-delimited serialization.

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
) -> None:
    self.logical_type = logical_type or self.default_logical_type
    self.frame = jelly.RdfStreamFrame()
    self.rows = self.frame.rows
    #: Set by streams of Jelly-RDF 1.2: the statements of the frame being built
    self.batch: ColumnBatch | None = None

BoundedFrameFlow(*, logical_type=None, frame_size=None)

Bases: FrameFlow

Produce frames automatically when a fixed number of rows is reached.

In Jelly-RDF 1.2, it is a number of statements. Used for delimited encoding (default mode).

Methods:

Name	Description
frame_from_bounds	Emit frame from flow if full.

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
    frame_size: int | None = None,
) -> None:
    super().__init__(logical_type=logical_type)
    self.frame_size = frame_size or DEFAULT_FRAME_SIZE

frame_from_bounds()

Emit frame from flow if full.

Returns: jelly.RdfStreamFrame | None: stream frame

Source code in pyjelly/serialize/flows.py

@override
def frame_from_bounds(self) -> jelly.RdfStreamFrame | None:
    """
    Emit frame from flow if full.

    Returns:
        jelly.RdfStreamFrame | None: stream frame

    """
    batch = self.batch
    if batch is None:
        if len(self.rows) >= self.frame_size:
            return self.to_stream_frame()
    elif batch.row_count >= self.frame_size or not batch.has_room():
        return self.to_stream_frame()
    return None

GraphsFrameFlow(*, logical_type=None)

Bases: FrameFlow

Methods:

Name	Description
frame_from_graph	Emit current flow content (one graph) as jelly frame.

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
) -> None:
    self.logical_type = logical_type or self.default_logical_type
    self.frame = jelly.RdfStreamFrame()
    self.rows = self.frame.rows
    #: Set by streams of Jelly-RDF 1.2: the statements of the frame being built
    self.batch: ColumnBatch | None = None

frame_from_graph()

Emit current flow content (one graph) as jelly frame.

Returns: jelly.RdfStreamFrame | None: jelly frame or none if flow is empty.

Source code in pyjelly/serialize/flows.py

def frame_from_graph(self) -> jelly.RdfStreamFrame | None:
    """
    Emit current flow content (one graph) as jelly frame.

    Returns:
        jelly.RdfStreamFrame | None: jelly frame or none if
            flow is empty.

    """
    self.end_message()
    return self.to_stream_frame()

DatasetsFrameFlow(*, logical_type=None)

Bases: FrameFlow

Methods:

Name	Description
frame_from_dataset	Emit current flow content (dataset) as jelly frame.

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
) -> None:
    self.logical_type = logical_type or self.default_logical_type
    self.frame = jelly.RdfStreamFrame()
    self.rows = self.frame.rows
    #: Set by streams of Jelly-RDF 1.2: the statements of the frame being built
    self.batch: ColumnBatch | None = None

frame_from_dataset()

Emit current flow content (dataset) as jelly frame.

Returns: jelly.RdfStreamFrame | None: jelly frame or none if flow is empty.

Source code in pyjelly/serialize/flows.py

def frame_from_dataset(self) -> jelly.RdfStreamFrame | None:
    """
    Emit current flow content (dataset) as jelly frame.

    Returns:
        jelly.RdfStreamFrame | None: jelly frame or none if
            flow is empty.

    """
    self.end_message()
    return self.to_stream_frame()

flow_for_type(logical_type)

Return flow based on logical type requested.

Note: uses base logical type for subtypes (i.e., SUBJECT_GRAPHS uses the same flow as its base type GRAPHS).

Args: logical_type (jelly.LogicalStreamType): logical type requested.

Raises: NotImplementedError: if (base) logical stream type is not supported.

Returns: type[FrameFlow]: FrameFlow for respective logical type.

Source code in pyjelly/serialize/flows.py

def flow_for_type(logical_type: jelly.LogicalStreamType) -> type[FrameFlow]:
    """
    Return flow based on logical type requested.

    Note: uses base logical type for subtypes (i.e., SUBJECT_GRAPHS uses
        the same flow as its base type GRAPHS).

    Args:
        logical_type (jelly.LogicalStreamType): logical type requested.

    Raises:
        NotImplementedError: if (base) logical stream type is not supported.

    Returns:
        type[FrameFlow]: FrameFlow for respective logical type.

    """
    try:
        base_logical_type_value = logical_type % 10
        base_name = jelly.LogicalStreamType.Name(base_logical_type_value)
        return FLOW_DISPATCH[getattr(jelly.LogicalStreamType, base_name)]
    except KeyError:
        msg = (
            "unsupported logical stream type: "
            f"{jelly.LogicalStreamType.Name(logical_type)}"
        )
        raise NotImplementedError(msg) from None

ioutils

Functions:

Name	Description
varint	Encode a non-negative integer as a protobuf varint.
write_delimited	Write a frame prefixed with its size (the delimited variant of Jelly).

varint(value)

Encode a non-negative integer as a protobuf varint.

varint(300) b'\xac\x02'

Source code in pyjelly/serialize/ioutils.py

def varint(value: int) -> bytes:
    r"""
    Encode a non-negative integer as a protobuf varint.

    >>> varint(300)
    b'\xac\x02'
    """
    out = bytearray()
    while value > _VARINT_PAYLOAD:
        out.append((value & _VARINT_PAYLOAD) | _VARINT_CONTINUATION)
        value >>= 7
    out.append(value)
    return bytes(out)

write_delimited(frame, output_stream)

Write a frame prefixed with its size (the delimited variant of Jelly).

Source code in pyjelly/serialize/ioutils.py

def write_delimited(frame: Message, output_stream: IO[bytes]) -> None:
    """Write a frame prefixed with its size (the delimited variant of Jelly)."""
    # protobuf's serialize_length_prefixed() needs write() to return the number of
    # bytes written, which not every stream does (e.g., rdflib's result streams)
    data = frame.SerializeToString()
    output_stream.write(varint(len(data)))
    output_stream.write(data)

lookup

Classes:

Name	Description
Lookup	Fixed-size 1-based string-to-index mapping.
PackedEntries	The new entries of a lookup for the next frame of a Jelly-RDF 1.2 stream.
LookupEncoder	Shared base for RDF lookup encoders using Jelly compression.

Lookup(max_size)

Fixed-size 1-based string-to-index mapping.

• Assigns incrementing indices starting from 1.
• When full, evicts entries in index order, skipping those used recently. This way new entries mostly take the previous index + 1, which Jelly writes as 0, making files smaller. The same strategy as in Jelly-JVM.
• Index 0 is reserved for delta encoding in Jelly streams.

An entry is cold if it was not used in the last max_size // 2 uses. An entry is pinned if it was used in the current epoch (see new_epoch()): what is being encoded still refers to it, so it must not be evicted.

An epoch may have a budget: how many entries it may use. remaining counts down for every entry the epoch uses for the first time, and is negative once the epoch went over its budget.

Parameters:

Name	Type	Description	Default
max_size	int	Maximum number of entries. Zero disables lookup.	required

Methods:

Name	Description
new_epoch	Start a new epoch: only the entries used from now on are pinned.
touch	Mark the entry at the index as used.
use	Mark the entry as used, and return its index, or 0 if there is none.
victim	Return the index of the entry to evict from the full lookup.

Source code in pyjelly/serialize/lookup.py

def __init__(self, max_size: int) -> None:
    self.data: dict[str, int] = {}
    self.max_size = max_size
    # By index, starting with an unused index 0: the key, and when the entry
    # was last used or added, counted in uses of this lookup (`now`)
    self.keys: list[str] = [""]
    self.last_use: list[int] = [0]
    self.now = 0
    self.epoch_start = _NO_EPOCH
    self.remaining = NO_BUDGET
    self.cold_age = max_size >> 1
    # The index of the last entry that replaced an evicted one, 0 if none yet
    self.last_evicted = 0

new_epoch(budget=NO_BUDGET)

Start a new epoch: only the entries used from now on are pinned.

Source code in pyjelly/serialize/lookup.py

def new_epoch(self, budget: int = NO_BUDGET) -> None:
    """Start a new epoch: only the entries used from now on are pinned."""
    self.epoch_start = self.now
    self.remaining = budget

touch(index)

Mark the entry at the index as used.

Source code in pyjelly/serialize/lookup.py

def touch(self, index: int) -> None:
    """Mark the entry at the index as used."""
    last_use = self.last_use
    if last_use[index] <= self.epoch_start:
        self.remaining -= 1
    self.now += 1
    last_use[index] = self.now

use(key)

Mark the entry as used, and return its index, or 0 if there is none.

Source code in pyjelly/serialize/lookup.py

def use(self, key: str) -> int:
    """Mark the entry as used, and return its index, or 0 if there is none."""
    index = self.data.get(key, 0)
    if index:
        last_use = self.last_use
        if last_use[index] <= self.epoch_start:
            self.remaining -= 1
        self.now += 1
        last_use[index] = self.now
    return index

victim()

Return the index of the entry to evict from the full lookup.

That is the next index in order (after the last evicted one), if it is cold and not pinned. Else, the least recently used one that is not pinned among a few indices spread evenly over the table, starting with that next index. The next eviction then goes on from the index picked, which is usually among a run of other old entries.

Source code in pyjelly/serialize/lookup.py

def victim(self) -> int:
    """
    Return the index of the entry to evict from the full lookup.

    That is the next index in order (after the last evicted one), if it is cold
    and not pinned. Else, the least recently used one that is not pinned among
    a few indices spread evenly over the table, starting with that next index.
    The next eviction then goes on from the index picked, which is usually
    among a run of other old entries.
    """
    size = self.max_size
    last_use = self.last_use
    epoch_start = self.epoch_start
    index = self.last_evicted + 1 if 1 <= self.last_evicted < size else 1
    last = last_use[index]
    if self.now - last >= self.cold_age and last <= epoch_start:
        return index
    best = 0
    best_last = _NO_EPOCH
    step = max(1, size // _SAMPLES)
    for _ in range(min(_SAMPLES, size)):
        last = last_use[index]
        if last < best_last and last <= epoch_start:
            best = index
            best_last = last
        index += step
        if index > size:
            index -= size
    if best:
        return best
    # The epoch pins every index sampled: the first index that it does not pin
    for _ in range(size):
        if last_use[index] <= epoch_start:
            return index
        index = 1 if index == size else index + 1
    msg = (
        f"a lookup of size {size} is too small to encode a single statement or "
        "row of results: it uses every one of its entries"
    )
    raise JellyConformanceError(msg)

PackedEntries()

The new entries of a lookup for the next frame of a Jelly-RDF 1.2 stream.

Entries with consecutive indices are packed into one RdfLookupEntryPacked.

Methods:

Name	Description
write	Add the packed entries to a repeated RdfLookupEntryPacked field.

Source code in pyjelly/serialize/lookup.py

def __init__(self) -> None:
    # The first index of each packed entry (0 if it follows the entry before),
    # and the values of each
    self.ids: list[int] = []
    self.values: list[list[str]] = []
    # The index of the last entry added, in this frame or an earlier one
    self.last_index = 0

write(out)

Add the packed entries to a repeated RdfLookupEntryPacked field.

Source code in pyjelly/serialize/lookup.py

def write(self, out: Any) -> None:
    """Add the packed entries to a repeated RdfLookupEntryPacked field."""
    for index, values in zip(self.ids, self.values, strict=True):
        entry = out.add()
        entry.id = index
        entry.values.extend(values)
    self.ids.clear()
    self.values.clear()

LookupEncoder(*, lookup_size)

Shared base for RDF lookup encoders using Jelly compression.

Tracks the last assigned and last reused index.

Parameters:

Name	Type	Description	Default
lookup_size	int	Maximum lookup size.	required

Methods:

Name	Description
encode_entry_index	Get or assign the index to use in an entry, and mark the entry as used.
encode_entry	Get or add the entry for the key, and mark it as used.
name_index_to_write	Return how to write a reference to the name entry at the index.
prefix_index_to_write	Return how to write a reference to the prefix entry at the index.

Source code in pyjelly/serialize/lookup.py

def __init__(self, *, lookup_size: int) -> None:
    self.lookup = Lookup(max_size=lookup_size)
    self.last_assigned_index = 0
    self.last_reused_index = 0
    self.packed = PackedEntries()

encode_entry_index(key)

Get or assign the index to use in an entry, and mark the entry as used.

Call it once for every use of the key, before encode_*_term_index().

Returns:

Type	Description
int or None	0 if the new index is sequential (last_assigned_index + 1) actual assigned/reused index otherwise None if the key already exists
If the return value is None, the entry is already in the lookup and does not
need to be emitted. Any integer value (including 0) means the entry is new
and should be emitted.

Source code in pyjelly/serialize/lookup.py

def encode_entry_index(self, key: str) -> int | None:
    """
    Get or assign the index to use in an entry, and mark the entry as used.

    Call it once for every use of the key, before encode_*_term_index().

    Returns
    -------
    int or None
        - 0 if the new index is sequential (`last_assigned_index + 1`)
        - actual assigned/reused index otherwise
        - None if the key already exists

    If the return value is None, the entry is already in the lookup and does not
    need to be emitted. Any integer value (including 0) means the entry is new
    and should be emitted.

    """
    _, set_index = self.encode_entry(key)
    return None if set_index < 0 else set_index

encode_entry(key)

Get or add the entry for the key, and mark it as used.

Returns:

Type	Description
tuple[int, int]	The entry's index, and the index to write in a new entry row: -1 if the entry was already there (no row needed), 0 if the new index is sequential (last_assigned_index + 1), else the index.

Source code in pyjelly/serialize/lookup.py

def encode_entry(self, key: str) -> tuple[int, int]:
    """
    Get or add the entry for the key, and mark it as used.

    Returns
    -------
    tuple[int, int]
        The entry's index, and the index to write in a new entry row: -1 if
        the entry was already there (no row needed), 0 if the new index is
        sequential (`last_assigned_index + 1`), else the index.

    """
    index = self.lookup.use(key)
    if index:
        return index, -1
    previous_index = self.last_assigned_index
    index = self.lookup.insert(key)
    self.last_assigned_index = index
    if index == previous_index + 1:
        return index, 0
    return index, index

name_index_to_write(index)

Return how to write a reference to the name entry at the index.

Source code in pyjelly/serialize/lookup.py

def name_index_to_write(self, index: int) -> int:
    """Return how to write a reference to the name entry at the index."""
    previous_index = self.last_reused_index
    self.last_reused_index = index
    if index == previous_index + 1:
        return 0
    return index

prefix_index_to_write(index, *, empty)

Return how to write a reference to the prefix entry at the index.

Source code in pyjelly/serialize/lookup.py

def prefix_index_to_write(self, index: int, *, empty: bool) -> int:
    """Return how to write a reference to the prefix entry at the index."""
    previous_index = self.last_reused_index
    if empty and previous_index == 0:
        return 0
    self.last_reused_index = index
    if previous_index == 0:
        return index
    if index == previous_index:
        return 0
    return index

sparql

Jelly-SPARQL writer: SPARQL query results in frames.

Classes:

Name	Description
VariableColumn	Cells of one variable in the frame being built.
SparqlStream	Writer of SPARQL query results as Jelly-SPARQL frames.
FrameWriter	Writer of frames to a file, either delimited or as a single frame.

Functions:

Name	Description
links_value	Join link IRIs with LF, for the "link" metadata key.
variable_names	Strip a leading "?" or "$" from each variable name.
error_message	Format an exception as the error message of a trailer.
write_solutions	Write a solution sequence (the result of a SELECT query) as one result set.
write_bindings	Like write_solutions(), with rows given as bindings.

VariableColumn()

Bases: Column

Cells of one variable in the frame being built.

Source code in pyjelly/serialize/sparql.py

def __init__(self) -> None:
    super().__init__(_ANY, "variable")
    # Prefix and name indices of the last IRI in this column's triple terms.
    # The encoder holds them while it encodes this column (see TermEncoder).
    self.triple_prefix_index = -1
    self.triple_name_index = 0
    # Largest number of IRIs in one triple term of this column so far
    self.max_triple_iris = 1

SparqlStream(encoder, options=None)

Writer of SPARQL query results as Jelly-SPARQL frames.

A result set is either a solution sequence (SELECT) or a boolean result (ASK). For a solution sequence, call start(), then row() for each row, then end(). row() returns a frame when one is full, and end() returns the last frame. ask() returns the single frame of a boolean result.

A stream holds one result set, or any number of them with SparqlOptions(punctuated=True).

Args: encoder (TermEncoder): term encoder, with the same lookup sizes as the options. options (SparqlOptions | None): options of the stream.

Methods:

Name	Description
start	Start a solution sequence.
row	Add a row (solution) to the result set.
binding	Add a row (solution) given as a mapping from variables to values.
rows	Add many rows, see row().
add	Add a value to a column. None is an unbound cell.
end_frame	End the current frame and return it.
write_frame	Fill in the frame: options, header, lookup entries and rows.
write_start	Add the stream options and the header, if they are still due.
end	End the result set and return its last frame, which has the trailer.
ask	Return the frame of a boolean result (of an ASK query).

Source code in pyjelly/serialize/sparql.py

def __init__(
    self, encoder: TermEncoder, options: SparqlOptions | None = None
) -> None:
    if options is None:
        options = SparqlOptions()
    if encoder.lookup_preset != options.lookup_preset:
        msg = "the encoder must have the lookup sizes of the options"
        raise ValueError(msg)
    self.encoder = encoder
    self.options = options
    encoder.column_layout = True
    encoder.rdf_version = options.rdf_version
    self.encoded = EncodedTerm()
    #: Variables of the current result set, None between result sets
    self.variables: list[str] | None = None
    self.columns: list[VariableColumn] = []
    # Reused buffer for the values of a binding, see binding()
    self.values: list[object | None] = []
    self.row_count = 0
    self.rows_per_frame = 1
    # Set when the single frame of a non-delimited stream is full
    self.full = False
    self.links: list[str] = []
    # Whether the stream options and the header still have to be written
    self.options_pending = True
    self.header_pending = False
    # `ended`: the single result set of a FLAT stream is done. `failed`: a row
    # failed to encode. Either way, nothing more can be written.
    self.ended = False
    self.failed = False
    # Column whose triple-term IRI state the encoder currently holds
    self.triple_owner: VariableColumn | None = None

start(variables, links=())

Start a solution sequence.

Args: variables (Iterable[str]): variable names, in projection order. A leading "?" or "$" is removed. links (Sequence[str]): link IRIs for the result set, as in head.link of SPARQL JSON results.

Source code in pyjelly/serialize/sparql.py

def start(self, variables: Iterable[str], links: Sequence[str] = ()) -> None:
    """
    Start a solution sequence.

    Args:
        variables (Iterable[str]): variable names, in projection order. A
            leading "?" or "$" is removed.
        links (Sequence[str]): link IRIs for the result set, as in head.link of
            SPARQL JSON results.

    """
    self._check_can_start()
    names = variable_names(variables)
    self.variables = names
    self.columns = [VariableColumn() for _ in names]
    self.values = [None] * len(names)
    self.rows_per_frame = (
        min(self.options.rows_per_frame(len(names)), MAX_ROWS)
        if self.options.delimited
        else MAX_ROWS
    )
    self.header_pending = True
    self.links = list(links)
    self.new_frame()

row(values)

Add a row (solution) to the result set.

Args: values (Sequence[object | None]): one value per variable, None if the variable is unbound.

Returns: jelly.SparqlResultsFrame | None: the frame, if this row filled it.

Source code in pyjelly/serialize/sparql.py

def row(self, values: Sequence[object | None]) -> jelly.SparqlResultsFrame | None:
    """
    Add a row (solution) to the result set.

    Args:
        values (Sequence[object | None]): one value per variable, None if the
            variable is unbound.

    Returns:
        jelly.SparqlResultsFrame | None: the frame, if this row filled it.

    """
    columns = self.columns
    count = len(values)
    self._check_row(count)
    try:
        # mypyc indexes a typed tuple or list directly. Any other sequence goes
        # through the generic item lookup with a boxed int, which is slow.
        if type(values) is tuple:
            row_tuple: tuple[object | None, ...] = values
            for i in range(count):
                self.add(columns[i], row_tuple[i])
        elif type(values) is list:
            row_list: list[object | None] = values
            for i in range(count):
                self.add(columns[i], row_list[i])
        else:
            for i in range(count):
                self.add(columns[i], values[i])
    except BaseException:
        # The frame is half-written and can't be finished
        self.failed = True
        raise
    self.row_count += 1
    # End the frame as soon as it's full, rather than at the start of the next
    # row, so that a failing next row can't lose it
    if self.row_count >= self.rows_per_frame or not self.encoder.has_room():
        if self.options.delimited:
            return self.end_frame()
        self.full = True
    return None

binding(binding, variables)

Add a row (solution) given as a mapping from variables to values.

Args: binding (Mapping[Any, object | None]): values of the bound variables, keyed by variable or by variable name. variables (Sequence[object]): variables of the result set, in order.

Returns: jelly.SparqlResultsFrame | None: the frame, if this row filled it.

Source code in pyjelly/serialize/sparql.py

def binding(
    self, binding: Mapping[Any, object | None], variables: Sequence[object]
) -> jelly.SparqlResultsFrame | None:
    """
    Add a row (solution) given as a mapping from variables to values.

    Args:
        binding (Mapping[Any, object | None]): values of the bound variables,
            keyed by variable or by variable name.
        variables (Sequence[object]): variables of the result set, in order.

    Returns:
        jelly.SparqlResultsFrame | None: the frame, if this row filled it.

    """
    values = self.values
    if len(variables) != len(values):
        msg = f"expected {len(values)} variables, got {len(variables)}"
        raise JellyConformanceError(msg)
    # A typed list is cheaper to index, see row()
    keys: list[object] = variables if type(variables) is list else list(variables)
    bound = 0
    if type(binding) is dict:
        # Same loop, but with direct dict calls instead of method lookups
        mapping: dict[Any, object | None] = binding
        for i in range(len(values)):
            value = mapping.get(keys[i])
            values[i] = value
            if value is not None:
                bound += 1
    else:
        for i in range(len(values)):
            value = binding.get(keys[i])
            values[i] = value
            if value is not None:
                bound += 1
    if bound != len(binding):
        # Some keys are not the variable objects themselves (rdflib's
        # Variable("x") does not equal "x"), so look the values up by name
        by_name = {str(key): value for key, value in binding.items()}
        for i in range(len(values)):
            values[i] = by_name.get(str(variables[i]))
    return self.row(values)

rows(rows)

Add many rows, see row().

Yields: jelly.SparqlResultsFrame: each frame as it fills up.

Source code in pyjelly/serialize/sparql.py

def rows(
    self, rows: Iterable[Sequence[object | None]]
) -> Iterator[jelly.SparqlResultsFrame]:
    """
    Add many rows, see row().

    Yields:
        jelly.SparqlResultsFrame: each frame as it fills up.

    """
    for values in rows:
        frame = self.row(values)
        if frame is not None:
            yield frame

add(column, term)

Add a value to a column. None is an unbound cell.

Source code in pyjelly/serialize/sparql.py

def add(self, column: VariableColumn, term: object | None) -> None:
    """Add a value to a column. None is an unbound cell."""
    if term is None:
        if not column.run_unbound:
            column.end_run()
            column.run_unbound = True
        column.run_length += 1
        return
    if (
        column.run_length
        and not column.run_unbound
        and self.encoder.same_term(column.run_term, term)
    ):
        column.run_length += 1
        return
    column.end_run()
    if self.triple_owner is not None and self.triple_owner is not column:
        self._switch_triple_iris(column)
    encoded = self.encoded
    self.encoder.encode_spo(term, encoded)
    column.add_value(encoded)
    column.run_term = term
    column.run_length = 1
    if encoded.triple_term is not None:
        self._triple_term_added(column)

end_frame()

End the current frame and return it.

Source code in pyjelly/serialize/sparql.py

def end_frame(self) -> jelly.SparqlResultsFrame:
    """End the current frame and return it."""
    if self.variables is None or self.failed:
        if self.failed:
            raise _failed()
        msg = "there is no result set in progress"
        raise JellyConformanceError(msg)
    frame = jelly.SparqlResultsFrame()
    self.write_frame(frame)
    return frame

write_frame(frame)

Fill in the frame: options, header, lookup entries and rows.

Source code in pyjelly/serialize/sparql.py

def write_frame(self, frame: jelly.SparqlResultsFrame) -> None:
    """Fill in the frame: options, header, lookup entries and rows."""
    self.write_start(frame)
    encoder = self.encoder
    row_count = self.row_count
    frame.row_count = row_count
    encoder.names.packed.write(frame.names)
    encoder.prefixes.packed.write(frame.prefixes)
    encoder.datatypes.packed.write(frame.datatypes)
    # Columns are left out of a frame with no rows
    if row_count:
        for column in self.columns:
            column.write(frame.columns.add())
    self.new_frame()

write_start(frame)

Add the stream options and the header, if they are still due.

Source code in pyjelly/serialize/sparql.py

def write_start(self, frame: jelly.SparqlResultsFrame) -> None:
    """Add the stream options and the header, if they are still due."""
    if self.options_pending:
        frame.options.CopyFrom(self.stream_options())
        self.options_pending = False
    if self.header_pending:
        assert self.variables is not None
        frame.variables.extend(self.variables)
        self.header_pending = False
        if self.links:
            frame.metadata[LINK_KEY] = links_value(self.links)

end(error='')

End the result set and return its last frame, which has the trailer.

Args: error (str): reason the result set is incomplete, if it is. It tells readers that some results are missing.

Returns: jelly.SparqlResultsFrame: the last frame.

Source code in pyjelly/serialize/sparql.py

def end(self, error: str = "") -> jelly.SparqlResultsFrame:
    """
    End the result set and return its last frame, which has the trailer.

    Args:
        error (str): reason the result set is incomplete, if it is. It tells
            readers that some results are missing.

    Returns:
        jelly.SparqlResultsFrame: the last frame.

    """
    frame = jelly.SparqlResultsFrame()
    if self.failed:
        if not error or self.variables is None:
            raise _failed()
        # Drop the frame's rows and their lookup entries. The lookups are then
        # out of sync with the reader, so the stream must end here.
        self.write_start(frame)
        self.ended = True
    elif self.variables is None:
        msg = "there is no result set to end, start one with start()"
        raise JellyConformanceError(msg)
    else:
        self.write_frame(frame)
        if not self.options.punctuated:
            self.ended = True
    frame.trailer.SetInParent()
    frame.trailer.error = error
    self.variables = None
    return frame

ask(value, links=(), error='')

Return the frame of a boolean result (of an ASK query).

Args: value (bool): the result. links (Sequence[str]): link IRIs for the result. error (str): reason the result is incomplete, if it is.

Returns: jelly.SparqlResultsFrame: the result's only frame.

Source code in pyjelly/serialize/sparql.py

def ask(
    self,
    value: bool,  # noqa: FBT001
    links: Sequence[str] = (),
    error: str = "",
) -> jelly.SparqlResultsFrame:
    """
    Return the frame of a boolean result (of an ASK query).

    Args:
        value (bool): the result.
        links (Sequence[str]): link IRIs for the result.
        error (str): reason the result is incomplete, if it is.

    Returns:
        jelly.SparqlResultsFrame: the result's only frame.

    """
    self._check_can_start()
    frame = jelly.SparqlResultsFrame()
    self.write_start(frame)
    frame.ask_result.value = value
    frame.trailer.SetInParent()
    frame.trailer.error = error
    if links:
        frame.metadata[LINK_KEY] = links_value(links)
    if not self.options.punctuated:
        self.ended = True
    return frame

FrameWriter(out, *, delimited)

Writer of frames to a file, either delimited or as a single frame.

Each frame is flushed, so a reader on the other end of a connection gets the rows right away.

Source code in pyjelly/serialize/sparql.py

def __init__(self, out: IO[bytes], *, delimited: bool) -> None:
    self.out = out
    self.delimited = delimited
    self.written = False

links_value(links)

Join link IRIs with LF, for the "link" metadata key.

links_value(["http://example.org/a", "http://example.org/b"]) b'http://example.org/a\nhttp://example.org/b'

Source code in pyjelly/serialize/sparql.py

def links_value(links: Sequence[str]) -> bytes:
    r"""
    Join link IRIs with LF, for the "link" metadata key.

    >>> links_value(["http://example.org/a", "http://example.org/b"])
    b'http://example.org/a\nhttp://example.org/b'
    """
    for link in links:
        if "\n" in link:
            msg = f"a link cannot contain the LF character, as no IRI can: {link!r}"
            raise JellyConformanceError(msg)
    return "\n".join(links).encode()

variable_names(variables)

Strip a leading "?" or "$" from each variable name.

variable_names(["?s", "$p", "o"]) ['s', 'p', 'o']

Source code in pyjelly/serialize/sparql.py

def variable_names(variables: Iterable[str]) -> list[str]:
    """
    Strip a leading "?" or "$" from each variable name.

    >>> variable_names(["?s", "$p", "o"])
    ['s', 'p', 'o']
    """
    names = []
    for variable in variables:
        name = str(variable)
        if name[:1] in ("?", "$"):
            name = name[1:]
        if not name:
            msg = "variable names cannot be empty"
            raise JellyConformanceError(msg)
        names.append(name)
    return names

error_message(error)

Format an exception as the error message of a trailer.

error_message(KeyError("x")) "KeyError: 'x'" error_message(JellyIncompleteResultsError("...", "timeout")) 'timeout'

Source code in pyjelly/serialize/sparql.py

def error_message(error: BaseException) -> str:
    """
    Format an exception as the error message of a trailer.

    >>> error_message(KeyError("x"))
    "KeyError: 'x'"
    >>> error_message(JellyIncompleteResultsError("...", "timeout"))
    'timeout'
    """
    if isinstance(error, JellyIncompleteResultsError):
        # Passing on the error of a stream that was itself incomplete
        return error.error
    message = str(error)
    name = type(error).__name__
    return f"{name}: {message}" if message else name

write_solutions(stream, variables, rows, writer, *, links=(), error='')

Write a solution sequence (the result of a SELECT query) as one result set.

If iterating the rows raises (e.g., the query fails midway), the result set is ended with that error so readers know it is incomplete, and the exception is re-raised. Pass error if the rows are already known to be incomplete.

Source code in pyjelly/serialize/sparql.py

def write_solutions(  # noqa: PLR0913 keyword-only options
    stream: SparqlStream,
    variables: Iterable[str],
    rows: Iterable[Sequence[object | None]],
    writer: FrameWriter,
    *,
    links: Sequence[str] = (),
    error: str = "",
) -> None:
    """
    Write a solution sequence (the result of a SELECT query) as one result set.

    If iterating the rows raises (e.g., the query fails midway), the result set is
    ended with that error so readers know it is incomplete, and the exception is
    re-raised. Pass `error` if the rows are already known to be incomplete.
    """
    stream.start(variables, links)
    try:
        for values in rows:
            frame = stream.row(values)
            if frame is not None:
                writer.write(frame)
    except Exception as failure:
        _end_with(stream, writer, failure)
        raise
    writer.write(stream.end(error))

write_bindings(stream, variables, bindings, writer, *, links=(), error='')

Like write_solutions(), with rows given as bindings.

Args: stream (SparqlStream): the stream. variables (Sequence[object]): the variables, as str objects (e.g., rdflib's Variable). Their names are written. bindings (Iterable[Mapping[Any, object | None]]): the rows, see SparqlStream.binding(). writer (FrameWriter): writer of the frames. links (Sequence[str]): link IRIs for the results. error (str): reason the rows are incomplete, if known.

Source code in pyjelly/serialize/sparql.py

def write_bindings(  # noqa: PLR0913 keyword-only options
    stream: SparqlStream,
    variables: Sequence[object],
    bindings: Iterable[Mapping[Any, object | None]],
    writer: FrameWriter,
    *,
    links: Sequence[str] = (),
    error: str = "",
) -> None:
    """
    Like write_solutions(), with rows given as bindings.

    Args:
        stream (SparqlStream): the stream.
        variables (Sequence[object]): the variables, as str objects (e.g., rdflib's
            Variable). Their names are written.
        bindings (Iterable[Mapping[Any, object | None]]): the rows, see
            SparqlStream.binding().
        writer (FrameWriter): writer of the frames.
        links (Sequence[str]): link IRIs for the results.
        error (str): reason the rows are incomplete, if known.

    """
    stream.start([str(variable) for variable in variables], links)
    try:
        for binding in bindings:
            frame = stream.binding(binding, variables)
            if frame is not None:
                writer.write(frame)
    except Exception as failure:
        _end_with(stream, writer, failure)
        raise
    writer.write(stream.end(error))

streams

Classes:

Name	Description
SerializerOptions	Options for writing a Jelly-RDF stream.
Stream
TripleStream
QuadStream

Functions:

Name	Description
stream_for_type	Give a Stream based on physical type specified.

SerializerOptions(flow=None, frame_size=DEFAULT_FRAME_SIZE, logical_type=jelly.LOGICAL_STREAM_TYPE_UNSPECIFIED, params=StreamParameters(), lookup_preset=LookupPreset())

Options for writing a Jelly-RDF stream.

Attributes: flow: how statements are split into frames. If None, it is chosen from the logical stream type. frame_size: target number of rows per frame. logical_type: logical stream type. If unspecified, it is guessed from the data. params: other stream parameters, e.g., the protocol version. lookup_preset: lookup sizes.

Stream(*, encoder, options=None)

Methods:

Name	Description
column_batch	Set the stream up to write Jelly-RDF 1.2, and return its column batch.
infer_flow	Return flow based on the stream options provided.
enroll	Initialize start of the stream.
stream_options	Encode and append stream options row to the current flow.
namespace_declaration	Add namespace declaration to jelly stream.
set_subject	Set the subject of the next statement.
set_predicate	Set the predicate of the next statement.
set_object	Set the object of the next statement.
emit	Add a statement made of the terms set with the cursor API.
encode_changed_spo	Encode the subject, predicate and object that were set since the last emit.
add_changed_spo	Add the subject, predicate and object of the cursor API to their columns.
write_changed_spo	Write the terms encoded by encode_changed_spo() into the statement.
for_rdflib	Initialize stream with RDFLib encoder.

Source code in pyjelly/serialize/streams.py

def __init__(
    self,
    *,
    encoder: TermEncoder,
    options: SerializerOptions | None = None,
) -> None:
    self.encoder = encoder
    if options is None:
        options = SerializerOptions()
    self.options = options
    flow = options.flow
    if flow is None:
        flow = self.infer_flow()
    self.flow = flow
    # In the row layout, the encoder adds rows for new lookup entries to the
    # same flow
    encoder.flow = flow
    self.repeated_terms = [None] * len(Slot)
    # Which terms the cursor API set since the last emit()
    self.subject_changed = False
    self.predicate_changed = False
    self.object_changed = False
    self.graph_changed = False
    self.enrolled = False
    self.stream_types = StreamTypes(
        physical_type=self.physical_type,
        logical_type=self.flow.logical_type,
    )
    #: Jelly-RDF 1.2: the statements of the frame being built
    self.batch: ColumnBatch | None = None
    params = options.params
    if not is_row_layout(params.version):
        self.batch = self.column_batch(params)

column_batch(params)

Set the stream up to write Jelly-RDF 1.2, and return its column batch.

Source code in pyjelly/serialize/streams.py

def column_batch(self, params: StreamParameters) -> ColumnBatch:
    """Set the stream up to write Jelly-RDF 1.2, and return its column batch."""
    max_names = self.options.lookup_preset.max_names
    if max_names < MIN_COLUMN_NAME_LOOKUP_SIZE:
        msg = (
            f"the name lookup size is {max_names}, but in Jelly-RDF 1.2 it must "
            f"be at least {MIN_COLUMN_NAME_LOOKUP_SIZE}"
        )
        raise JellyConformanceError(msg)
    encoder = self.encoder
    encoder.column_layout = True
    encoder.rdf_version = params.rdf_version
    stream_type = stream_type_for(params, self.stream_types.logical_type)
    batch = ColumnBatch(
        encoder,
        quads=self.physical_type == jelly.PHYSICAL_STREAM_TYPE_QUADS,
        messages=stream_type == jelly.RDF_STREAM_TYPE_MESSAGES,
    )
    self.flow.batch = batch
    return batch

infer_flow()

Return flow based on the stream options provided.

Returns: FrameFlow: initialised FrameFlow object.

Source code in pyjelly/serialize/streams.py

def infer_flow(self) -> FrameFlow:
    """
    Return flow based on the stream options provided.

    Returns:
        FrameFlow: initialised FrameFlow object.

    """
    flow: FrameFlow
    if self.options.params.delimited:
        if self.options.logical_type != jelly.LOGICAL_STREAM_TYPE_UNSPECIFIED:
            flow_class = flow_for_type(self.options.logical_type)
        else:
            flow_class = self.default_delimited_flow_class

        if issubclass(flow_class, BoundedFrameFlow) and (
            self.options.logical_type
            in (
                jelly.LOGICAL_STREAM_TYPE_FLAT_TRIPLES,
                jelly.LOGICAL_STREAM_TYPE_FLAT_QUADS,
            )
        ):
            flow = flow_class(
                logical_type=self.options.logical_type,
                frame_size=self.options.frame_size,
            )
        else:
            flow = flow_class(logical_type=self.options.logical_type)
    else:
        flow = ManualFrameFlow(logical_type=self.options.logical_type)
    return flow

enroll()

Initialize start of the stream.

Source code in pyjelly/serialize/streams.py

def enroll(self) -> None:
    """Initialize start of the stream."""
    if not self.enrolled:
        self.stream_options()
        self.enrolled = True

stream_options()

Encode and append stream options row to the current flow.

Source code in pyjelly/serialize/streams.py

def stream_options(self) -> None:
    """Encode and append stream options row to the current flow."""
    self.flow.new_row().options.CopyFrom(
        encode_options(
            stream_types=self.stream_types,
            params=self.options.params,
            lookup_preset=self.options.lookup_preset,
        )
    )

namespace_declaration(name, iri)

Add namespace declaration to jelly stream.

Args: name (str): namespace prefix label iri (str): namespace iri

Returns: jelly.RdfStreamFrame | None: the frame before, if the declaration has to start a new one

Source code in pyjelly/serialize/streams.py

def namespace_declaration(self, name: str, iri: str) -> jelly.RdfStreamFrame | None:
    """
    Add namespace declaration to jelly stream.

    Args:
        name (str): namespace prefix label
        iri (str): namespace iri

    Returns:
        jelly.RdfStreamFrame | None: the frame before, if the declaration has
            to start a new one

    """
    batch = self.batch
    if batch is None:
        encode_namespace_declaration(
            name=name, value=iri, term_encoder=self.encoder
        )
        return None
    # In Jelly-RDF 1.2, the declarations of a frame come before its statements
    frame = None
    if batch.row_count or not batch.has_room():
        frame = self.flow.to_stream_frame()
    batch.add_namespace(name, iri)
    return frame

set_subject(s)

Set the subject of the next statement.

Source code in pyjelly/serialize/streams.py

def set_subject(self, s: object) -> None:
    """Set the subject of the next statement."""
    self.repeated_terms[0] = s
    self.subject_changed = True

set_predicate(p)

Set the predicate of the next statement.

Source code in pyjelly/serialize/streams.py

def set_predicate(self, p: object) -> None:
    """Set the predicate of the next statement."""
    self.repeated_terms[1] = p
    self.predicate_changed = True

set_object(o)

Set the object of the next statement.

Source code in pyjelly/serialize/streams.py

def set_object(self, o: object) -> None:
    """Set the object of the next statement."""
    self.repeated_terms[2] = o
    self.object_changed = True

emit()

Add a statement made of the terms set with the cursor API.

Returns: jelly.RdfStreamFrame | None: stream frame if flow supports frames slicing and current flow is full

Source code in pyjelly/serialize/streams.py

def emit(self) -> jelly.RdfStreamFrame | None:
    """
    Add a statement made of the terms set with the cursor API.

    Returns:
        jelly.RdfStreamFrame | None: stream frame if
            flow supports frames slicing and current flow is full

    """
    raise NotImplementedError

encode_changed_spo()

Encode the subject, predicate and object that were set since the last emit.

They are encoded in this order no matter in which order they were set: IRIs are encoded relative to the IRI before them, in the order the parser reads them. Rows for new lookup entries are added to the flow.

Returns: int: which of them were encoded, as bits 0 (subject) to 2 (object).

Source code in pyjelly/serialize/streams.py

def encode_changed_spo(self) -> int:
    """
    Encode the subject, predicate and object that were set since the last emit.

    They are encoded in this order no matter in which order they were set:
    IRIs are encoded relative to the IRI before them, in the order the parser
    reads them. Rows for new lookup entries are added to the flow.

    Returns:
        int: which of them were encoded, as bits 0 (subject) to 2 (object).

    """
    terms = self.repeated_terms
    encoded = self.encoder.encoded
    changed = 0
    if self.subject_changed:
        self.encoder.encode_spo(terms[0], encoded[0])
        self.subject_changed = False
        changed = 1
    if self.predicate_changed:
        self.encoder.encode_spo(terms[1], encoded[1])
        self.predicate_changed = False
        changed |= 2
    if self.object_changed:
        self.encoder.encode_spo(terms[2], encoded[2])
        self.object_changed = False
        changed |= 4
    return changed

add_changed_spo(batch)

Add the subject, predicate and object of the cursor API to their columns.

Source code in pyjelly/serialize/streams.py

def add_changed_spo(self, batch: ColumnBatch) -> None:
    """Add the subject, predicate and object of the cursor API to their columns."""
    terms = self.repeated_terms
    batch.add_changed(batch.subjects, terms[0], changed=self.subject_changed)
    batch.add_changed(batch.predicates, terms[1], changed=self.predicate_changed)
    batch.add_changed(batch.objects, terms[2], changed=self.object_changed)
    self.subject_changed = False
    self.predicate_changed = False
    self.object_changed = False

write_changed_spo(statement, changed)

Write the terms encoded by encode_changed_spo() into the statement.

Source code in pyjelly/serialize/streams.py

def write_changed_spo(self, statement: Statement, changed: int) -> None:
    """Write the terms encoded by encode_changed_spo() into the statement."""
    encoded = self.encoder.encoded
    if not changed:
        # Every term is repeated: the row is an empty statement
        statement.SetInParent()
    if changed & 1:
        write_term(encoded[0], statement, 0)
    if changed & 2:
        write_term(encoded[1], statement, 1)
    if changed & 4:
        write_term(encoded[2], statement, 2)

for_rdflib(options=None)

Initialize stream with RDFLib encoder.

Args: options (SerializerOptions | None, optional): Stream options. Defaults to None.

Raises: TypeError: if Stream is passed, and not a Stream for specific physical type.

Returns: Stream: initialized stream with RDFLib encoder.

Source code in pyjelly/serialize/streams.py

@classmethod
def for_rdflib(cls, options: SerializerOptions | None = None) -> Stream:
    """
    Initialize stream with RDFLib encoder.

    Args:
        options (SerializerOptions | None, optional): Stream options.
            Defaults to None.

    Raises:
        TypeError: if Stream is passed, and not a Stream for specific physical type.

    Returns:
        Stream: initialized stream with RDFLib encoder.

    """
    if cls is Stream:
        msg = "Stream is an abstract base class, use a subclass instead"
        raise TypeError(msg)
    from pyjelly.integrations.rdflib.encode import (  # noqa: PLC0415
        RDFLibTermEncoder,
    )

    lookup_preset: LookupPreset | None = None
    if options is not None:
        lookup_preset = options.lookup_preset
    return cls(
        encoder=RDFLibTermEncoder(lookup_preset=lookup_preset),
        options=options,
    )

TripleStream(*, encoder, options=None)

Bases: Stream

Methods:

Name	Description
triple	Process one triple to Protobuf messages.
triples	Process triples, see triple().
emit	Add a triple made of the terms set with the cursor API.

Source code in pyjelly/serialize/streams.py

def __init__(
    self,
    *,
    encoder: TermEncoder,
    options: SerializerOptions | None = None,
) -> None:
    self.encoder = encoder
    if options is None:
        options = SerializerOptions()
    self.options = options
    flow = options.flow
    if flow is None:
        flow = self.infer_flow()
    self.flow = flow
    # In the row layout, the encoder adds rows for new lookup entries to the
    # same flow
    encoder.flow = flow
    self.repeated_terms = [None] * len(Slot)
    # Which terms the cursor API set since the last emit()
    self.subject_changed = False
    self.predicate_changed = False
    self.object_changed = False
    self.graph_changed = False
    self.enrolled = False
    self.stream_types = StreamTypes(
        physical_type=self.physical_type,
        logical_type=self.flow.logical_type,
    )
    #: Jelly-RDF 1.2: the statements of the frame being built
    self.batch: ColumnBatch | None = None
    params = options.params
    if not is_row_layout(params.version):
        self.batch = self.column_batch(params)

triple(terms)

Process one triple to Protobuf messages.

Note: Adds new rows to the current flow and returns StreamFrame if frame size conditions are met.

Args: terms (Iterable[object]): RDF terms to encode.

Returns: jelly.RdfStreamFrame | None: stream frame if flow supports frames slicing and current flow is full

Source code in pyjelly/serialize/streams.py

def triple(self, terms: Iterable[object]) -> jelly.RdfStreamFrame | None:
    """
    Process one triple to Protobuf messages.

    Note:
        Adds new rows to the current flow and returns StreamFrame if
        frame size conditions are met.

    Args:
        terms (Iterable[object]): RDF terms to encode.

    Returns:
        jelly.RdfStreamFrame | None: stream frame if
            flow supports frames slicing and current flow is full

    """
    batch = self.batch
    if batch is None:
        encode_triple(terms, self.encoder, self.repeated_terms)
    else:
        batch.add_triple(terms, self.repeated_terms)
    return self.flow.frame_from_bounds()

triples(statements)

Process triples, see triple().

Yields: jelly.RdfStreamFrame: full frames.

Source code in pyjelly/serialize/streams.py

def triples(
    self, statements: Iterable[Iterable[object]]
) -> Iterator[jelly.RdfStreamFrame]:
    """
    Process triples, see triple().

    Yields:
        jelly.RdfStreamFrame: full frames.

    """
    for terms in statements:
        frame = self.triple(terms)
        if frame is not None:
            yield frame

emit()

Add a triple made of the terms set with the cursor API.

Returns: jelly.RdfStreamFrame | None: stream frame if flow supports frames slicing and current flow is full

Source code in pyjelly/serialize/streams.py

def emit(self) -> jelly.RdfStreamFrame | None:
    """
    Add a triple made of the terms set with the cursor API.

    Returns:
        jelly.RdfStreamFrame | None: stream frame if
            flow supports frames slicing and current flow is full

    """
    batch = self.batch
    if batch is not None:
        self.add_changed_spo(batch)
        batch.row_count += 1
        return self.flow.frame_from_bounds()
    self.encoder.new_epoch()
    # The terms first, as they add rows for new lookup entries
    changed = self.encode_changed_spo()
    self.write_changed_spo(self.flow.new_row().triple, changed)
    return self.flow.frame_from_bounds()

QuadStream(*, encoder, options=None)

Bases: Stream

Methods:

Name	Description
quad	Process one quad to Protobuf messages.
quads	Process quads, see quad().
set_graph	Set the graph name of the next quad.
emit	Add a quad made of the terms set with the cursor API.

Source code in pyjelly/serialize/streams.py

def __init__(
    self,
    *,
    encoder: TermEncoder,
    options: SerializerOptions | None = None,
) -> None:
    self.encoder = encoder
    if options is None:
        options = SerializerOptions()
    self.options = options
    flow = options.flow
    if flow is None:
        flow = self.infer_flow()
    self.flow = flow
    # In the row layout, the encoder adds rows for new lookup entries to the
    # same flow
    encoder.flow = flow
    self.repeated_terms = [None] * len(Slot)
    # Which terms the cursor API set since the last emit()
    self.subject_changed = False
    self.predicate_changed = False
    self.object_changed = False
    self.graph_changed = False
    self.enrolled = False
    self.stream_types = StreamTypes(
        physical_type=self.physical_type,
        logical_type=self.flow.logical_type,
    )
    #: Jelly-RDF 1.2: the statements of the frame being built
    self.batch: ColumnBatch | None = None
    params = options.params
    if not is_row_layout(params.version):
        self.batch = self.column_batch(params)

quad(terms)

Process one quad to Protobuf messages.

Args: terms (Iterable[object]): terms to encode.

Returns: jelly.RdfStreamFrame | None: stream frame if flow supports frames slicing and current flow is full

Source code in pyjelly/serialize/streams.py

def quad(self, terms: Iterable[object]) -> jelly.RdfStreamFrame | None:
    """
    Process one quad to Protobuf messages.

    Args:
        terms (Iterable[object]): terms to encode.

    Returns:
        jelly.RdfStreamFrame | None: stream frame if
            flow supports frames slicing and current flow is full

    """
    batch = self.batch
    if batch is None:
        encode_quad(terms, self.encoder, self.repeated_terms)
    else:
        batch.add_quad(terms, self.repeated_terms)
    return self.flow.frame_from_bounds()

quads(statements)

Process quads, see quad().

Yields: jelly.RdfStreamFrame: full frames.

Source code in pyjelly/serialize/streams.py

def quads(
    self, statements: Iterable[Iterable[object]]
) -> Iterator[jelly.RdfStreamFrame]:
    """
    Process quads, see quad().

    Yields:
        jelly.RdfStreamFrame: full frames.

    """
    for terms in statements:
        frame = self.quad(terms)
        if frame is not None:
            yield frame

set_graph(g)

Set the graph name of the next quad.

Source code in pyjelly/serialize/streams.py

def set_graph(self, g: object) -> None:
    """Set the graph name of the next quad."""
    self.repeated_terms[3] = g
    self.graph_changed = True

emit()

Add a quad made of the terms set with the cursor API.

Returns: jelly.RdfStreamFrame | None: stream frame if flow supports frames slicing and current flow is full

Source code in pyjelly/serialize/streams.py

def emit(self) -> jelly.RdfStreamFrame | None:
    """
    Add a quad made of the terms set with the cursor API.

    Returns:
        jelly.RdfStreamFrame | None: stream frame if
            flow supports frames slicing and current flow is full

    """
    batch = self.batch
    if batch is not None:
        self.add_changed_spo(batch)
        batch.add_graph_changed(self.repeated_terms[3], changed=self.graph_changed)
        self.graph_changed = False
        batch.row_count += 1
        return self.flow.frame_from_bounds()
    self.encoder.new_epoch()
    # The terms first, as they add rows for new lookup entries. The graph
    # comes last, as in the parser.
    changed = self.encode_changed_spo()
    graph_changed = self.graph_changed
    if graph_changed:
        self.encoder.encode_graph(self.repeated_terms[3], self.encoder.encoded[3])
        self.graph_changed = False
        changed |= 8
    quad = self.flow.new_row().quad
    self.write_changed_spo(quad, changed)
    if graph_changed:
        write_graph(self.encoder.encoded[3], quad)
    return self.flow.frame_from_bounds()

stream_for_type(physical_type)

Give a Stream based on physical type specified.

Args: physical_type (jelly.PhysicalStreamType): jelly stream physical type.

Raises: NotImplementedError: if no stream for requested physical type is available.

Returns: type[Stream]: jelly stream

Source code in pyjelly/serialize/streams.py

def stream_for_type(physical_type: jelly.PhysicalStreamType) -> type[Stream]:
    """
    Give a Stream based on physical type specified.

    Args:
        physical_type (jelly.PhysicalStreamType): jelly stream physical type.

    Raises:
        NotImplementedError: if no stream for requested physical type is available.

    Returns:
        type[Stream]: jelly stream

    """
    try:
        stream_cls = STREAM_DISPATCH[physical_type]
    except KeyError:
        if physical_type == jelly.PHYSICAL_STREAM_TYPE_GRAPHS:
            msg = (
                "writing streams of the GRAPHS physical type is not supported "
                "(they can still be read). Use QUADS instead."
            )
        else:
            msg = (
                "no stream class for physical type "
                f"{jelly.PhysicalStreamType.Name(physical_type)}"
            )
        raise NotImplementedError(msg) from None
    return stream_cls