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.

`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.

`integrations`

Modules:

Name	Description
`generic`
`rdflib`

`generic`

Modules:

Name	Description
`generic_sink`
`parse`
`serialize`

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

BlankNode(identifier)

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

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

def __init__(self, identifier: str) -> None:
    self._identifier: str = identifier

IRI(iri)

Class for IRIs, storing IRI as a string.

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

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

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

Class for literals.

Notes: Consists of: lexical form, and optional language tag and datatype. All parts of literal are stored as strings.

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

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

Triple

Bases: NamedTuple

Class for RDF triples.

Quad

Bases: NamedTuple

Class for RDF quads.

Prefix

Bases: NamedTuple

Class for generic namespace declaration.

GenericStatementSink(identifier=DefaultGraph)

Notes: _store preserves the order of statements.

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

Attributes:

Name	Type	Description
`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] = {}
    self._identifier = identifier

is_triples_sink

Check if the sink contains triples or quads.

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

`parse`

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.

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.

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/parse.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/parse.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/parse.py

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

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)
    return

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)
    return

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 parse_triples_stream(
            frames=frames,
            options=options,
            **{"frame_metadata": frame_metadata} if frame_metadata is not None else {},
        ):
            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 parse_quads_stream(
            frames=frames,
            options=options,
            **{"frame_metadata": frame_metadata} if frame_metadata is not None else {},
        ):
            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:
        for triples in parse_triples_stream(frames=frames, options=options):
            yield from triples
        return
    if options.stream_types.physical_type in (
        jelly.PHYSICAL_STREAM_TYPE_QUADS,
        jelly.PHYSICAL_STREAM_TYPE_GRAPHS,
    ):
        for quads in parse_quads_stream(frames=frames, options=options):
            yield from quads
        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`

Modules:

Name	Description
`jelly`	Generated protocol buffer code.

Classes:

Name	Description
`BlankNode`	Class for blank nodes, storing BN's identifier as a string.
`GenericSinkTermEncoder`
`GenericStatementSink`
`GraphStream`
`IRI`	Class for IRIs, storing IRI as a string.
`Literal`	Class for literals.
`Quad`	Class for RDF quads.
`QuadStream`
`SerializerOptions`
`Stream`
`TermEncoder`
`Triple`	Class for RDF triples.
`TripleStream`

Functions:

Name	Description
`stream_frames`

Attributes:

Name	Type	Description
`HasGraph`		Represent a PEP 604 union type
`QUAD_ARITY`		int([x]) -> integer
`Statement`		Represent a PEP 604 union type
`__file__`		str(object='') -> str
`__name__`		str(object='') -> str
`__package__`		str(object='') -> str

HasGraph = rdf_pb2.RdfQuad | rdf_pb2.RdfGraphStart

Represent a PEP 604 union type

E.g. for int | str

QUAD_ARITY = 4

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

Statement = rdf_pb2.RdfQuad | rdf_pb2.RdfTriple

Represent a PEP 604 union type

E.g. for int | str

__file__ = '/home/runner/work/pyjelly/pyjelly/pyjelly/integrations/generic/serialize.cpython-310-x86_64-linux-gnu.so'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__name__ = 'pyjelly.integrations.generic.serialize'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__package__ = 'pyjelly.integrations.generic'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

BlankNode(identifier)

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

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

def __init__(self, identifier: str) -> None:
    self._identifier: str = identifier

GenericSinkTermEncoder(lookup_preset=None)

Bases: pyjelly.serialize.encode.TermEncoder

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

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)

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.integrations.generic.serialize'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('lookup_preset', 'names', 'prefixes', 'datatypes')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

GenericStatementSink(identifier=DefaultGraph)

Notes: _store preserves the order of statements.

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

Attributes:

Name	Type	Description
`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] = {}
    self._identifier = identifier

is_triples_sink

Check if the sink contains triples or quads.

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

GraphStream(*args, **kwargs)

Bases: pyjelly.serialize.streams.TripleStream

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.
`physical_type`		int([x]) -> integer

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

physical_type = 3

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

IRI(iri)

Class for IRIs, storing IRI as a string.

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

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

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

Class for literals.

Notes: Consists of: lexical form, and optional language tag and datatype. All parts of literal are stored as strings.

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

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

Quad

Bases: NamedTuple

Class for RDF quads.

QuadStream(*args, **kwargs)

Bases: pyjelly.serialize.streams.Stream

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.
`physical_type`		int([x]) -> integer

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

physical_type = 2

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

