RDF compatibility

rdf

RDF/TriG/RDF 1.2 compatibility and RDF Message Log support.

This module deliberately keeps RDF compatibility outside the N3 rule parser. The N3 parser remains formula/rule-oriented; RDF compatibility uses rdflib for ordinary RDF syntaxes and a small surface-syntax adapter for RDF 1.2 constructs that rdflib does not yet accept uniformly.

RdfSyntaxError

Bases: SyntaxError

Raised when the RDF compatibility parser rejects input.

assert_rdf12_surface_syntax

assert_rdf12_surface_syntax(text: str, *, format: str = 'turtle') -> None

Reject common RDF 1.2 negative-suite surface forms before parsing.

rdflib is intentionally liberal in places and not yet complete for RDF 1.2 triple terms / directional language syntax. These checks mirror the surface guards Eyeling uses for the W3C RDF 1.2 syntax tests.

Source code in pyling/rdf.py

def assert_rdf12_surface_syntax(text: str, *, format: str = "turtle") -> None:
    """Reject common RDF 1.2 negative-suite surface forms before parsing.

    rdflib is intentionally liberal in places and not yet complete for RDF 1.2
    triple terms / directional language syntax. These checks mirror the surface
    guards Eyeling uses for the W3C RDF 1.2 syntax tests.
    """
    data = str(text or "")
    if re.search(r"\\u[dD][89a-fA-F][0-9a-fA-F]{2}", data):
        raise RdfSyntaxError("RDF 1.2 numeric escapes must not encode UTF-16 surrogate code points")

    # Line syntaxes have stricter absolute IRI and no annotation constraints.
    if format in {"nt", "ntriples", "n-triples", "nq", "nquads", "n-quads"}:
        i = 0
        while i < len(data):
            ch = data[i]
            if ch == "#":
                while i < len(data) and data[i] not in "\r\n":
                    i += 1
                continue
            if ch in {'"', "'"}:
                end = _read_string_at(data, i)
                j = end
                if data.startswith("@", j):
                    j += 1
                    start = j
                    while j < len(data) and re.match(r"[A-Za-z0-9-]", data[j]):
                        j += 1
                    _assert_valid_lang_tag(data[start:j])
                elif data.startswith("^^<", j):
                    dt_end = _read_iri_at(data, j + 2)
                    dt = data[j + 3 : dt_end - 1]
                    if dt in {str(RDF.langString), str(RDF) + "dirLangString"}:
                        raise RdfSyntaxError(f"RDF datatype {dt} requires a language tag")
                i = end
                continue
            if ch == "<":
                if data.startswith("<<", i):
                    i += 2
                    continue
                end = _read_iri_at(data, i)
                iri = data[i + 1 : end - 1]
                if not _is_abs_iri(iri):
                    raise RdfSyntaxError(f"RDF line-syntax IRIREF must be absolute: <{iri}>")
                i = end
                continue
            if data.startswith("{|", i) or data.startswith("|}", i):
                raise RdfSyntaxError("RDF line syntax does not allow Turtle annotation syntax")
            i += 1

term_from_rdflib

term_from_rdflib(term) -> Term

Convert an RDFLib term into a Pyling term.

Source code in pyling/rdf.py

def term_from_rdflib(term) -> Term:
    """Convert an RDFLib term into a Pyling term."""
    return _rdflib_term_to_term(term, cache={})

triple_from_rdflib

triple_from_rdflib(triple) -> Triple

Convert an RDFLib triple-like tuple into a Pyling triple.

Source code in pyling/rdf.py

def triple_from_rdflib(triple) -> Triple:
    """Convert an RDFLib triple-like tuple into a Pyling triple."""
    cache: dict[object, Term] = {}
    s, p, o = triple
    return Triple(
        _rdflib_term_to_term(s, cache=cache),
        _rdflib_term_to_term(p, cache=cache),
        _rdflib_term_to_term(o, cache=cache),
    )

parse_rdf_graph

parse_rdf_graph(graph: Graph | Dataset) -> Document

Convert an RDFLib Graph or Dataset into an Eyeling Document.

Source code in pyling/rdf.py

