jsonld

This serialiser will output an RDF Graph as a JSON-LD formatted document. See http://json-ld.org/

Example

>>> from rdflib import Graph
>>> testrdf = '''
... @prefix dc: <http://purl.org/dc/terms/> .
... <http://example.org/about>
...     dc:title "Someone's Homepage"@en .
... '''

>>> g = Graph().parse(data=testrdf, format='n3')

>>> print(g.serialize(format='json-ld', indent=2))
[
  {
    "@id": "http://example.org/about",
    "http://purl.org/dc/terms/title": [
      {
        "@language": "en",
        "@value": "Someone's Homepage"
      }
    ]
  }
]

Classes:

• JsonLDSerializer –

  JSON-LD RDF graph serializer.

Functions:

• from_rdf –

PLAIN_LITERAL_TYPES module-attribute

PLAIN_LITERAL_TYPES = {boolean, integer, double, string}

__all__ module-attribute

__all__ = ['JsonLDSerializer', 'from_rdf']

Converter

Converter(context: Context, use_native_types: bool, use_rdf_type: bool)

Methods:

• add_to_node –
• convert –
• from_graph –
• process_subject –
• to_collection –
• to_raw_value –
• type_coerce –

Attributes:

• context –
• use_native_types –
• use_rdf_type –

Source code in rdflib/plugins/serializers/jsonld.py

def __init__(self, context: Context, use_native_types: bool, use_rdf_type: bool):
    self.context = context
    self.use_native_types = context.active or use_native_types
    self.use_rdf_type = use_rdf_type

context instance-attribute

context = context

use_native_types instance-attribute

use_native_types = active or use_native_types

use_rdf_type instance-attribute

use_rdf_type = use_rdf_type

add_to_node

add_to_node(graph: Graph, s: IdentifiedNode, p: IdentifiedNode, o: Identifier, s_node: Dict[str, Any], nodemap)

Source code in rdflib/plugins/serializers/jsonld.py

def add_to_node(
    self,
    graph: Graph,
    s: IdentifiedNode,
    p: IdentifiedNode,
    o: Identifier,
    s_node: Dict[str, Any],
    nodemap,
):
    context = self.context

    if isinstance(o, Literal):
        datatype = str(o.datatype) if o.datatype else None
        language = o.language
        term = context.find_term(str(p), datatype, language=language)
    else:
        containers = [LIST, None] if graph.value(o, RDF.first) else [None]
        for container in containers:
            for coercion in (ID, VOCAB, UNDEF):
                # type error: Argument 2 to "find_term" of "Context" has incompatible type "object"; expected "Union[str, Defined, None]"
                # type error: Argument 3 to "find_term" of "Context" has incompatible type "Optional[str]"; expected "Union[Defined, str]"
                term = context.find_term(str(p), coercion, container)  # type: ignore[arg-type]
                if term:
                    break
            if term:
                break

    node = None
    use_set = not context.active

    if term:
        p_key = term.name

        if term.type:
            node = self.type_coerce(o, term.type)
        # type error: "Identifier" has no attribute "language"
        elif term.language and o.language == term.language:  # type: ignore[attr-defined]
            node = str(o)
        # type error: Right operand of "and" is never evaluated
        elif context.language and (term.language is None and o.language is None):  # type: ignore[unreachable]
            node = str(o)  # type: ignore[unreachable]

        if LIST in term.container:
            node = [
                self.type_coerce(v, term.type)
                or self.to_raw_value(graph, s, v, nodemap)
                for v in self.to_collection(graph, o)
            ]
        elif LANG in term.container and language:
            value = s_node.setdefault(p_key, {})
            values = value.get(language)
            node = str(o)
            if values or SET in term.container:
                if not isinstance(values, list):
                    value[language] = values = [values]
                values.append(node)
            else:
                value[language] = node
            return
        elif SET in term.container:
            use_set = True

    else:
        p_key = context.to_symbol(p)
        # TODO: for coercing curies - quite clumsy; unify to_symbol and find_term?
        key_term = context.terms.get(p_key)
        if key_term and (key_term.type or key_term.container):
            p_key = p
        if not term and p == RDF.type and not self.use_rdf_type:
            if isinstance(o, URIRef):
                node = context.to_symbol(o)
            p_key = context.type_key

    if node is None:
        node = self.to_raw_value(graph, s, o, nodemap)

    value = s_node.get(p_key)
    if value:
        if not isinstance(value, list):
            value = [value]
        value.append(node)
    elif use_set:
        value = [node]
    else:
        value = node
    s_node[p_key] = value

convert

convert(graph: Graph)

Source code in rdflib/plugins/serializers/jsonld.py