SerializerOptions(flow=None, frame_size=250, logical_type=0, params=<factory>, lookup_preset=<factory>)

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__annotations__`		dict() -> new empty dictionary
`__dataclass_fields__`		dict() -> new empty dictionary
`__doc__`		str(object='') -> str
`__match_args__`		Built-in immutable sequence.
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

__annotations__ = {'flow': <class 'type'>, 'frame_size': <class 'int'>, 'logical_type': <class 'type'>, 'params': <class 'pyjelly.options.StreamParameters'>, 'lookup_preset': <class 'pyjelly.options.LookupPreset'>}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__dataclass_fields__ = {'flow': Field(name='flow',type=<class 'type'>,default=None,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'frame_size': Field(name='frame_size',type=<class 'int'>,default=250,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'logical_type': Field(name='logical_type',type=<class 'type'>,default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'params': Field(name='params',type=<class 'pyjelly.options.StreamParameters'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<class 'pyjelly.options.StreamParameters'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'lookup_preset': Field(name='lookup_preset',type=<class 'pyjelly.options.LookupPreset'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<class 'pyjelly.options.LookupPreset'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__match_args__ = ('flow', 'frame_size', 'logical_type', 'params', 'lookup_preset')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('flow', 'frame_size', 'logical_type', 'params', 'lookup_preset')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

Stream(*args, **kwargs)

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

TermEncoder(lookup_preset=None)

Methods:

Name	Description
`encode_iri_indices`	Encode lookup indices for IRI.
`encode_iri`	Encode iri.
`encode_default_graph`	Encode default graph.
`encode_literal`	Encode literal.
`encode_quoted_triple`	Encode a quoted triple.
`get_iri_field`	Get IRI field directly based on slot.
`get_literal_field`	Get literal field directly based on slot.
`set_bnode_field`	Set bnode field directly based on slot.
`get_triple_field`	Get triple term field directly based on slot.

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)

encode_iri_indices(iri_string)

Encode lookup indices for IRI.

Args: iri_string (str): full iri in string format.

Returns: tuple[Rows, int, int]: additional rows (if any) and indices in prefix and name tables.

Source code in pyjelly/serialize/encode.py

def encode_iri_indices(self, iri_string: str) -> tuple[Rows, int, int]:
    """
    Encode lookup indices for IRI.

    Args:
        iri_string (str): full iri in string format.

    Returns:
        tuple[Rows, int, int]: additional rows (if any) and
            indices in prefix and name tables.

    """
    prefix, name = split_iri(iri_string)
    if self.prefixes.lookup.max_size:
        prefix_entry_index = self.prefixes.encode_entry_index(prefix)
    else:
        name = iri_string
        prefix_entry_index = None

    name_entry_index = self.names.encode_entry_index(name)
    term_rows = []

    if prefix_entry_index is not None:
        prefix_entry = jelly.RdfPrefixEntry(id=prefix_entry_index, value=prefix)
        term_rows.append(jelly.RdfStreamRow(prefix=prefix_entry))

    if name_entry_index is not None:
        name_entry = jelly.RdfNameEntry(id=name_entry_index, value=name)
        term_rows.append(jelly.RdfStreamRow(name=name_entry))

    prefix_index = self.prefixes.encode_prefix_term_index(prefix)
    name_index = self.names.encode_name_term_index(name)
    return term_rows, prefix_index, name_index

encode_iri(iri_string, iri)

Encode iri.

Args: iri_string (str): full iri in string format. iri (jelly.RdfIri): iri to fill

Returns: Rows: extra rows for prefix and name tables, if any.

Source code in pyjelly/serialize/encode.py

def encode_iri(self, iri_string: str, iri: jelly.RdfIri) -> Rows:
    """
    Encode iri.

    Args:
        iri_string (str): full iri in string format.
        iri (jelly.RdfIri): iri to fill

    Returns:
        Rows: extra rows for prefix and name tables, if any.

    """
    term_rows, prefix_index, name_index = self.encode_iri_indices(iri_string)
    iri.prefix_id = prefix_index
    iri.name_id = name_index
    return term_rows

encode_default_graph(g_default_graph)

Encode default graph.

Returns: Rows: empty extra rows (for API consistency)

Source code in pyjelly/serialize/encode.py

def encode_default_graph(self, g_default_graph: jelly.RdfDefaultGraph) -> Rows:
    """
    Encode default graph.

    Returns:
        Rows: empty extra rows (for API consistency)

    """
    g_default_graph.CopyFrom(jelly.RdfDefaultGraph())
    return ()

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

Encode literal.

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. literal (jelly.RdfLiteral): literal to fill.

Raises: JellyConformanceError: if datatype specified while datatable is not used.

Returns: Rows: extra rows (i.e., datatype entries).

Source code in pyjelly/serialize/encode.py

def encode_literal(
    self,
    *,
    lex: str,
    language: str | None = None,
    datatype: str | None = None,
    literal: jelly.RdfLiteral,
) -> Rows:
    """
    Encode literal.

    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.
        literal (jelly.RdfLiteral): literal to fill.

    Raises:
        JellyConformanceError: if datatype specified while
            datatable is not used.

    Returns:
        Rows: extra rows (i.e., datatype entries).

    """
    datatype_id = None
    term_rows: tuple[()] | tuple[jelly.RdfStreamRow] = ()

    if datatype and datatype != options.STRING_DATATYPE_IRI:
        if self.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)
        datatype_entry_id = self.datatypes.encode_entry_index(datatype)

        if datatype_entry_id is not None:
            entry = jelly.RdfDatatypeEntry(id=datatype_entry_id, value=datatype)
            term_rows = (jelly.RdfStreamRow(datatype=entry),)

        datatype_id = self.datatypes.encode_datatype_term_index(datatype)

    literal.lex = lex
    if language:
        literal.langtag = language
    if datatype_id:
        literal.datatype = datatype_id
    return term_rows

encode_quoted_triple(terms, quoted_statement)

Encode a quoted triple.

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. quoted_statement (jelly.RdfTriple): quoted triple to fill.

Returns: Rows: additional stream rows with preceeding information (prefixes, names, datatypes rows, if any).

Source code in pyjelly/serialize/encode.py

def encode_quoted_triple(
    self, terms: Iterable[object], quoted_statement: jelly.RdfTriple
) -> Rows:
    """
    Encode a quoted triple.

    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.
        quoted_statement (jelly.RdfTriple): quoted triple to fill.

    Returns:
        Rows: additional stream rows with preceeding
            information (prefixes, names, datatypes rows, if any).

    """
    rows: list[jelly.RdfStreamRow] = []
    terms = iter(terms)
    extra_rows = self.encode_spo(next(terms), Slot.subject, quoted_statement)
    rows.extend(extra_rows)
    extra_rows = self.encode_spo(next(terms), Slot.predicate, quoted_statement)
    rows.extend(extra_rows)
    extra_rows = self.encode_spo(next(terms), Slot.object, quoted_statement)
    rows.extend(extra_rows)
    return rows

get_iri_field(statement, slot)

Get IRI field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_iri_field(self, statement: Statement, slot: Slot) -> jelly.RdfIri:
    """Get IRI field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_iri
    if slot == Slot.predicate:
        return statement.p_iri
    return statement.o_iri

get_literal_field(statement, slot)

Get literal field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_literal_field(self, statement: Statement, slot: Slot) -> jelly.RdfLiteral:
    """Get literal field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_literal
    if slot == Slot.predicate:
        return statement.p_literal
    return statement.o_literal

set_bnode_field(statement, slot, identifier)

Set bnode field directly based on slot.

Source code in pyjelly/serialize/encode.py

def set_bnode_field(
    self, statement: Statement, slot: Slot, identifier: str
) -> None:
    """Set bnode field directly based on slot."""
    if slot == Slot.subject:
        statement.s_bnode = identifier
    elif slot == Slot.predicate:
        statement.p_bnode = identifier
    else:
        statement.o_bnode = identifier

get_triple_field(statement, slot)

Get triple term field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_triple_field(self, statement: Statement, slot: Slot) -> jelly.RdfTriple:
    """Get triple term field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_triple_term
    if slot == Slot.predicate:
        return statement.p_triple_term
    return statement.o_triple_term

Triple

Bases: NamedTuple

Class for RDF triples.

TripleStream(*args, **kwargs)

Bases: pyjelly.serialize.streams.Stream

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.
`physical_type`		int([x]) -> integer

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

physical_type = 1

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

stream_frames()

jelly

Generated protocol buffer code.

`rdflib`

Modules:

Name	Description
`parse`
`serialize`

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.

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"

`parse`

Classes:

Name	Description
`Triple`	Describe RDFLib triple.
`Quad`	Describe RDFLib quad.
`Prefix`	Describe RDF Prefix(i.e, namespace declaration).
`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.
`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.

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.

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.

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).

RDFLibAdapter(options, parsing_mode=ParsingMode.FLAT)

Bases: Adapter

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

Args: Adapter (): abstract 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

RDFLibTriplesAdapter(options)

Bases: RDFLibAdapter

Triples adapter RDFLib implementation.

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

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

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

RDFLibQuadsAdapter(options)

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/parse.py

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

RDFLibGraphsAdapter(options)

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/parse.py

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

RDFLibJellyParser

Bases: Parser

Methods:

Name	Description
`parse`	Parse jelly file into provided RDFLib Graph.

parse(source, sink)

Parse jelly file into provided RDFLib Graph.

Args: source (InputSource): jelly file as buffered binary stream InputSource obj sink (Graph): RDFLib Graph

Raises: TypeError: raises error if invalid input

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

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

    Args:
        source (InputSource): jelly file as buffered binary stream InputSource obj
        sink (Graph): RDFLib Graph

    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)
    parse_jelly_to_graph(
        inp,
        graph_factory=lambda: Graph(store=sink.store, identifier=sink.identifier),
        dataset_factory=lambda: Dataset(store=sink.store),
    )

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/rdflib/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 = RDFLibTriplesAdapter(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)
    return

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/rdflib/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[RDFLibQuadsBaseAdapter]
    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_QUADS:
        adapter_class = RDFLibQuadsAdapter
    else:
        adapter_class = RDFLibGraphsAdapter
    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)
    return

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

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

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(
    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,
) -> 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



    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 parse_triples_stream(
            frames=frames,
            options=options,
            frame_metadata=frame_metadata,
        ):
            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 parse_quads_stream(
            frames=frames, options=options, frame_metadata=frame_metadata
        ):
            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)

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.

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,
) -> 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.

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

    """
    options, frames = get_options_and_frames(inp)

    if options.stream_types.physical_type == jelly.PHYSICAL_STREAM_TYPE_TRIPLES:
        sink = graph_factory()
    if options.stream_types.physical_type in (
        jelly.PHYSICAL_STREAM_TYPE_QUADS,
        jelly.PHYSICAL_STREAM_TYPE_GRAPHS,
    ):
        quad_sink = dataset_factory()
        sink = quad_sink

    for item in parse_jelly_flat(inp=inp, frames=frames, options=options):
        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
            context = quad_sink.get_context(graph_name)
            quad_sink.add((s, p, o, context))
    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/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,
) -> 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.

    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:
        for triples in parse_triples_stream(frames=frames, options=options):
            yield from triples
        return
    if options.stream_types.physical_type in (
        jelly.PHYSICAL_STREAM_TYPE_QUADS,
        jelly.PHYSICAL_STREAM_TYPE_GRAPHS,
    ):
        for quads in parse_quads_stream(frames=frames, options=options):
            yield from quads
        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
`RDFLibTermEncoder`
`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.
`graphs_stream_frames`	Serialize a Dataset into jelly frames as a stream of graphs.
`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.

RDFLibTermEncoder(lookup_preset=None)

Bases: TermEncoder

Methods:

Name	Description
`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)

encode_spo(term, slot, statement)

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

Args: term (object): term to encode slot (Slot): its place in statement. statement (Statement): Triple/Quad message to fill with s/p/o terms.

Returns: Rows: encoded extra rows

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

def encode_spo(self, term: object, slot: Slot, statement: Statement) -> Rows:
    """
    Encode s/p/o term based on its RDFLib object.

    Args:
        term (object): term to encode
        slot (Slot): its place in statement.
        statement (Statement): Triple/Quad message to fill with s/p/o terms.

    Returns:
        Rows: encoded extra rows

    """
    if isinstance(term, rdflib.URIRef):
        iri = self.get_iri_field(statement, slot)
        return self.encode_iri(term, iri)

    if isinstance(term, rdflib.Literal):
        literal = self.get_literal_field(statement, slot)
        return self.encode_literal(
            lex=str(term),
            language=term.language,
            # `datatype` is cast to `str` explicitly because
            # `URIRef.__eq__` overrides `str.__eq__` in an incompatible manner
            datatype=term.datatype and str(term.datatype),
            literal=literal,
        )

    if isinstance(term, rdflib.BNode):
        self.set_bnode_field(statement, slot, str(term))
        return ()

    return super().encode_spo(term, slot, statement)  # error if not handled

encode_graph(term, statement)

Encode graph name term based on its RDFLib object.

Args: term (object): term to encode statement (HasGraph): Quad/GraphStart message to fill g_{} in.

Returns: Rows: encoded extra rows

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

def encode_graph(self, term: object, statement: HasGraph) -> Rows:
    """
    Encode graph name term based on its RDFLib object.

    Args:
        term (object): term to encode
        statement (HasGraph): Quad/GraphStart message to fill g_{} in.

    Returns:
        Rows: encoded extra rows

    """
    if term == DATASET_DEFAULT_GRAPH_ID:
        return self.encode_default_graph(statement.g_default_graph)

    if isinstance(term, rdflib.URIRef):
        return self.encode_iri(term, statement.g_iri)

    if isinstance(term, rdflib.BNode):
        statement.g_bnode = str(term)
        return ()
    return super().encode_graph(term, statement)  # error if not handled

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)
    write = write_delimited if stream.options.params.delimited else write_single
    for stream_frame in stream_frames(stream, self.store):
        write(stream_frame, out)