def parse_rdf_graph(graph: Graph | Dataset) -> Document:
    """Convert an RDFLib Graph or Dataset into an Eyeling Document."""
    env = PrefixEnv({})
    triples: list[Triple] = []
    cache: dict[object, Term] = {}

    if isinstance(graph, Dataset):
        default_id = str(graph.default_graph.identifier)
        by_graph: dict[Term | None, list[Triple]] = {}
        for s, p, o, g in graph.quads((None, None, None, None)):
            tr = Triple(
                _rdflib_term_to_term(s, cache=cache),
                _rdflib_term_to_term(p, cache=cache),
                _rdflib_term_to_term(o, cache=cache),
            )
            gid = (
                None
                if str(g) == default_id or str(g).endswith("default")
                else _rdflib_term_to_term(g, cache=cache)
            )
            by_graph.setdefault(gid, []).append(tr)
        for gid, body in by_graph.items():
            if gid is None:
                triples.extend(body)
            else:
                triples.append(Triple(gid, Iri(LOG_NAME_OF), GraphTerm(body)))
    else:
        for s, p, o in graph:
            triples.append(
                Triple(
                    _rdflib_term_to_term(s, cache=cache),
                    _rdflib_term_to_term(p, cache=cache),
                    _rdflib_term_to_term(o, cache=cache),
                )
            )

    _bind_rdflib_namespaces(env, graph, triples)
    return Document(env, triples, [], [], [])

document_from_rdflib

document_from_rdflib(graph: Graph | Dataset) -> Document

Convert an RDFLib Graph or Dataset into a Pyling Document.

Source code in pyling/rdf.py

def document_from_rdflib(graph: Graph | Dataset) -> Document:
    """Convert an RDFLib Graph or Dataset into a Pyling Document."""
    return parse_rdf_graph(graph)

term_to_rdflib

term_to_rdflib(term: Term)

Convert an ordinary Pyling term into an RDFLib term.

Source code in pyling/rdf.py

def term_to_rdflib(term: Term):
    """Convert an ordinary Pyling term into an RDFLib term."""
    return _term_to_rdflib_term(term, {})

triple_to_rdflib

triple_to_rdflib(triple: Triple)

Convert an ordinary Pyling triple into an RDFLib triple tuple.

Source code in pyling/rdf.py

def triple_to_rdflib(triple: Triple):
    """Convert an ordinary Pyling triple into an RDFLib triple tuple."""
    cache: dict[Term, object] = {}
    return (
        _term_to_rdflib_term(triple.s, cache),
        _term_to_rdflib_term(triple.p, cache),
        _term_to_rdflib_term(triple.o, cache),
    )

triples_to_rdflib_graph

triples_to_rdflib_graph(triples: Iterable[Triple], prefixes: PrefixEnv | None = None, *, graph: Graph | None = None) -> Graph

Convert ordinary pyling triples into an RDFLib Graph.

Formula terms, lists, open lists, and variables are intentionally rejected: they are Notation3 structures and do not have a lossless representation in a normal RDFLib Graph.

Source code in pyling/rdf.py

def triples_to_rdflib_graph(
    triples: Iterable[Triple],
    prefixes: PrefixEnv | None = None,
    *,
    graph: Graph | None = None,
) -> Graph:
    """Convert ordinary pyling triples into an RDFLib Graph.

    Formula terms, lists, open lists, and variables are intentionally rejected:
    they are Notation3 structures and do not have a lossless representation in a
    normal RDFLib Graph.
    """
    target = Graph() if graph is None else graph
    if prefixes is not None:
        for name in sorted(prefixes.declared):
            if name in prefixes.map:
                target.bind(name, URIRef(prefixes.map[name]))
    cache: dict[Term, object] = {}
    for tr in triples:
        target.add(
            (
                _term_to_rdflib_term(tr.s, cache),
                _term_to_rdflib_term(tr.p, cache),
                _term_to_rdflib_term(tr.o, cache),
            )
        )
    return target

document_to_rdflib

document_to_rdflib(document: Document, *, graph: Graph | None = None, include_input_facts: bool = True) -> Graph

Convert a Pyling Document's ordinary fact triples into an RDFLib Graph.

Rules, formulas, lists, open lists, and variables are still rejected by triples_to_rdflib_graph because they do not have a lossless ordinary RDFLib Graph representation.

Source code in pyling/rdf.py

def document_to_rdflib(
    document: Document,
    *,
    graph: Graph | None = None,
    include_input_facts: bool = True,
) -> Graph:
    """Convert a Pyling Document's ordinary fact triples into an RDFLib Graph.

    Rules, formulas, lists, open lists, and variables are still rejected by
    ``triples_to_rdflib_graph`` because they do not have a lossless ordinary
    RDFLib Graph representation.
    """
    triples = document.triples if include_input_facts else []
    return triples_to_rdflib_graph(triples, document.prefixes, graph=graph)

parse_rdf_text

