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
rdflib

rdflib

Modules:

Name	Description
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.

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

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

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

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

serialize

Classes:

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

RDFLibJellySerializer(store)

Bases: Serializer

RDFLib serializer for writing graphs in Jelly RDF stream format.

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

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)

jelly

Modules:

Name	Description
rdf_pb2	Generated protocol buffer code.

rdf_pb2

Generated protocol buffer code.

options

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.

register_mimetypes(extension='.jelly')

Associate files that have Jelly extension with Jelly MIME types.

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

Source code in pyjelly/options.py

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

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

parse

Modules:

Name	Description
ioutils
lookup

ioutils

Functions:

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

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)

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

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

lookup

Classes:

Name	Description
LookupDecoder	Shared base for RDF lookup encoders using Jelly compression.

LookupDecoder(*, lookup_size)

Shared base for RDF lookup encoders using Jelly compression.

Tracks the last assigned and last reused index.

Parameters:

Name	Type	Description	Default
lookup_size	int	Maximum lookup size.	required

Source code in pyjelly/parse/lookup.py

def __init__(self, *, lookup_size: int) -> None:
    if lookup_size > MAX_LOOKUP_SIZE:
        msg = f"lookup size must be less than {MAX_LOOKUP_SIZE}"
        raise JellyAssertionError(msg)
    self.lookup_size = lookup_size
    placeholders = (None,) * lookup_size
    self.data: deque[str | None] = deque(placeholders, maxlen=lookup_size)
    self.last_assigned_index = 0
    self.last_reused_index = 0

serialize

Modules:

Name	Description
encode
flows
lookup

encode

Classes:

Name	Description
Slot	Slots for encoding RDF terms.

Functions:

Name	Description
new_repeated_terms	Create a new dictionary for repeated terms.

Slot

Bases: str, Enum

Slots for encoding RDF terms.

new_repeated_terms()

Create a new dictionary for repeated terms.

Source code in pyjelly/serialize/encode.py

def new_repeated_terms() -> dict[Slot, object]:
    """Create a new dictionary for repeated terms."""
    return dict.fromkeys(Slot)

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	Produces frames automatically when a fixed number of rows is reached.

FrameFlow

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.

Methods:

Name	Description
__init_subclass__	Register subclasses of FrameFlow with their logical stream type.

__init_subclass__()

Register subclasses of FrameFlow with their logical stream type.

This allows for dynamic dispatch based on the logical stream type.

Source code in pyjelly/serialize/flows.py

def __init_subclass__(cls) -> None:
    """
    Register subclasses of FrameFlow with their logical stream type.

    This allows for dynamic dispatch based on the logical stream type.
    """
    if cls.logical_type != jelly.LOGICAL_STREAM_TYPE_UNSPECIFIED:
        cls.registry[cls.logical_type] = cls

ManualFrameFlow

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.

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

Bases: FrameFlow

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

Used for delimited encoding (default mode).

Source code in pyjelly/serialize/flows.py

def __init__(
    self,
    initlist: Iterable[jelly.RdfStreamRow] | None = None,
    *,
    frame_size: int | None = None,
) -> None:
    super().__init__(initlist)
    self.frame_size = frame_size or self.default_frame_size

lookup

Classes:

Name	Description
Lookup	Fixed-size 1-based string-to-index mapping with LRU eviction.
LookupEncoder	Shared base for RDF lookup encoders using Jelly compression.

Lookup(max_size)

Fixed-size 1-based string-to-index mapping with LRU eviction.

• Assigns incrementing indices starting from 1.
• After reaching the maximum size, reuses the existing indices from evicting the least-recently-used entries.
• Index 0 is reserved for delta encoding in Jelly streams.

To check if a key exists, use .move(key) and catch KeyError. If KeyError is raised, the key can be inserted with .insert(key).

Parameters:

Name	Type	Description	Default
max_size	int	Maximum number of entries. Zero disables lookup.	required

Source code in pyjelly/serialize/lookup.py

def __init__(self, max_size: int) -> None:
    self.data = OrderedDict[str, int]()
    self.max_size = max_size
    self._evicting = False

LookupEncoder(*, lookup_size)

Shared base for RDF lookup encoders using Jelly compression.

Tracks the last assigned and last reused index.

Parameters:

Name	Type	Description	Default
lookup_size	int	Maximum lookup size.	required

Methods:

Name	Description
encode_entry_index	Get or assign the index to use in an entry.

Source code in pyjelly/serialize/lookup.py

def __init__(self, *, lookup_size: int) -> None:
    self.lookup = Lookup(max_size=lookup_size)
    self.last_assigned_index = 0
    self.last_reused_index = 0

encode_entry_index(key)

Get or assign the index to use in an entry.

Returns:

Type	Description
int or None	0 if the new index is sequential (last_assigned_index + 1) actual assigned/reused index otherwise None if the key already exists
If the return value is None, the entry is already in the lookup and does not
need to be emitted. Any integer value (including 0) means the entry is new
and should be emitted.

Source code in pyjelly/serialize/lookup.py

def encode_entry_index(self, key: str) -> int | None:
    """
    Get or assign the index to use in an entry.

    Returns
    -------
    int or None
        - 0 if the new index is sequential (`last_assigned_index + 1`)
        - actual assigned/reused index otherwise
        - None if the key already exists

    If the return value is None, the entry is already in the lookup and does not
    need to be emitted. Any integer value (including 0) means the entry is new
    and should be emitted.

    """
    try:
        self.lookup.make_last_to_evict(key)
        return None  # noqa: TRY300
    except KeyError:
        previous_index = self.last_assigned_index
        index = self.lookup.insert(key)
        self.last_assigned_index = index
        if index == previous_index + 1:
            return 0
        return index