triples_stream_frames(stream, data)

Serialize a Graph/Dataset into jelly frames.

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

Notes: if Dataset 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 | Generator[Triple],
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a Graph/Dataset into jelly frames.

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

    Notes:
        if Dataset 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:
        namespace_declarations(data, stream)

    graphs = (data,) if not isinstance(data, Dataset) else data.graphs()
    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 Dataset into jelly frames.

Notes: Emits one frame per dataset if flow is of DatasetsFrameFlow.

Args: stream (QuadStream): stream that specifies quads processing data (Dataset | Generator[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 | Generator[Quad],
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a Dataset into jelly frames.

    Notes:
        Emits one frame per dataset if flow is of DatasetsFrameFlow.

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

    Yields:
        Generator[jelly.RdfStreamFrame]: jelly frames

    """
    stream.enroll()
    if stream.options.params.namespace_declarations:
        namespace_declarations(data, stream)  # type: ignore[arg-type]
    iterator: Generator[Quad, None, None]
    if isinstance(data, Dataset):
        iterator = cast(Generator[Quad, None, None], data.quads())
    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

graphs_stream_frames(stream, data)

Serialize a Dataset into jelly frames as a stream of graphs.

Notes: If flow of DatasetsFrameFlow type, the whole dataset will be encoded into one frame.

Args: stream (GraphStream): stream that specifies graphs processing data (Dataset | Generator[Quad]): Dataset to serialize.

Yields: Generator[jelly.RdfStreamFrame]: jelly frames

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

@stream_frames.register(GraphStream)
def graphs_stream_frames(
    stream: GraphStream,
    data: Dataset | Generator[Quad],
) -> Generator[jelly.RdfStreamFrame]:
    """
    Serialize a Dataset into jelly frames as a stream of graphs.

    Notes:
        If flow of DatasetsFrameFlow type, the whole dataset
        will be encoded into one frame.

    Args:
        stream (GraphStream): stream that specifies graphs processing
        data (Dataset | Generator[Quad]): Dataset to serialize.

    Yields:
        Generator[jelly.RdfStreamFrame]: jelly frames

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

    if isinstance(data, Dataset):
        graphs = data.graphs()
    else:
        ds = Dataset()
        for quad in data:
            ctx = ds.get_context(quad.g)
            ctx.add((quad.s, quad.p, quad.o))
        graphs = ds.graphs()

    for graph in graphs:
        yield from stream.graph(graph_id=graph.identifier, graph=graph)

    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)

Guess the serializer options based on the store type.

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

def guess_options(sink: Graph | Dataset) -> SerializerOptions:
    """
    Guess the serializer options based on the store type.

    >>> guess_options(Graph()).logical_type
    1
    >>> guess_options(Dataset()).logical_type
    2
    """
    logical_type = (
        jelly.LOGICAL_STREAM_TYPE_FLAT_QUADS
        if isinstance(sink, Dataset)
        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 is given, initializes TripleStream

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 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, Dataset
    ):
        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)
            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)

    combined: Generator[Triple | Quad] | Graph = (
        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 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)

`jelly`

Modules:

Name	Description
`rdf_pb2`	Generated protocol buffer code.

`rdf_pb2`

Generated protocol buffer code.

`options`

Classes:

Name	Description
`StreamTypes`

Functions:

Name	Description
`register_mimetypes`	Associate files that have Jelly extension with Jelly MIME types.

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.

`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.

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})"

`register_mimetypes(extension='.jelly')`

Associate files that have Jelly extension with Jelly MIME types.

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)

`parse`

Modules:

Name	Description
`decode`
`ioutils`
`lookup`

`decode`

Classes:

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

Functions:

Name	Description
`options_from_frame`	Fill stream options based on the options row.

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

`Decoder(adapter)`

Initializes decoder with a lookup tables with preset sizes, integration-dependent adapter and empty repeated terms dictionary.

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

Methods:

Name	Description
`iter_rows`	Iterate through rows in the frame.
`decode_row`	Decode a row based on its type.
`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.
`decode_bnode`	Decode string message to blank node (BN) using a custom adapter.
`decode_literal`	Decode RdfLiteral to literal based on custom adapter implementation.
`decode_statement`	Decode a triple/quad message.

Source code in pyjelly/parse/decode.py

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

    Initializes decoder with a lookup tables with preset sizes,
    integration-dependent adapter and empty repeated terms dictionary.

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

    """
    self.adapter = adapter
    self.names = LookupDecoder(lookup_size=self.options.lookup_preset.max_names)
    self.prefixes = LookupDecoder(
        lookup_size=self.options.lookup_preset.max_prefixes
    )
    self.datatypes = LookupDecoder(
        lookup_size=self.options.lookup_preset.max_datatypes
    )
    self.repeated_terms: dict[str, jelly.RdfIri | str | jelly.RdfLiteral] = {}

    self.row_handlers: dict[type[Any], RowHandler] = {
        t: getattr(self, name) for t, name in self._ROW_HANDLER_NAMES.items()
    }
    self.term_handlers: dict[type[Any], TermHandler] = {
        t: getattr(self, name) for t, name in self._TERM_HANDLER_NAMES.items()
    }

iter_rows(frame)

Iterate through rows in the frame.

Args: frame (jelly.RdfStreamFrame): jelly frame Yields: Iterator[Any]: decoded rows

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
    Yields:
        Iterator[Any]: decoded rows

    """
    for row_owner in frame.rows:
        row = getattr(row_owner, row_owner.WhichOneof("row"))
        decoded_row = self.decode_row(row)
        if isinstance(
            row, (jelly.RdfTriple, jelly.RdfQuad, jelly.RdfNamespaceDeclaration)
        ):
            yield decoded_row

decode_row(row)

Decode a row based on its type.

Notes: uses custom adapters to decode triples/quads, namespace declarations, graph start/end.

Args: row (Any): protobuf row message

Raises: TypeError: raises error if this type of protobuf message does not have a respective handler

Returns: Any | None: decoded row - result from calling decode_row (row type appropriate handler)

Source code in pyjelly/parse/decode.py

def decode_row(self, row: Any) -> Any | None:
    """
    Decode a row based on its type.

    Notes: uses custom adapters to decode triples/quads, namespace declarations,
           graph start/end.

    Args:
        row (Any): protobuf row message

    Raises:
        TypeError: raises error if this type of protobuf message does not have
                   a respective handler

    Returns:
        Any | None: decoded row -
                    result from calling decode_row (row type appropriate handler)

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

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.

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.

    Args:
        iri (jelly.RdfIri): RdfIri message

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

    """
    name = self.names.decode_name_term_index(iri.name_id)
    prefix = self.prefixes.decode_prefix_term_index(iri.prefix_id)
    return self.adapter.iri(iri=prefix + name)

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.

Notes: checks for langtag existence; for datatype checks for non-zero table size and datatype field presence

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.

    Notes: checks for langtag existence;
           for datatype checks for non-zero table size and datatype field presence

    Args:
        literal (jelly.RdfLiteral): RdfLiteral message

    Returns:
        Any: literal returned by the custom adapter

    """
    language = datatype = None
    if literal.langtag:
        language = literal.langtag
    elif self.datatypes.lookup_size and literal.HasField("datatype"):
        datatype = self.datatypes.decode_datatype_term_index(literal.datatype)
    return self.adapter.literal(
        lex=literal.lex,
        language=language,
        datatype=datatype,
    )

decode_statement(statement, oneofs)

Decode a triple/quad message.

Notes: also updates repeated terms dictionary

Args: statement (jelly.RdfTriple | jelly.RdfQuad): triple/quad message oneofs (Sequence[str]): terms s/p/o/g(if quads)

Raises: ValueError: 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,
    statement: jelly.RdfTriple | jelly.RdfQuad,
    oneofs: Sequence[str],
) -> Any:
    """
    Decode a triple/quad message.

    Notes: also updates repeated terms dictionary

    Args:
        statement (jelly.RdfTriple | jelly.RdfQuad): triple/quad message
        oneofs (Sequence[str]): terms s/p/o/g(if quads)

    Raises:
        ValueError: if a missing repeated term is encountered

    Returns:
        Any: a list of decoded terms

    """
    terms = []
    for oneof in oneofs:
        field = statement.WhichOneof(oneof)
        if field:
            jelly_term = getattr(statement, field)
            decoded_term = self.decode_term(jelly_term)
            self.repeated_terms[oneof] = decoded_term
        else:
            decoded_term = self.repeated_terms[oneof]
            if decoded_term is None:
                msg = f"missing repeated term {oneof}"
                raise ValueError(msg)
        terms.append(decoded_term)
    return terms

`options_from_frame(frame, *, delimited)`

Fill stream options based on the options row.

Notes: generalized_statements, rdf_star, and namespace declarations are set to false by default

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

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.

    Notes:
        generalized_statements, rdf_star, and namespace declarations
        are set to false by default

    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

    """
    row = frame.rows[0]
    options = row.options
    nd = getattr(options, "namespace_declarations", False) or (
        options.version >= MAX_VERSION
    )
    return ParserOptions(
        stream_types=StreamTypes(
            physical_type=options.physical_type,
            logical_type=options.logical_type,
        ),
        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=nd,
        ),
    )