parse_rdf_text(text: str, *, format: str | None = None, base_iri: str | None = None, rdf12: bool = True, label: str | None = None) -> Document

Parse RDF/Turtle/TriG/N-Triples/N-Quads text into an Eyeling Document.

Source code in pyling/rdf.py

def parse_rdf_text(text: str, *, format: str | None = None, base_iri: str | None = None, rdf12: bool = True, label: str | None = None) -> Document:
    """Parse RDF/Turtle/TriG/N-Triples/N-Quads text into an Eyeling Document."""
    source = str(text or "")
    fmt = _guess_format(source, format)
    placeholders: dict[str, _Rdf12Placeholder] = {}
    if rdf12:
        assert_rdf12_surface_syntax(source, format=fmt)
        data, placeholders = _normalize_rdf12(source, base_iri=base_iri, format=fmt)
    else:
        data = source

    env = PrefixEnv({})
    if base_iri:
        env.base_iri = base_iri

    graph = _rdflib_parse(data, format=fmt, base_iri=base_iri)
    _validate_placeholder_positions(graph, placeholders)
    triples: list[Triple] = []

    # Preserve only prefixes actually declared in the source. rdflib attaches a
    # long list of common namespaces to every graph; emitting those would make
    # Eyeling output noisy and unlike the JavaScript implementation.
    for line in source.splitlines():
        m = PREFIX_LINE_RE.match(line)
        if not m:
            continue
        if m.group(1) is not None or m.group(3) is not None:
            raw = m.group(1) or m.group(3) or ":"
            iri = m.group(2) or m.group(4) or ""
            if raw.endswith(":"):
                env.set_prefix(raw[:-1], iri, declared=True)
        elif m.group(5) or m.group(6):
            env.base_iri = m.group(5) or m.group(6)

    if isinstance(graph, Dataset):
        default_id = str(graph.default_graph.identifier)
        by_graph: dict[Term | None, list[Triple]] = {}
        for s, p, o, g in graph.quads((None, None, None, None)):
            tr = Triple(
                _rdflib_term_to_term(s, placeholders),
                _rdflib_term_to_term(p, placeholders),
                _rdflib_term_to_term(o, placeholders),
            )
            gid = (
                None
                if str(g) == default_id or str(g).endswith("default")
                else _rdflib_term_to_term(g, placeholders)
            )
            by_graph.setdefault(gid, []).append(tr)
        for gid, body in by_graph.items():
            if gid is None:
                triples.extend(body)
            else:
                triples.append(Triple(gid, Iri(LOG_NAME_OF), GraphTerm(body)))
    else:
        for s, p, o in graph:
            triples.append(
                Triple(
                    _rdflib_term_to_term(s, placeholders),
                    _rdflib_term_to_term(p, placeholders),
                    _rdflib_term_to_term(o, placeholders),
                )
            )
    return Document(env, triples, [], [], [])

parse_rdf_message_log

parse_rdf_message_log(text: str, *, base_iri: str | None = None, label: str | None = None) -> Document

Parse a whole RDF Message Log into Eyeling's replay vocabulary.

Source code in pyling/rdf.py

def parse_rdf_message_log(text: str, *, base_iri: str | None = None, label: str | None = None) -> Document:
    """Parse a whole RDF Message Log into Eyeling's replay vocabulary."""
    source = str(text or "")
    if not is_rdf_message_log(source):
        raise RdfSyntaxError("input is not an RDF Message Log")
    chunks = split_rdf_messages(source)
    preludes = _message_preludes(chunks)
    payloads = [
        _parse_payload(chunk, prelude, idx + 1, base_iri)
        for idx, (chunk, prelude) in enumerate(zip(chunks, preludes))
    ]
    return _message_replay_document(source, payloads, base_iri=base_iri)

iter_rdf_message_documents

iter_rdf_message_documents(text: str, *, base_iri: str | None = None) -> Iterator[Document]

Yield one replay document per RDF Message Log message.

Source code in pyling/rdf.py

def iter_rdf_message_documents(text: str, *, base_iri: str | None = None) -> Iterator[Document]:
    """Yield one replay document per RDF Message Log message."""
    source = str(text or "")
    if not is_rdf_message_log(source):
        raise RdfSyntaxError("input is not an RDF Message Log")
    chunks = split_rdf_messages(source)
    for idx, (chunk, prelude) in enumerate(zip(chunks, _message_preludes(chunks)), start=1):
        payload = _parse_payload(chunk, prelude, idx, base_iri)
        yield _message_replay_document(source, [payload], base_iri=base_iri, first_index=idx)