def convert(self, graph: Graph):
    # TODO: bug in rdflib dataset parsing (nquads et al):
    # plain triples end up in separate unnamed graphs (rdflib issue #436)
    if graph.context_aware:
        # type error: "Graph" has no attribute "contexts"
        all_contexts = list(graph.contexts())  # type: ignore[attr-defined]
        has_dataset_default_id = any(
            c.identifier == DATASET_DEFAULT_GRAPH_ID for c in all_contexts
        )
        if (
            has_dataset_default_id
            # # type error: "Graph" has no attribute "contexts"
            and graph.default_context.identifier == DATASET_DEFAULT_GRAPH_ID  # type: ignore[attr-defined]
        ):
            default_graph = graph.default_context  # type: ignore[attr-defined]
        else:
            default_graph = Graph()
        graphs = [default_graph]
        default_graph_id = default_graph.identifier

        for g in all_contexts:
            if g in graphs:
                continue
            if isinstance(g.identifier, URIRef):
                graphs.append(g)
            else:
                default_graph += g
    else:
        graphs = [graph]
        default_graph_id = graph.identifier

    context = self.context

    objs: List[Any] = []
    for g in graphs:
        obj = {}
        graphname = None

        if isinstance(g.identifier, URIRef):
            if g.identifier != default_graph_id:
                graphname = context.shrink_iri(g.identifier)
                obj[context.id_key] = graphname

        nodes = self.from_graph(g)

        if not graphname and len(nodes) == 1:
            obj.update(nodes[0])
        else:
            if not nodes:
                continue
            obj[context.graph_key] = nodes

        if objs and objs[0].get(context.get_key(ID)) == graphname:
            objs[0].update(obj)
        else:
            objs.append(obj)

    if len(graphs) == 1 and len(objs) == 1 and not self.context.active:
        default = objs[0]
        items = default.get(context.graph_key)
        if len(default) == 1 and items:
            objs = items
    elif len(objs) == 1 and self.context.active:
        objs = objs[0]

    return objs

from_graph

from_graph(graph: Graph)

Source code in rdflib/plugins/serializers/jsonld.py

def from_graph(self, graph: Graph):
    nodemap: Dict[Any, Any] = {}

    for s in set(graph.subjects()):
        ## only iri:s and unreferenced (rest will be promoted to top if needed)
        if isinstance(s, URIRef) or (
            isinstance(s, BNode) and not any(graph.subjects(None, s))
        ):
            self.process_subject(graph, s, nodemap)

    return list(nodemap.values())

process_subject

process_subject(graph: Graph, s: IdentifiedNode, nodemap)

Source code in rdflib/plugins/serializers/jsonld.py

def process_subject(self, graph: Graph, s: IdentifiedNode, nodemap):
    if isinstance(s, URIRef):
        node_id = self.context.shrink_iri(s)
    elif isinstance(s, BNode):
        node_id = s.n3()
    else:
        # This does not seem right, this probably should be an error.
        node_id = None

    # used_as_object = any(graph.subjects(None, s))
    if node_id in nodemap:
        return None

    node = {}
    node[self.context.id_key] = node_id
    nodemap[node_id] = node

    for p, o in graph.predicate_objects(s):
        # type error: Argument 3 to "add_to_node" of "Converter" has incompatible type "Node"; expected "IdentifiedNode"
        # type error: Argument 4 to "add_to_node" of "Converter" has incompatible type "Node"; expected "Identifier"
        self.add_to_node(graph, s, p, o, node, nodemap)  # type: ignore[arg-type]

    return node

to_collection

to_collection(graph: Graph, l_: Identifier)

Source code in rdflib/plugins/serializers/jsonld.py

def to_collection(self, graph: Graph, l_: Identifier):
    if l_ != RDF.nil and not graph.value(l_, RDF.first):
        return None
    list_nodes: List[Optional[_ObjectType]] = []
    chain = set([l_])
    while l_:
        if l_ == RDF.nil:
            return list_nodes
        if isinstance(l_, URIRef):
            return None
        first, rest = None, None
        for p, o in graph.predicate_objects(l_):
            if not first and p == RDF.first:
                first = o
            elif not rest and p == RDF.rest:
                rest = o
            elif p != RDF.type or o != RDF.List:
                return None
        list_nodes.append(first)
        # type error: Incompatible types in assignment (expression has type "Optional[Node]", variable has type "Identifier")
        l_ = rest  # type: ignore[assignment]
        if l_ in chain:
            return None
        chain.add(l_)

to_raw_value

to_raw_value(graph: Graph, s: IdentifiedNode, o: Identifier, nodemap: Dict[str, Any])

Source code in rdflib/plugins/serializers/jsonld.py