`ioutils`

Functions:

Name	Description
`delimited_jelly_hint`	Detect whether a Jelly file is delimited from its first 3 bytes.
`get_options_and_frames`	Return stream options and frames from the buffered binary stream.

`delimited_jelly_hint(header)`

Detect whether a Jelly file is delimited from its first 3 bytes.

Truth table (notation: 0A = 0x0A, NN = not 0x0A, ?? = don't care):

Byte 1	Byte 2	Byte 3	Result
`NN`	`??`	`??`	Delimited
`0A`	`NN`	`??`	Non-delimited
`0A`	`0A`	`NN`	Delimited (size = 10)
`0A`	`0A`	`0A`	Non-delimited (stream options size = 10)

Source code in pyjelly/parse/ioutils.py

def delimited_jelly_hint(header: bytes) -> bool:
    """
    Detect whether a Jelly file is delimited from its first 3 bytes.

    Truth table (notation: `0A` = `0x0A`, `NN` = `not 0x0A`, `??` = _don't care_):

    | Byte 1 | Byte 2 | Byte 3 | Result                                   |
    |--------|--------|--------|------------------------------------------|
    | `NN`   |  `??`  |  `??`  | Delimited                                |
    | `0A`   |  `NN`  |  `??`  | Non-delimited                            |
    | `0A`   |  `0A`  |  `NN`  | Delimited (size = 10)                    |
    | `0A`   |  `0A`  |  `0A`  | Non-delimited (stream options size = 10) |

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

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

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

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

    >>> delimited_jelly_hint(bytes([0x0A, 0x00, 0x00]))
    False

    >>> delimited_jelly_hint(bytes([0x0A, 0x00, 0x0A]))
    False

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

    >>> delimited_jelly_hint(bytes([0x0A, 0x0A, 0x0A]))
    False
    """
    magic = 0x0A
    return len(header) >= 3 and (  # noqa: PLR2004
        header[0] != magic or (header[1] == magic and header[2] != magic)
    )

`get_options_and_frames(inp)`

Return stream options and frames from the buffered binary stream.

Args: inp (IO[bytes]): jelly buffered binary stream

Raises: JellyConformanceError: if no non-empty frames detected in the delimited stream JellyConformanceError: if non-delimited, error is raised if no rows are detected (empty frame)

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.

    Args:
        inp (IO[bytes]): jelly buffered binary stream

    Raises:
        JellyConformanceError: if no non-empty frames detected in the delimited stream
        JellyConformanceError: if non-delimited,
            error is raised if no rows are detected (empty frame)

    Returns:
        tuple[ParserOptions, Iterator[jelly.RdfStreamFrame]]: ParserOptions holds:
            stream types, lookup presets and other stream options

    """
    if not inp.seekable():
        # Input may not be seekable (e.g. a network stream) -- then we need to buffer
        # it to determine if it's delimited.
        # See also: https://github.com/Jelly-RDF/pyjelly/issues/298
        inp = io.BufferedReader(inp)  # type: ignore[arg-type, type-var, unused-ignore]
        is_delimited = delimited_jelly_hint(inp.peek(3))
    else:
        is_delimited = delimited_jelly_hint(bytes_read := inp.read(3))
        inp.seek(-len(bytes_read), os.SEEK_CUR)

    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:
            msg = "No non-empty frames found in the stream"
            raise JellyConformanceError(msg)

        options = options_from_frame(first_frame, delimited=True)
        return options, chain(skipped_frames, (first_frame,), frames)

    frame = parse(jelly.RdfStreamFrame, inp.read())

    if not frame.rows:
        msg = "The stream is corrupted (only contains an empty frame)"
        raise JellyConformanceError(msg)

    options = options_from_frame(frame, delimited=False)
    return options, iter((frame,))

`lookup`

Classes:

Name	Description
`JellyAssertionError`	Raised when a recommended assertion from the specification fails.
`JellyConformanceError`	Raised when Jelly conformance is violated.
`LookupDecoder`

Attributes:

Name	Type	Description
`MAX_LOOKUP_SIZE`		int([x]) -> integer
`__file__`		str(object='') -> str
`__name__`		str(object='') -> str
`__package__`		str(object='') -> str

`MAX_LOOKUP_SIZE = 4096`

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

`file = '/home/runner/work/pyjelly/pyjelly/.venv/lib/python3.10/site-packages/pyjelly/parse/lookup.cpython-310-x86_64-linux-gnu.so'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`name = 'pyjelly.parse.lookup'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`package = 'pyjelly.parse'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`JellyAssertionError`

Bases: AssertionError

Raised when a recommended assertion from the specification fails.

`JellyConformanceError`

Bases: Exception

Raised when Jelly conformance is violated.

`LookupDecoder(*args, **kwargs)`

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__annotations__`		dict() -> new empty dictionary
`__dataclass_fields__`		dict() -> new empty dictionary
`__doc__`		str(object='') -> str
`__match_args__`		Built-in immutable sequence.
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

__annotations__ = {'last_assigned_index': <class 'int'>, 'last_reused_index': <class 'int'>}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__dataclass_fields__ = {'last_assigned_index': Field(name='last_assigned_index',type=<class 'int'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'last_reused_index': Field(name='last_reused_index',type=<class 'int'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__match_args__ = ('last_assigned_index', 'last_reused_index')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__module__ = 'pyjelly.parse.lookup'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('last_assigned_index', 'last_reused_index', 'lookup_size', 'data')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

`serialize`

Modules:

Name	Description
`encode`
`flows`
`lookup`
`streams`

`encode`

Classes:

Name	Description
`TermEncoder`

Functions:

Name	Description
`split_iri`	Split iri into prefix and name.
`encode_spo`	Encode the s/p/o of a statement.
`encode_triple`	Encode one triple.
`encode_quad`	Encode one quad.
`encode_namespace_declaration`	Encode namespace declaration.
`encode_options`	Encode stream options to ProtoBuf message.

`TermEncoder(lookup_preset=None)`

Methods:

Name	Description
`encode_iri_indices`	Encode lookup indices for IRI.
`encode_iri`	Encode iri.
`encode_default_graph`	Encode default graph.
`encode_literal`	Encode literal.
`encode_quoted_triple`	Encode a quoted triple.
`get_iri_field`	Get IRI field directly based on slot.
`get_literal_field`	Get literal field directly based on slot.
`set_bnode_field`	Set bnode field directly based on slot.
`get_triple_field`	Get triple term field directly based on slot.

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)

encode_iri_indices(iri_string)

Encode lookup indices for IRI.

Args: iri_string (str): full iri in string format.

Returns: tuple[Rows, int, int]: additional rows (if any) and indices in prefix and name tables.

Source code in pyjelly/serialize/encode.py

def encode_iri_indices(self, iri_string: str) -> tuple[Rows, int, int]:
    """
    Encode lookup indices for IRI.

    Args:
        iri_string (str): full iri in string format.

    Returns:
        tuple[Rows, int, int]: additional rows (if any) and
            indices in prefix and name tables.

    """
    prefix, name = split_iri(iri_string)
    if self.prefixes.lookup.max_size:
        prefix_entry_index = self.prefixes.encode_entry_index(prefix)
    else:
        name = iri_string
        prefix_entry_index = None

    name_entry_index = self.names.encode_entry_index(name)
    term_rows = []

    if prefix_entry_index is not None:
        prefix_entry = jelly.RdfPrefixEntry(id=prefix_entry_index, value=prefix)
        term_rows.append(jelly.RdfStreamRow(prefix=prefix_entry))

    if name_entry_index is not None:
        name_entry = jelly.RdfNameEntry(id=name_entry_index, value=name)
        term_rows.append(jelly.RdfStreamRow(name=name_entry))

    prefix_index = self.prefixes.encode_prefix_term_index(prefix)
    name_index = self.names.encode_name_term_index(name)
    return term_rows, prefix_index, name_index

encode_iri(iri_string, iri)

Encode iri.

Args: iri_string (str): full iri in string format. iri (jelly.RdfIri): iri to fill

Returns: Rows: extra rows for prefix and name tables, if any.

Source code in pyjelly/serialize/encode.py

def encode_iri(self, iri_string: str, iri: jelly.RdfIri) -> Rows:
    """
    Encode iri.

    Args:
        iri_string (str): full iri in string format.
        iri (jelly.RdfIri): iri to fill

    Returns:
        Rows: extra rows for prefix and name tables, if any.

    """
    term_rows, prefix_index, name_index = self.encode_iri_indices(iri_string)
    iri.prefix_id = prefix_index
    iri.name_id = name_index
    return term_rows

encode_default_graph(g_default_graph)

Encode default graph.

Returns: Rows: empty extra rows (for API consistency)

Source code in pyjelly/serialize/encode.py

def encode_default_graph(self, g_default_graph: jelly.RdfDefaultGraph) -> Rows:
    """
    Encode default graph.

    Returns:
        Rows: empty extra rows (for API consistency)

    """
    g_default_graph.CopyFrom(jelly.RdfDefaultGraph())
    return ()

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

Encode literal.

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. literal (jelly.RdfLiteral): literal to fill.

Raises: JellyConformanceError: if datatype specified while datatable is not used.

Returns: Rows: extra rows (i.e., datatype entries).

Source code in pyjelly/serialize/encode.py

def encode_literal(
    self,
    *,
    lex: str,
    language: str | None = None,
    datatype: str | None = None,
    literal: jelly.RdfLiteral,
) -> Rows:
    """
    Encode literal.

    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.
        literal (jelly.RdfLiteral): literal to fill.

    Raises:
        JellyConformanceError: if datatype specified while
            datatable is not used.

    Returns:
        Rows: extra rows (i.e., datatype entries).

    """
    datatype_id = None
    term_rows: tuple[()] | tuple[jelly.RdfStreamRow] = ()

    if datatype and datatype != options.STRING_DATATYPE_IRI:
        if self.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)
        datatype_entry_id = self.datatypes.encode_entry_index(datatype)

        if datatype_entry_id is not None:
            entry = jelly.RdfDatatypeEntry(id=datatype_entry_id, value=datatype)
            term_rows = (jelly.RdfStreamRow(datatype=entry),)

        datatype_id = self.datatypes.encode_datatype_term_index(datatype)

    literal.lex = lex
    if language:
        literal.langtag = language
    if datatype_id:
        literal.datatype = datatype_id
    return term_rows

encode_quoted_triple(terms, quoted_statement)

Encode a quoted triple.

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. quoted_statement (jelly.RdfTriple): quoted triple to fill.

Returns: Rows: additional stream rows with preceeding information (prefixes, names, datatypes rows, if any).

Source code in pyjelly/serialize/encode.py

def encode_quoted_triple(
    self, terms: Iterable[object], quoted_statement: jelly.RdfTriple
) -> Rows:
    """
    Encode a quoted triple.

    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.
        quoted_statement (jelly.RdfTriple): quoted triple to fill.

    Returns:
        Rows: additional stream rows with preceeding
            information (prefixes, names, datatypes rows, if any).

    """
    rows: list[jelly.RdfStreamRow] = []
    terms = iter(terms)
    extra_rows = self.encode_spo(next(terms), Slot.subject, quoted_statement)
    rows.extend(extra_rows)
    extra_rows = self.encode_spo(next(terms), Slot.predicate, quoted_statement)
    rows.extend(extra_rows)
    extra_rows = self.encode_spo(next(terms), Slot.object, quoted_statement)
    rows.extend(extra_rows)
    return rows

get_iri_field(statement, slot)

Get IRI field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_iri_field(self, statement: Statement, slot: Slot) -> jelly.RdfIri:
    """Get IRI field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_iri
    if slot == Slot.predicate:
        return statement.p_iri
    return statement.o_iri

get_literal_field(statement, slot)

Get literal field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_literal_field(self, statement: Statement, slot: Slot) -> jelly.RdfLiteral:
    """Get literal field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_literal
    if slot == Slot.predicate:
        return statement.p_literal
    return statement.o_literal

set_bnode_field(statement, slot, identifier)

Set bnode field directly based on slot.

Source code in pyjelly/serialize/encode.py

def set_bnode_field(
    self, statement: Statement, slot: Slot, identifier: str
) -> None:
    """Set bnode field directly based on slot."""
    if slot == Slot.subject:
        statement.s_bnode = identifier
    elif slot == Slot.predicate:
        statement.p_bnode = identifier
    else:
        statement.o_bnode = identifier

get_triple_field(statement, slot)

Get triple term field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_triple_field(self, statement: Statement, slot: Slot) -> jelly.RdfTriple:
    """Get triple term field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_triple_term
    if slot == Slot.predicate:
        return statement.p_triple_term
    return statement.o_triple_term

`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.

    """
    name = iri_string
    prefix = ""
    for sep in "#", "/":
        prefix, char, name = iri_string.rpartition(sep)
        if char:
            return prefix + char, name
    return prefix, name

`encode_spo(terms, term_encoder, repeated_terms, statement)`

Encode the s/p/o of a statement.

Args: terms (Iterator[object]): iterator for original terms to encode term_encoder (TermEncoder): encoder with lookup tables repeated_terms (list[object | None): list of repeated terms. statement (Statement): Triple/Quad to fill.

Returns: list[jelly.RdfStreamRow] extra rows to append.

Source code in pyjelly/serialize/encode.py

def encode_spo(
    terms: Iterator[object],
    term_encoder: TermEncoder,
    repeated_terms: list[object | None],
    statement: Statement,
) -> list[jelly.RdfStreamRow]:
    """
    Encode the s/p/o of a statement.

    Args:
        terms (Iterator[object]): iterator for original terms to encode
        term_encoder (TermEncoder): encoder with lookup tables
        repeated_terms (list[object | None): list of repeated terms.
        statement (Statement): Triple/Quad to fill.

    Returns:
        list[jelly.RdfStreamRow] extra rows to append.

    """
    rows: list[jelly.RdfStreamRow] = []
    s = next(terms)
    if repeated_terms[Slot.subject] != s:
        extra_rows = term_encoder.encode_spo(s, Slot.subject, statement)
        rows.extend(extra_rows)
        repeated_terms[Slot.subject] = s
    p = next(terms)
    if repeated_terms[Slot.predicate] != p:
        extra_rows = term_encoder.encode_spo(p, Slot.predicate, statement)
        rows.extend(extra_rows)
        repeated_terms[Slot.predicate] = p
    o = next(terms)
    if repeated_terms[Slot.object] != o:
        extra_rows = term_encoder.encode_spo(o, Slot.object, statement)
        rows.extend(extra_rows)
        repeated_terms[Slot.object] = o
    return rows

`encode_triple(terms, term_encoder, repeated_terms)`

Encode one triple.

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.

Returns: list[jelly.RdfStreamRow]: list of rows to add to the current flow.

Source code in pyjelly/serialize/encode.py

def encode_triple(
    terms: Iterable[object],
    term_encoder: TermEncoder,
    repeated_terms: list[object | None],
) -> list[jelly.RdfStreamRow]:
    """
    Encode one triple.

    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.

    Returns:
        list[jelly.RdfStreamRow]: list of rows to add to the current flow.

    """
    triple = jelly.RdfTriple()
    terms = iter(terms)
    rows = encode_spo(terms, term_encoder, repeated_terms, triple)
    row = jelly.RdfStreamRow(triple=triple)
    rows.append(row)
    return rows

`encode_quad(terms, term_encoder, repeated_terms)`

Encode one quad.

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.

Returns: list[jelly.RdfStreamRow]: list of messages to append to current flow.

Source code in pyjelly/serialize/encode.py

def encode_quad(
    terms: Iterable[object],
    term_encoder: TermEncoder,
    repeated_terms: list[object | None],
) -> list[jelly.RdfStreamRow]:
    """
    Encode one quad.

    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.

    Returns:
        list[jelly.RdfStreamRow]: list of messages to append to current flow.

    """
    terms = iter(terms)
    quad = jelly.RdfQuad()
    rows = encode_spo(terms, term_encoder, repeated_terms, quad)
    g = next(terms)
    if repeated_terms[Slot.graph] != g:
        extra_rows = term_encoder.encode_graph(g, quad)
        rows.extend(extra_rows)
        repeated_terms[Slot.graph] = g
    row = jelly.RdfStreamRow(quad=quad)
    rows.append(row)
    return rows

`encode_namespace_declaration(name, value, term_encoder)`

Encode namespace declaration.

Args: name (str): namespace prefix label value (str): namespace iri term_encoder (TermEncoder): current encoder

Returns: list[jelly.RdfStreamRow]: list of messages to append to current flow.

Source code in pyjelly/serialize/encode.py

def encode_namespace_declaration(
    name: str,
    value: str,
    term_encoder: TermEncoder,
) -> list[jelly.RdfStreamRow]:
    """
    Encode namespace declaration.

    Args:
        name (str): namespace prefix label
        value (str): namespace iri
        term_encoder (TermEncoder): current encoder

    Returns:
        list[jelly.RdfStreamRow]: list of messages to append to current flow.

    """
    iri = jelly.RdfIri()
    [*rows] = term_encoder.encode_iri(value, iri=iri)
    declaration = jelly.RdfNamespaceDeclaration(name=name, value=iri)
    row = jelly.RdfStreamRow(namespace=declaration)
    rows.append(row)
    return rows

`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.RdfStreamRow: encoded stream options row

Source code in pyjelly/serialize/encode.py

def encode_options(
    lookup_preset: options.LookupPreset,
    stream_types: options.StreamTypes,
    params: options.StreamParameters,
) -> jelly.RdfStreamRow:
    """
    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.RdfStreamRow: encoded stream options row

    """
    return jelly.RdfStreamRow(
        options=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,
        )
    )

