Generic API

This guide explains how to use pyjelly’s generic API to write and read RDF statements into the Jelly format without any external library.

Installation

Install pyjelly from PyPI:

pip install pyjelly

The requirements are the same as with RDFLib.

Usage without external libraries

Unlike the example in getting started, the Generic API does not use the RDFLib or any other third-party libraries, but it works in much the same way.

Serializing statements to a Jelly file

To make a set of triples/quads and write them to a Jelly file, use:

from pyjelly.integrations.generic.generic_sink import *

# Create a generic sink object
generic_sink = GenericStatementSink()

# Let's add triples one by one
generic_sink.add(
    Triple(
        IRI("http://example.com/subject"),
        IRI("http://example.com/predicate"),
        Literal("Hello", langtag="en"),
    )
)
generic_sink.add(
    Triple(
        BlankNode("B1"),
        IRI("http://example.com/hasName"),
        Literal("Bob"),
    )
)

# Write into a Jelly file
with open("output.jelly", "wb") as out_file:
    generic_sink.serialize(out_file)

print("All done.")

This example uses pyjelly’s simple custom triple/quad type, which is easy to create and work with.

The terms in the statements are IRI, BlankNode and Literal. You can read their parts with IRI.iri, BlankNode.identifier, and Literal.lex, Literal.langtag, Literal.datatype (the datatype IRI as a string) and Literal.direction. str() gives a term in N-Triples syntax, e.g., <http://example.com/subject> or "Hello"@en.

Parsing statements from a Jelly file

To load triples/quads from a .jelly file into a GenericStatementSink, see:

from pyjelly.integrations.generic.generic_sink import *

# Create a generic sink object
generic_sink = GenericStatementSink()

# Load triples from the Jelly file
with open("output.jelly", "rb") as in_file:
    generic_sink.parse(in_file)

# Let's inspect them statement by statement
for statement in generic_sink:
    if isinstance(statement, Triple):
        print(statement)

print("All done.")

GenericStatementSink keeps all statements in memory. To read a large file statement by statement instead, use parse_jelly_flat in a generator:

from pyjelly.integrations.generic.parse import parse_jelly_flat


# Helper generator that streams statements from a Jelly file path
def stream_statements(jelly_path):
    with open(jelly_path, "rb") as f:
        # Statements are read one by one, as they are needed
        yield from parse_jelly_flat(f)


# Example usage, just printing:
for statement in stream_statements("output.jelly"):
    print(statement)

print("All done.")

Only the statement that is being processed is kept in memory.

Parsing a stream of graphs

Similarly, to process a Jelly stream as a stream of graphs through generic API, see:

import gzip
import urllib.request
from itertools import islice

from pyjelly.integrations.generic.parse import parse_jelly_grouped

# Dataset: Katrina weather measurements (10k graphs)
# Documentation: https://w3id.org/riverbench/datasets/lod-katrina/dev
url = "https://w3id.org/riverbench/datasets/lod-katrina/dev/files/jelly_10K.jelly.gz"

# Load, uncompress .gz file, and pass to Jelly parser, all in a streaming manner
with (
    urllib.request.urlopen(url) as response,
    gzip.open(response) as jelly_stream,
):
    # Parse into sinks (one per graph)
    graphs = parse_jelly_grouped(jelly_stream)

    # Only the first 50 graphs, the rest will not be parsed
    for i, graph in enumerate(islice(graphs, 50)):
        print(f"Graph {i} in the stream has {len(graph)} triples")

We use a dataset of weather measurements and count the number of triples in each graph.

Parsing a stream of statements

You can also process a Jelly stream as a flat stream of statements. We look through a fragment of Denmark's OpenStreetMap to find all city names:

import gzip
import urllib.request

from pyjelly.integrations.generic.generic_sink import *
from pyjelly.integrations.generic.parse import parse_jelly_flat

# Dataset: OpenStreetMap data for Denmark (first 10k objects)
# Documentation: https://w3id.org/riverbench/datasets/osm2rdf-denmark/dev
url = (
    "https://w3id.org/riverbench/datasets/osm2rdf-denmark/dev/files/jelly_10K.jelly.gz"
)

# We are looking for city names in the dataset
predicate_to_look_for = IRI("https://www.openstreetmap.org/wiki/Key:addr:city")
city_names = set()