def to_raw_value(
    self, graph: Graph, s: IdentifiedNode, o: Identifier, nodemap: Dict[str, Any]
):
    context = self.context
    coll = self.to_collection(graph, o)
    if coll is not None:
        coll = [
            self.to_raw_value(graph, s, lo, nodemap)
            for lo in self.to_collection(graph, o)
        ]
        return {context.list_key: coll}
    elif isinstance(o, BNode):
        embed = (
            False  # TODO: self.context.active or using startnode and only one ref
        )
        onode = self.process_subject(graph, o, nodemap)
        if onode:
            if embed and not any(s2 for s2 in graph.subjects(None, o) if s2 != s):
                return onode
            else:
                nodemap[onode[context.id_key]] = onode
        return {context.id_key: o.n3()}
    elif isinstance(o, URIRef):
        # TODO: embed if o != startnode (else reverse)
        return {context.id_key: context.shrink_iri(o)}
    elif isinstance(o, Literal):
        # TODO: if compact
        native = self.use_native_types and o.datatype in PLAIN_LITERAL_TYPES
        if native:
            v = o.toPython()
        else:
            v = str(o)
        if o.datatype:
            if native and self.context.active:
                return v
            return {
                context.type_key: context.to_symbol(o.datatype),
                context.value_key: v,
            }
        elif o.language and o.language != context.language:
            return {context.lang_key: o.language, context.value_key: v}
        # type error: Right operand of "and" is never evaluated
        elif not context.active or context.language and not o.language:  # type: ignore[unreachable]
            return {context.value_key: v}
        else:
            return v

type_coerce

type_coerce(o: Identifier, coerce_type: str)

Source code in rdflib/plugins/serializers/jsonld.py

def type_coerce(self, o: Identifier, coerce_type: str):
    if coerce_type == ID:
        if isinstance(o, URIRef):
            return self.context.shrink_iri(o)
        elif isinstance(o, BNode):
            return o.n3()
        else:
            return o
    elif coerce_type == VOCAB and isinstance(o, URIRef):
        return self.context.to_symbol(o)
    elif isinstance(o, Literal) and str(o.datatype) == coerce_type:
        return o
    else:
        return None

JsonLDSerializer

JsonLDSerializer(store: Graph)

Bases: Serializer

JSON-LD RDF graph serializer.

Methods:

• serialize –

Source code in rdflib/plugins/serializers/jsonld.py

def __init__(self, store: Graph):
    super(JsonLDSerializer, self).__init__(store)

serialize

serialize(stream: IO[bytes], base: Optional[str] = None, encoding: Optional[str] = None, **kwargs: Any) -> None

Source code in rdflib/plugins/serializers/jsonld.py

def serialize(
    self,
    stream: IO[bytes],
    base: Optional[str] = None,
    encoding: Optional[str] = None,
    **kwargs: Any,
) -> None:
    # TODO: docstring w. args and return value
    encoding = encoding or "utf-8"
    if encoding not in ("utf-8", "utf-16"):
        warnings.warn(
            "JSON should be encoded as unicode. " f"Given encoding was: {encoding}"
        )

    context_data = kwargs.get("context")
    use_native_types = (kwargs.get("use_native_types", False),)
    use_rdf_type = kwargs.get("use_rdf_type", False)
    auto_compact = kwargs.get("auto_compact", False)

    indent = kwargs.get("indent", 2)
    separators = kwargs.get("separators", (",", ": "))
    sort_keys = kwargs.get("sort_keys", True)
    ensure_ascii = kwargs.get("ensure_ascii", False)

    obj = from_rdf(
        self.store,
        context_data,
        base,
        use_native_types,
        use_rdf_type,
        auto_compact=auto_compact,
    )
    if _HAS_ORJSON:
        option: int = orjson.OPT_NON_STR_KEYS
        if indent is not None:
            option |= orjson.OPT_INDENT_2
        if sort_keys:
            option |= orjson.OPT_SORT_KEYS
        if ensure_ascii:
            warnings.warn("Cannot use ensure_ascii with orjson")
        data_bytes = orjson.dumps(obj, option=option)
        stream.write(data_bytes)
    else:
        data = json.dumps(
            obj,
            indent=indent,
            separators=separators,
            sort_keys=sort_keys,
            ensure_ascii=ensure_ascii,
        )
        stream.write(data.encode(encoding, "replace"))

from_rdf

from_rdf(graph, context_data=None, base=None, use_native_types=False, use_rdf_type=False, auto_compact=False, startnode=None, index=False)

Source code in rdflib/plugins/serializers/jsonld.py

def from_rdf(
    graph,
    context_data=None,
    base=None,
    use_native_types=False,
    use_rdf_type=False,
    auto_compact=False,
    startnode=None,
    index=False,
):
    # TODO: docstring w. args and return value
    # TODO: support for index and startnode

    if not context_data and auto_compact:
        context_data = dict(
            (pfx, str(ns))
            for (pfx, ns) in graph.namespaces()
            if pfx and str(ns) != "http://www.w3.org/XML/1998/namespace"
        )

    if isinstance(context_data, Context):
        context = context_data
        context_data = context.to_dict()
    else:
        context = Context(context_data, base=base)

    converter = Converter(context, use_native_types, use_rdf_type)
    result = converter.convert(graph)

    if converter.context.active:
        if isinstance(result, list):
            result = {context.get_key(GRAPH): result}
        result[CONTEXT] = context_data

    return result