`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(initlist=None, *, logical_type=None, **__kwargs)`

Bases: UserList[RdfStreamRow]

Abstract base class for producing Jelly frames from RDF stream rows.

Collects stream rows and assembles them into RdfStreamFrame objects when ready.

Allows for passing LogicalStreamType, required for logical subtypes and non-delimited streams.

Methods:

Name	Description
`frame_from_graph`	Treat the current rows as a graph and produce a frame.
`frame_from_dataset`	Treat the current rows as a dataset and produce a frame.
`to_stream_frame`	Create stream frame from flow content.

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
    **__kwargs: Any,
) -> None:
    super().__init__(initlist)
    self.logical_type = logical_type or self.__class__.logical_type

frame_from_graph()

Treat the current rows as a graph and produce a frame.

Default implementation returns None.

Source code in pyjelly/serialize/flows.py

def frame_from_graph(self) -> jelly.RdfStreamFrame | None:
    """
    Treat the current rows as a graph and produce a frame.

    Default implementation returns None.
    """
    return None

frame_from_dataset()

Treat the current rows as a dataset and produce a frame.

Default implementation returns None.

Source code in pyjelly/serialize/flows.py

def frame_from_dataset(self) -> jelly.RdfStreamFrame | None:
    """
    Treat the current rows as a dataset and produce a frame.

    Default implementation returns None.
    """
    return None