with (
    urllib.request.urlopen(url) as response,
    gzip.open(response) as jelly_stream,
):
    for event in parse_jelly_flat(jelly_stream):
        if isinstance(event, Triple):  # we are only interested in triples
            if event.p == predicate_to_look_for:
                city_names.add(event.o.lex)

print(f"Found {len(city_names)} unique city names in the dataset.")
print("Sample city names:")
for city in list(city_names)[:10]:
    print(f"- {city}")

parse_jelly_flat returns a generator of statements, so the file is processed statement by statement.

Serializing streams

Serializing a stream of graphs

If you have a generator object containing graphs, you can use a generic approach for serialization:

from pyjelly.integrations.generic.generic_sink import *
from pyjelly.integrations.generic.serialize import grouped_stream_to_file
import random


# Example generator that yields graphs
def generate_sample_sinks():
    for _ in range(10):
        sink = GenericStatementSink()
        sink.add(
            Triple(
                IRI("http://example.com/sensor"),
                IRI("http://example.com/humidity"),
                Literal(str(random.random())),
            )
        )
        yield sink


output_file = "output.jelly"
print(f"Streaming graphs into {output_file}…")
with open(output_file, "wb") as out_f:
    grouped_stream_to_file(generate_sample_sinks(), out_f)
print("All done.")

Each sink is written as one RDF Message, so the grouping is kept when the stream is read with parse_jelly_grouped.

Serializing a stream of statements

If you have a generator object containing statements, you can serialize it like this:

from pyjelly.integrations.generic.serialize import flat_stream_to_file
from pyjelly.integrations.generic.generic_sink import *
import random


# Example generator that yields raw triples
def generate_sample_triples():
    for _ in range(10):
        yield Triple(
            IRI("http://example.com/sensor"),
            IRI("http://example.com/humidity"),
            Literal(str(random.random())),
        )


output_file = "flat_output.jelly"
print(f"Streaming triples into {output_file}…")
sample_triples = generate_sample_triples()
with open(output_file, "wb") as out_f:
    flat_stream_to_file(sample_triples, out_f)
print("All done.")

The statements are written in order, as one flat stream.

Working with byte buffers and Kafka

When working with Kafka or other message brokers, you may want to write Jelly data to a byte buffer instead of a file. You can do this by using the BytesIO class from the io module:

import io

from pyjelly.integrations.generic.generic_sink import *

g1 = GenericStatementSink()

g1.add(
    Triple(
        IRI("http://example.com/subject"),
        IRI("http://example.com/predicate"),
        Literal("Hello", langtag="en"),
    )
)

# Write the data into a byte buffer (bytes type)
with io.BytesIO() as write_buffer:
    g1.serialize(write_buffer)
    data = write_buffer.getvalue()

print(f"Serialized data size: {len(data)} bytes")

# Parse the data back
g2 = GenericStatementSink()
with io.BytesIO(data) as read_buffer:
    g2.parse(read_buffer)

print("\nParsed triples:")
for statement in g2:
    print(statement)

The data variable is of type bytes, and can be passed to Kafka with KafkaProducer.send(value=data), or any other API that accepts byte buffers. With the RDFLib integration, write to the buffer with graph.serialize(destination=buffer, format="jelly"), and read the bytes back with graph.parse(data=data, format="jelly").

When working with Kafka, you should be aware of the broker's offset management and partitioning strategies. Data within one Jelly stream must be strictly ordered and no frames may be dropped. If you have less strict ordering guarantees, you should split up the stream into multiple Jelly streams, each with guaranteed consistency.

Accessing stream frame metadata

It is possible to access RdfStreamFrame metadata in pyjelly via context variables. To do that, define a context variable before parsing and pass it as argument to parser functions, e.g.:

from collections.abc import MutableMapping
from contextvars import ContextVar
import urllib.request

from pyjelly.integrations.generic.parse import parse_jelly_grouped

frame_metadata: ContextVar[MutableMapping[str, bytes]] = ContextVar("frame_metadata")

url = "https://registry.petapico.org/nanopubs.jelly"
with urllib.request.urlopen(url) as response:
    graphs = parse_jelly_grouped(response, frame_metadata=frame_metadata)
    # Just show the first graph
    graph = next(graphs)
    print(f"The first graph in the stream has {len(graph)} triples.")
    print(f"Its frame metadata: {frame_metadata.get()}")