diff --git a/case_example_mapping/__init__.py b/case_example_mapping/__init__.py index f102a9c..b84e288 100644 --- a/case_example_mapping/__init__.py +++ b/case_example_mapping/__init__.py @@ -1 +1,5 @@ __version__ = "0.0.1" + +from typing import Union + +JSON = Union[None, bool, dict[str, "JSON"], float, int, list["JSON"], str] diff --git a/case_example_mapping/rdf_to_case.py b/case_example_mapping/rdf_to_case.py index cf5a86f..eaf8ee1 100644 --- a/case_example_mapping/rdf_to_case.py +++ b/case_example_mapping/rdf_to_case.py @@ -1,8 +1,69 @@ +import json import sys +from io import TextIOWrapper from rdflib import Graph, Literal, Namespace, URIRef from rdflib.namespace import RDF, XSD +from case_example_mapping import JSON + +STR_XSD_STRING = str(XSD.string) + + +def compact_xsd_strings(json_object: JSON) -> JSON: + """ + Some JSON-LD serializers export all literals with datatype + annotations. A string-literal without a datatype can be optionally + exported with a datatype annotation of ``xsd:string``, or left to be + interpreted as a ``xsd:string`` by default. This function compacts + string-literals that are ``xsd:string``-typed into a JSON String + instead of a JSON Object. + + >>> from rdflib import XSD + >>> x = { + ... "@id": "http://example.org/s", + ... "http://example.org/p": [ + ... { + ... "@type": str(XSD.string), + ... "@value": "o" + ... } + ... ] + ... } + >>> x + {'@id': 'http://example.org/s', 'http://example.org/p': [{'@type': 'http://www.w3.org/2001/XMLSchema#string', '@value': 'o'}]} + >>> compact_xsd_strings(x) + {'@id': 'http://example.org/s', 'http://example.org/p': ['o']} + """ + if json_object is None: + return json_object + elif isinstance(json_object, (bool, float, int, str)): + return json_object + elif isinstance(json_object, dict): + if "@value" in json_object: + # Reviewing Literal. + if json_object.get("@type") == STR_XSD_STRING: + assert isinstance(json_object["@value"], str) + return json_object["@value"] + else: + return json_object + else: + return {k: compact_xsd_strings(json_object[k]) for k in json_object.keys()} + elif isinstance(json_object, list): + return [compact_xsd_strings(x) for x in json_object] + else: + raise TypeError("Unexpected type of object: %s." % type(json_object)) + + +def serialize_jsonld(graph: Graph, fh: TextIOWrapper) -> None: + """ + This function serializes the graph with a step taken to compact + ``xsd:string`` Literals that became JSON Objects. + """ + json_string = graph.serialize(format="json-ld") + json_object = json.loads(json_string) + json_object = compact_xsd_strings(json_object) + json.dump(json_object, fh, indent=4) + def main() -> None: # Process the command line arguments to get the output path @@ -52,7 +113,7 @@ def main() -> None: # Write the CASE graph to a file try: with open(output_path, "w") as case_file: - case_file.write(g.serialize(format="json-ld", indent=4)) + serialize_jsonld(g, case_file) print(f"CASE graph exported to: {output_path}") except IOError: print(f"Error writing to path: {output_path}")