to_stream_frame()

Create stream frame from flow content.

Notes: Clears flow content after creating the frame.

Returns: jelly.RdfStreamFrame | None: stream frame

Source code in pyjelly/serialize/flows.py

def to_stream_frame(self) -> jelly.RdfStreamFrame | None:
    """
    Create stream frame from flow content.

    Notes:
        Clears flow content after creating the frame.

    Returns:
        jelly.RdfStreamFrame | None: stream frame

    """
    if not self:
        return None
    frame = jelly.RdfStreamFrame(rows=self)
    self.clear()
    return frame

`ManualFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

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,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
    **__kwargs: Any,
) -> None:
    super().__init__(initlist)
    self.logical_type = logical_type or self.__class__.logical_type

`BoundedFrameFlow(initlist=None, logical_type=None, *, frame_size=None)`

Bases: FrameFlow

Produce frames automatically when a fixed number of rows is reached.

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

@override
def __init__(
    self,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    logical_type: jelly.LogicalStreamType | None = None,
    *,
    frame_size: int | None = None,
) -> None:
    super().__init__(initlist, 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

    """
    if len(self) >= self.frame_size:
        return self.to_stream_frame()
    return None

`GraphsFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

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,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
    **__kwargs: Any,
) -> None:
    super().__init__(initlist)
    self.logical_type = logical_type or self.__class__.logical_type

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.

    """
    return self.to_stream_frame()

`DatasetsFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

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,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
    **__kwargs: Any,
) -> None:
    super().__init__(initlist)
    self.logical_type = logical_type or self.__class__.logical_type

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.

    """
    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

`lookup`

Classes:

Name	Description
`Lookup`
`LookupEncoder`

Attributes:

Name	Type	Description
`__file__`		str(object='') -> str
`__name__`		str(object='') -> str
`__package__`		str(object='') -> str

`file = '/home/runner/work/pyjelly/pyjelly/.venv/lib/python3.10/site-packages/pyjelly/serialize/lookup.cpython-310-x86_64-linux-gnu.so'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`name = 'pyjelly.serialize.lookup'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`package = 'pyjelly.serialize'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`Lookup(*args, **kwargs)`

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.
`__repr__`	Return repr(self).

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.lookup'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('data', 'max_size', '_evicting')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

__repr__()

Return repr(self).

`LookupEncoder(*args, **kwargs)`

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__annotations__`		dict() -> new empty dictionary
`__dataclass_fields__`		dict() -> new empty dictionary
`__doc__`		str(object='') -> str
`__match_args__`		Built-in immutable sequence.
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

__annotations__ = {'last_assigned_index': <class 'int'>, 'last_reused_index': <class 'int'>}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__dataclass_fields__ = {'last_assigned_index': Field(name='last_assigned_index',type=<class 'int'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'last_reused_index': Field(name='last_reused_index',type=<class 'int'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__match_args__ = ('last_assigned_index', 'last_reused_index')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__module__ = 'pyjelly.serialize.lookup'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('last_assigned_index', 'last_reused_index', 'lookup')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

`streams`

Modules:

Name	Description
`jelly`	Generated protocol buffer code.

Classes:

Name	Description
`BoundedFrameFlow`	Produce frames automatically when a fixed number of rows is reached.
`FrameFlow`	Abstract base class for producing Jelly frames from RDF stream rows.
`GraphStream`
`ManualFrameFlow`	Produces frames only when manually requested (never automatically).
`QuadStream`
`SerializerOptions`
`Stream`
`StreamTypes`
`TermEncoder`
`TripleStream`

Functions:

Name	Description
`encode_namespace_declaration`	Encode namespace declaration.
`encode_options`	Encode stream options to ProtoBuf message.
`encode_quad`	Encode one quad.
`encode_triple`	Encode one triple.
`flow_for_type`	Return flow based on logical type requested.

Attributes:

Name	Type	Description
`DEFAULT_FRAME_SIZE`		int([x]) -> integer
`STREAM_DISPATCH`		dict() -> new empty dictionary
`TYPE_CHECKING`		bool(x) -> bool
`__file__`		str(object='') -> str
`__name__`		str(object='') -> str
`__package__`		str(object='') -> str

`DEFAULT_FRAME_SIZE = 250`

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

`STREAM_DISPATCH = {1: <class 'pyjelly.serialize.streams.TripleStream'>, 2: <class 'pyjelly.serialize.streams.QuadStream'>, 3: <class 'pyjelly.serialize.streams.GraphStream'>}`

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`TYPE_CHECKING = False`

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

`file = '/home/runner/work/pyjelly/pyjelly/pyjelly/serialize/streams.cpython-310-x86_64-linux-gnu.so'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`name = 'pyjelly.serialize.streams'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`package = 'pyjelly.serialize'`

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`BoundedFrameFlow(initlist=None, logical_type=None, *, frame_size=None)`

Bases: FrameFlow

Produce frames automatically when a fixed number of rows is reached.

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

@override
def __init__(
    self,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    logical_type: jelly.LogicalStreamType | None = None,
    *,
    frame_size: int | None = None,
) -> None:
    super().__init__(initlist, 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

    """
    if len(self) >= self.frame_size:
        return self.to_stream_frame()
    return None

`FrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

Bases: UserList[RdfStreamRow]

Abstract base class for producing Jelly frames from RDF stream rows.

Collects stream rows and assembles them into RdfStreamFrame objects when ready.

Allows for passing LogicalStreamType, required for logical subtypes and non-delimited streams.

Methods:

Name	Description
`frame_from_graph`	Treat the current rows as a graph and produce a frame.
`frame_from_dataset`	Treat the current rows as a dataset and produce a frame.
`to_stream_frame`	Create stream frame from flow content.

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
    **__kwargs: Any,
) -> None:
    super().__init__(initlist)
    self.logical_type = logical_type or self.__class__.logical_type

frame_from_graph()

Treat the current rows as a graph and produce a frame.

Default implementation returns None.

Source code in pyjelly/serialize/flows.py

def frame_from_graph(self) -> jelly.RdfStreamFrame | None:
    """
    Treat the current rows as a graph and produce a frame.

    Default implementation returns None.
    """
    return None

frame_from_dataset()

Treat the current rows as a dataset and produce a frame.

Default implementation returns None.

Source code in pyjelly/serialize/flows.py

def frame_from_dataset(self) -> jelly.RdfStreamFrame | None:
    """
    Treat the current rows as a dataset and produce a frame.

    Default implementation returns None.
    """
    return None

to_stream_frame()

Create stream frame from flow content.

Notes: Clears flow content after creating the frame.

Returns: jelly.RdfStreamFrame | None: stream frame

Source code in pyjelly/serialize/flows.py

def to_stream_frame(self) -> jelly.RdfStreamFrame | None:
    """
    Create stream frame from flow content.

    Notes:
        Clears flow content after creating the frame.

    Returns:
        jelly.RdfStreamFrame | None: stream frame

    """
    if not self:
        return None
    frame = jelly.RdfStreamFrame(rows=self)
    self.clear()
    return frame

`GraphStream(*args, **kwargs)`

Bases: pyjelly.serialize.streams.TripleStream

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.
`physical_type`		int([x]) -> integer

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

physical_type = 3

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

`ManualFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

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,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    *,
    logical_type: jelly.LogicalStreamType | None = None,
    **__kwargs: Any,
) -> None:
    super().__init__(initlist)
    self.logical_type = logical_type or self.__class__.logical_type

`QuadStream(*args, **kwargs)`

Bases: pyjelly.serialize.streams.Stream

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.
`physical_type`		int([x]) -> integer

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

physical_type = 2

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

`SerializerOptions(flow=None, frame_size=250, logical_type=0, params=<factory>, lookup_preset=<factory>)`

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__annotations__`		dict() -> new empty dictionary
`__dataclass_fields__`		dict() -> new empty dictionary
`__doc__`		str(object='') -> str
`__match_args__`		Built-in immutable sequence.
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

__annotations__ = {'flow': <class 'type'>, 'frame_size': <class 'int'>, 'logical_type': <class 'type'>, 'params': <class 'pyjelly.options.StreamParameters'>, 'lookup_preset': <class 'pyjelly.options.LookupPreset'>}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__dataclass_fields__ = {'flow': Field(name='flow',type=<class 'type'>,default=None,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'frame_size': Field(name='frame_size',type=<class 'int'>,default=250,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'logical_type': Field(name='logical_type',type=<class 'type'>,default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'params': Field(name='params',type=<class 'pyjelly.options.StreamParameters'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<class 'pyjelly.options.StreamParameters'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'lookup_preset': Field(name='lookup_preset',type=<class 'pyjelly.options.LookupPreset'>,default=<dataclasses._MISSING_TYPE object at 0x7fbb2cd56cb0>,default_factory=<class 'pyjelly.options.LookupPreset'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__match_args__ = ('flow', 'frame_size', 'logical_type', 'params', 'lookup_preset')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('flow', 'frame_size', 'logical_type', 'params', 'lookup_preset')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

`Stream(*args, **kwargs)`

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

`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.

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})"

`TermEncoder(lookup_preset=None)`

Methods:

Name	Description
`encode_iri_indices`	Encode lookup indices for IRI.
`encode_iri`	Encode iri.
`encode_default_graph`	Encode default graph.
`encode_literal`	Encode literal.
`encode_quoted_triple`	Encode a quoted triple.
`get_iri_field`	Get IRI field directly based on slot.
`get_literal_field`	Get literal field directly based on slot.
`set_bnode_field`	Set bnode field directly based on slot.
`get_triple_field`	Get triple term field directly based on slot.

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)

encode_iri_indices(iri_string)

Encode lookup indices for IRI.

Args: iri_string (str): full iri in string format.

Returns: tuple[Rows, int, int]: additional rows (if any) and indices in prefix and name tables.

Source code in pyjelly/serialize/encode.py

def encode_iri_indices(self, iri_string: str) -> tuple[Rows, int, int]:
    """
    Encode lookup indices for IRI.

    Args:
        iri_string (str): full iri in string format.

    Returns:
        tuple[Rows, int, int]: additional rows (if any) and
            indices in prefix and name tables.

    """
    prefix, name = split_iri(iri_string)
    if self.prefixes.lookup.max_size:
        prefix_entry_index = self.prefixes.encode_entry_index(prefix)
    else:
        name = iri_string
        prefix_entry_index = None

    name_entry_index = self.names.encode_entry_index(name)
    term_rows = []

    if prefix_entry_index is not None:
        prefix_entry = jelly.RdfPrefixEntry(id=prefix_entry_index, value=prefix)
        term_rows.append(jelly.RdfStreamRow(prefix=prefix_entry))

    if name_entry_index is not None:
        name_entry = jelly.RdfNameEntry(id=name_entry_index, value=name)
        term_rows.append(jelly.RdfStreamRow(name=name_entry))

    prefix_index = self.prefixes.encode_prefix_term_index(prefix)
    name_index = self.names.encode_name_term_index(name)
    return term_rows, prefix_index, name_index

encode_iri(iri_string, iri)

Encode iri.

Args: iri_string (str): full iri in string format. iri (jelly.RdfIri): iri to fill

Returns: Rows: extra rows for prefix and name tables, if any.

Source code in pyjelly/serialize/encode.py

def encode_iri(self, iri_string: str, iri: jelly.RdfIri) -> Rows:
    """
    Encode iri.

    Args:
        iri_string (str): full iri in string format.
        iri (jelly.RdfIri): iri to fill

    Returns:
        Rows: extra rows for prefix and name tables, if any.

    """
    term_rows, prefix_index, name_index = self.encode_iri_indices(iri_string)
    iri.prefix_id = prefix_index
    iri.name_id = name_index
    return term_rows

encode_default_graph(g_default_graph)

Encode default graph.

Returns: Rows: empty extra rows (for API consistency)

Source code in pyjelly/serialize/encode.py

def encode_default_graph(self, g_default_graph: jelly.RdfDefaultGraph) -> Rows:
    """
    Encode default graph.

    Returns:
        Rows: empty extra rows (for API consistency)

    """
    g_default_graph.CopyFrom(jelly.RdfDefaultGraph())
    return ()

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

Encode literal.

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. literal (jelly.RdfLiteral): literal to fill.

Raises: JellyConformanceError: if datatype specified while datatable is not used.

Returns: Rows: extra rows (i.e., datatype entries).

Source code in pyjelly/serialize/encode.py

def encode_literal(
    self,
    *,
    lex: str,
    language: str | None = None,
    datatype: str | None = None,
    literal: jelly.RdfLiteral,
) -> Rows:
    """
    Encode literal.

    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.
        literal (jelly.RdfLiteral): literal to fill.

    Raises:
        JellyConformanceError: if datatype specified while
            datatable is not used.

    Returns:
        Rows: extra rows (i.e., datatype entries).

    """
    datatype_id = None
    term_rows: tuple[()] | tuple[jelly.RdfStreamRow] = ()

    if datatype and datatype != options.STRING_DATATYPE_IRI:
        if self.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)
        datatype_entry_id = self.datatypes.encode_entry_index(datatype)

        if datatype_entry_id is not None:
            entry = jelly.RdfDatatypeEntry(id=datatype_entry_id, value=datatype)
            term_rows = (jelly.RdfStreamRow(datatype=entry),)

        datatype_id = self.datatypes.encode_datatype_term_index(datatype)

    literal.lex = lex
    if language:
        literal.langtag = language
    if datatype_id:
        literal.datatype = datatype_id
    return term_rows

encode_quoted_triple(terms, quoted_statement)

Encode a quoted triple.

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. quoted_statement (jelly.RdfTriple): quoted triple to fill.

Returns: Rows: additional stream rows with preceeding information (prefixes, names, datatypes rows, if any).

Source code in pyjelly/serialize/encode.py

def encode_quoted_triple(
    self, terms: Iterable[object], quoted_statement: jelly.RdfTriple
) -> Rows:
    """
    Encode a quoted triple.

    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.
        quoted_statement (jelly.RdfTriple): quoted triple to fill.

    Returns:
        Rows: additional stream rows with preceeding
            information (prefixes, names, datatypes rows, if any).

    """
    rows: list[jelly.RdfStreamRow] = []
    terms = iter(terms)
    extra_rows = self.encode_spo(next(terms), Slot.subject, quoted_statement)
    rows.extend(extra_rows)
    extra_rows = self.encode_spo(next(terms), Slot.predicate, quoted_statement)
    rows.extend(extra_rows)
    extra_rows = self.encode_spo(next(terms), Slot.object, quoted_statement)
    rows.extend(extra_rows)
    return rows

get_iri_field(statement, slot)

Get IRI field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_iri_field(self, statement: Statement, slot: Slot) -> jelly.RdfIri:
    """Get IRI field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_iri
    if slot == Slot.predicate:
        return statement.p_iri
    return statement.o_iri

get_literal_field(statement, slot)

Get literal field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_literal_field(self, statement: Statement, slot: Slot) -> jelly.RdfLiteral:
    """Get literal field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_literal
    if slot == Slot.predicate:
        return statement.p_literal
    return statement.o_literal

set_bnode_field(statement, slot, identifier)

Set bnode field directly based on slot.

Source code in pyjelly/serialize/encode.py

def set_bnode_field(
    self, statement: Statement, slot: Slot, identifier: str
) -> None:
    """Set bnode field directly based on slot."""
    if slot == Slot.subject:
        statement.s_bnode = identifier
    elif slot == Slot.predicate:
        statement.p_bnode = identifier
    else:
        statement.o_bnode = identifier

get_triple_field(statement, slot)

Get triple term field directly based on slot.

Source code in pyjelly/serialize/encode.py

def get_triple_field(self, statement: Statement, slot: Slot) -> jelly.RdfTriple:
    """Get triple term field directly based on slot."""
    if slot == Slot.subject:
        return statement.s_triple_term
    if slot == Slot.predicate:
        return statement.p_triple_term
    return statement.o_triple_term

`TripleStream(*args, **kwargs)`

Bases: pyjelly.serialize.streams.Stream

Methods:

Name	Description
`__new__`	Create and return a new object. See help(type) for accurate signature.

Attributes:

Name	Type	Description
`__doc__`		str(object='') -> str
`__module__`		str(object='') -> str
`__mypyc_attrs__`		Built-in immutable sequence.
`physical_type`		int([x]) -> integer

__doc__ = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__module__ = 'pyjelly.serialize.streams'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__mypyc_attrs__ = ('repeated_terms', 'encoder', 'options', 'flow', 'enrolled', 'stream_types')

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

physical_type = 1

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

__new__(*args, **kwargs)

Create and return a new object. See help(type) for accurate signature.

`encode_namespace_declaration(name, value, term_encoder)`

Encode namespace declaration.

Args: name (str): namespace prefix label value (str): namespace iri term_encoder (TermEncoder): current encoder

Returns: list[jelly.RdfStreamRow]: list of messages to append to current flow.

Source code in pyjelly/serialize/encode.py

def encode_namespace_declaration(
    name: str,
    value: str,
    term_encoder: TermEncoder,
) -> list[jelly.RdfStreamRow]:
    """
    Encode namespace declaration.

    Args:
        name (str): namespace prefix label
        value (str): namespace iri
        term_encoder (TermEncoder): current encoder

    Returns:
        list[jelly.RdfStreamRow]: list of messages to append to current flow.

    """
    iri = jelly.RdfIri()
    [*rows] = term_encoder.encode_iri(value, iri=iri)
    declaration = jelly.RdfNamespaceDeclaration(name=name, value=iri)
    row = jelly.RdfStreamRow(namespace=declaration)
    rows.append(row)
    return rows

`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.RdfStreamRow: encoded stream options row

Source code in pyjelly/serialize/encode.py

def encode_options(
    lookup_preset: options.LookupPreset,
    stream_types: options.StreamTypes,
    params: options.StreamParameters,
) -> jelly.RdfStreamRow:
    """
    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.RdfStreamRow: encoded stream options row

    """
    return jelly.RdfStreamRow(
        options=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,
        )
    )

`encode_quad(terms, term_encoder, repeated_terms)`

Encode one quad.

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.

Returns: list[jelly.RdfStreamRow]: list of messages to append to current flow.

Source code in pyjelly/serialize/encode.py

def encode_quad(
    terms: Iterable[object],
    term_encoder: TermEncoder,
    repeated_terms: list[object | None],
) -> list[jelly.RdfStreamRow]:
    """
    Encode one quad.

    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.

    Returns:
        list[jelly.RdfStreamRow]: list of messages to append to current flow.

    """
    terms = iter(terms)
    quad = jelly.RdfQuad()
    rows = encode_spo(terms, term_encoder, repeated_terms, quad)
    g = next(terms)
    if repeated_terms[Slot.graph] != g:
        extra_rows = term_encoder.encode_graph(g, quad)
        rows.extend(extra_rows)
        repeated_terms[Slot.graph] = g
    row = jelly.RdfStreamRow(quad=quad)
    rows.append(row)
    return rows

`encode_triple(terms, term_encoder, repeated_terms)`

Encode one triple.

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.

Returns: list[jelly.RdfStreamRow]: list of rows to add to the current flow.

Source code in pyjelly/serialize/encode.py

def encode_triple(
    terms: Iterable[object],
    term_encoder: TermEncoder,
    repeated_terms: list[object | None],
) -> list[jelly.RdfStreamRow]:
    """
    Encode one triple.

    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.

    Returns:
        list[jelly.RdfStreamRow]: list of rows to add to the current flow.

    """
    triple = jelly.RdfTriple()
    terms = iter(terms)
    rows = encode_spo(terms, term_encoder, repeated_terms, triple)
    row = jelly.RdfStreamRow(triple=triple)
    rows.append(row)
    return rows

`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

`jelly`

Generated protocol buffer code.

API reference

pyjelly

errors

JellyConformanceError

JellyAssertionError

JellyNotImplementedError

integrations

generic

generic_sink

parse

serialize

rdflib

register_extension_to_rdflib(extension='.jelly')

parse

serialize

jelly

rdf_pb2

options

INTEGRATION_SIDE_EFFECTS = True

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

__repr__()

register_mimetypes(extension='.jelly')

parse

decode

ParsingMode

Decoder(adapter)

options_from_frame(frame, *, delimited)

ioutils

delimited_jelly_hint(header)

get_options_and_frames(inp)

lookup

MAX_LOOKUP_SIZE = 4096

__file__ = '/home/runner/work/pyjelly/pyjelly/.venv/lib/python3.10/site-packages/pyjelly/parse/lookup.cpython-310-x86_64-linux-gnu.so'

__name__ = 'pyjelly.parse.lookup'

__package__ = 'pyjelly.parse'

JellyAssertionError

JellyConformanceError

LookupDecoder(*args, **kwargs)

serialize

encode

TermEncoder(lookup_preset=None)

split_iri(iri_string)

encode_spo(terms, term_encoder, repeated_terms, statement)

encode_triple(terms, term_encoder, repeated_terms)

encode_quad(terms, term_encoder, repeated_terms)

encode_namespace_declaration(name, value, term_encoder)

encode_options(lookup_preset, stream_types, params)

flows

FrameFlow(initlist=None, *, logical_type=None, **__kwargs)

ManualFrameFlow(initlist=None, *, logical_type=None, **__kwargs)

BoundedFrameFlow(initlist=None, logical_type=None, *, frame_size=None)

GraphsFrameFlow(initlist=None, *, logical_type=None, **__kwargs)

DatasetsFrameFlow(initlist=None, *, logical_type=None, **__kwargs)

flow_for_type(logical_type)

lookup

__file__ = '/home/runner/work/pyjelly/pyjelly/.venv/lib/python3.10/site-packages/pyjelly/serialize/lookup.cpython-310-x86_64-linux-gnu.so'

__name__ = 'pyjelly.serialize.lookup'

__package__ = 'pyjelly.serialize'

Lookup(*args, **kwargs)

LookupEncoder(*args, **kwargs)

streams

DEFAULT_FRAME_SIZE = 250

STREAM_DISPATCH = {1: <class 'pyjelly.serialize.streams.TripleStream'>, 2: <class 'pyjelly.serialize.streams.QuadStream'>, 3: <class 'pyjelly.serialize.streams.GraphStream'>}

TYPE_CHECKING = False

__file__ = '/home/runner/work/pyjelly/pyjelly/pyjelly/serialize/streams.cpython-310-x86_64-linux-gnu.so'

__name__ = 'pyjelly.serialize.streams'

__package__ = 'pyjelly.serialize'

BoundedFrameFlow(initlist=None, logical_type=None, *, frame_size=None)

FrameFlow(initlist=None, *, logical_type=None, **__kwargs)

GraphStream(*args, **kwargs)

ManualFrameFlow(initlist=None, *, logical_type=None, **__kwargs)

QuadStream(*args, **kwargs)

SerializerOptions(flow=None, frame_size=250, logical_type=0, params=<factory>, lookup_preset=<factory>)

Stream(*args, **kwargs)

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

TermEncoder(lookup_preset=None)

TripleStream(*args, **kwargs)

encode_namespace_declaration(name, value, term_encoder)

encode_options(lookup_preset, stream_types, params)

encode_quad(terms, term_encoder, repeated_terms)

`pyjelly`

`errors`

`JellyConformanceError`

`JellyAssertionError`

`JellyNotImplementedError`

`integrations`

`generic`

`generic_sink`

`parse`

`serialize`

`rdflib`

`register_extension_to_rdflib(extension='.jelly')`

`parse`

`serialize`

`jelly`

`rdf_pb2`

`options`

`INTEGRATION_SIDE_EFFECTS = True`

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

`repr()`

`register_mimetypes(extension='.jelly')`

`parse`

`decode`

`ParsingMode`

`Decoder(adapter)`

`options_from_frame(frame, *, delimited)`

`ioutils`

`delimited_jelly_hint(header)`

`get_options_and_frames(inp)`

`lookup`

`MAX_LOOKUP_SIZE = 4096`

`file = '/home/runner/work/pyjelly/pyjelly/.venv/lib/python3.10/site-packages/pyjelly/parse/lookup.cpython-310-x86_64-linux-gnu.so'`

`name = 'pyjelly.parse.lookup'`

`package = 'pyjelly.parse'`

`JellyAssertionError`

`JellyConformanceError`

`LookupDecoder(*args, **kwargs)`

`serialize`

`encode`

`TermEncoder(lookup_preset=None)`

`split_iri(iri_string)`

`encode_spo(terms, term_encoder, repeated_terms, statement)`

`encode_triple(terms, term_encoder, repeated_terms)`

`encode_quad(terms, term_encoder, repeated_terms)`

`encode_namespace_declaration(name, value, term_encoder)`

`encode_options(lookup_preset, stream_types, params)`

`flows`

`FrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

`ManualFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

`BoundedFrameFlow(initlist=None, logical_type=None, *, frame_size=None)`

`GraphsFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

`DatasetsFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

`flow_for_type(logical_type)`

`lookup`

`file = '/home/runner/work/pyjelly/pyjelly/.venv/lib/python3.10/site-packages/pyjelly/serialize/lookup.cpython-310-x86_64-linux-gnu.so'`

`name = 'pyjelly.serialize.lookup'`

`package = 'pyjelly.serialize'`

`Lookup(*args, **kwargs)`

`LookupEncoder(*args, **kwargs)`

`streams`

`DEFAULT_FRAME_SIZE = 250`

`STREAM_DISPATCH = {1: <class 'pyjelly.serialize.streams.TripleStream'>, 2: <class 'pyjelly.serialize.streams.QuadStream'>, 3: <class 'pyjelly.serialize.streams.GraphStream'>}`

`TYPE_CHECKING = False`

`file = '/home/runner/work/pyjelly/pyjelly/pyjelly/serialize/streams.cpython-310-x86_64-linux-gnu.so'`

`name = 'pyjelly.serialize.streams'`

`package = 'pyjelly.serialize'`

`BoundedFrameFlow(initlist=None, logical_type=None, *, frame_size=None)`

`FrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

`GraphStream(*args, **kwargs)`

`ManualFrameFlow(initlist=None, *, logical_type=None, **__kwargs)`

`QuadStream(*args, **kwargs)`

`SerializerOptions(flow=None, frame_size=250, logical_type=0, params=<factory>, lookup_preset=<factory>)`

`Stream(*args, **kwargs)`

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

`TermEncoder(lookup_preset=None)`

`TripleStream(*args, **kwargs)`

`encode_namespace_declaration(name, value, term_encoder)`

`encode_options(lookup_preset, stream_types, params)`

`encode_quad(terms, term_encoder, repeated_terms)`

`encode_triple(terms, term_encoder, repeated_terms)`