15909158736

Committed 26 Jun 2025 06:03PM UTC coverage: 81.092% (-0.6%) from 81.702%

Build # 15909158736

Build Type

Pull #850

github

Committed by

web-flow

Commit Message

Merge pull request #875 from bodintsov/feature/share-cleanupgrade-2025-type-annotations

[ENG-7443] Feature/share cleanupgrade 2025 type annotations

Pull Request Pull Request #850: [project][ENG-7225] share clean(up)grade 2025 (milestone 2: upgrade)

Run Details

485 of 534 new or added lines in 63 files covered. (90.82%)

12 existing lines in 5 files now uncovered.

6150 of 7584 relevant lines covered (81.09%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.89

/trove/render/jsonld.py

from __future__ import annotations
import contextlib
import datetime
import json
from typing import Any, Iterator, TYPE_CHECKING

from primitive_metadata import primitive_rdf as rdf

from trove import exceptions as trove_exceptions
from trove.vocab.namespaces import RDF, OWL, TROVE
from trove.vocab import mediatypes
from ._base import BaseRenderer
if TYPE_CHECKING:
    from trove.util.json import (
        JsonObject,
        JsonValue,
    )


_PREDICATES_OF_FLEXIBLE_CARDINALITY = {
    # RDF.type,
    RDF.value,
}


class RdfJsonldRenderer(BaseRenderer):
    MEDIATYPE = mediatypes.JSONLD
    INDEXCARD_DERIVER_IRI = TROVE['derive/osfmap_json']

    __visiting_iris: set[str] | None = None

    def simple_render_document(self) -> str:
        return json.dumps(
            self.render_jsonld(self.response_data, self.response_focus.single_iri()),
            indent=2,
            sort_keys=True,
        )

    def render_jsonld(
        self,
        rdfgraph: rdf.RdfGraph,
        focus_iri: str,
        with_context: bool = False,
    ) -> JsonObject:
        with self.iri_shorthand.track_used_shorts() as _used_shorts:
            _rendered = self.rdfobject_as_jsonld(focus_iri, rdfgraph.tripledict)
        if with_context:
            _rendered['@context'] = {
                _shorthand_name: self.iri_shorthand.expand_iri(_shorthand_name)
                for _shorthand_name in _used_shorts
            }
        return _rendered

    def literal_as_jsonld(self, rdfliteral: rdf.Literal) -> JsonObject:
        if not rdfliteral.datatype_iris or rdfliteral.datatype_iris == {RDF.string}:
            return {'@value': rdfliteral.unicode_value}
        if RDF.JSON in rdfliteral.datatype_iris:
            # NOTE: does not reset jsonld context (is that a problem?)
            return json.loads(rdfliteral.unicode_value)
        _language_tag = rdfliteral.language
        if _language_tag:  # standard language tag
            return {
                '@value': rdfliteral.unicode_value,
                '@language': _language_tag,
            }
        # datatype iri (or non-standard language iri)
        _datatype_iris = [
            self.iri_shorthand.compact_iri(_datatype_iri)
            for _datatype_iri in rdfliteral.datatype_iris
        ]
        return {
            '@value': rdfliteral.unicode_value,
            '@type': (
                _datatype_iris
                if len(_datatype_iris) != 1
                else _datatype_iris[0]
            ),
        }

    def rdfobject_as_jsonld(
        self,
        rdfobject: rdf.RdfObject,
        tripledict: rdf.RdfTripleDictionary | None = None,
    ) -> JsonObject:
        if isinstance(rdfobject, str):
            return self.iri_as_jsonld(rdfobject, tripledict)
        elif isinstance(rdfobject, frozenset):
            if (RDF.type, RDF.Seq) in rdfobject:
                # TODO: jsonld has lists but not sequences -- switch to lists?
                return {'@list': [
                    self.rdfobject_as_jsonld(_sequence_obj, tripledict)
                    for _sequence_obj in rdf.sequence_objects_in_order(rdfobject)
                ]}
            return self.blanknode_as_jsonld(rdfobject, tripledict)
        elif isinstance(rdfobject, rdf.Literal):
            return self.literal_as_jsonld(rdfobject)
        elif isinstance(rdfobject, (float, int, datetime.date)):
            return self.literal_as_jsonld(rdf.literal(rdfobject))
        raise trove_exceptions.UnsupportedRdfObject(rdfobject)

    def blanknode_as_jsonld(
        self,
        blanknode: rdf.RdfBlanknode,
        tripledict: rdf.RdfTripleDictionary | None = None,
    ) -> JsonObject:
        _twopledict = rdf.twopledict_from_twopleset(blanknode)
        _jsonld = {}
        for _pred, _objectset in _twopledict.items():
            if _objectset:
                _key = self.iri_shorthand.compact_iri(_pred)
                _jsonld[_key] = self._list_or_single_value(_pred, [
                    self.rdfobject_as_jsonld(_obj, tripledict)
                    for _obj in _objectset
                ])
        return _jsonld

    def iri_as_jsonld(
        self,
        iri: str,
        tripledict: rdf.RdfTripleDictionary | None = None,
    ) -> JsonObject:
        if (not tripledict) or (iri not in tripledict) or self.__already_visiting(iri):
            return {'@id': self.iri_shorthand.compact_iri(iri)}
        with self.__visiting(iri):
            _nested_obj = (
                {}
                if iri.startswith('_:')  # HACK: non-blank blank nodes (stop that)
                else {'@id': self.iri_shorthand.compact_iri(iri)}
            )
            for _pred, _objectset in tripledict[iri].items():
                if _objectset:
                    _nested_obj[self.iri_shorthand.compact_iri(_pred)] = self._list_or_single_value(
                        _pred,
                        [  # indirect recursion:
                            self.rdfobject_as_jsonld(_obj, tripledict)
                            for _obj in _objectset
                        ],
                    )
            return _nested_obj

    def _list_or_single_value(self, predicate_iri: str, objectlist: list[JsonValue]) -> JsonValue:
        _only_one_object = (
            (predicate_iri, RDF.type, OWL.FunctionalProperty) in self.thesaurus
        )
        if _only_one_object:
            if len(objectlist) > 1:
                raise trove_exceptions.OwlObjection((
                    f'expected at most one object for <{predicate_iri}>'
                    f' (got {objectlist})'
                ))
            try:
                (_only_obj,) = objectlist
            except ValueError:
                return None
            else:
                return _only_obj
        if predicate_iri in _PREDICATES_OF_FLEXIBLE_CARDINALITY and len(objectlist) == 1:
            return objectlist[0]
        return sorted(objectlist, key=_naive_sort_key)

    @contextlib.contextmanager
    def __visiting(self, iri: str) -> Iterator[None]:
        if self.__visiting_iris is None:
            self.__visiting_iris = set()
        self.__visiting_iris.add(iri)
        yield
        self.__visiting_iris.discard(iri)

    def __already_visiting(self, iri: str) -> bool:
        return bool(self.__visiting_iris and (iri in self.__visiting_iris))


def _naive_sort_key(jsonable_obj: Any) -> tuple[int, str]:
    _json = json.dumps(jsonable_obj)
    return len(_json), _json

1	from __future__ import annotations	1✔
2	import contextlib	1✔
3	import datetime	1✔
4	import json	1✔
5	from typing import Any, Iterator, TYPE_CHECKING	1✔
6
7	from primitive_metadata import primitive_rdf as rdf	1✔
8
9	from trove import exceptions as trove_exceptions	1✔
10	from trove.vocab.namespaces import RDF, OWL, TROVE	1✔
11	from trove.vocab import mediatypes	1✔
12	from ._base import BaseRenderer	1✔
13	if TYPE_CHECKING:	1✔
NEW 14	from trove.util.json import (	×
15	JsonObject,
16	JsonValue,
17	)
18
19
20	_PREDICATES_OF_FLEXIBLE_CARDINALITY = {	1✔
21	# RDF.type,
22	RDF.value,
23	}
24
25
26	class RdfJsonldRenderer(BaseRenderer):	1✔
27	MEDIATYPE = mediatypes.JSONLD	1✔
28	INDEXCARD_DERIVER_IRI = TROVE['derive/osfmap_json']	1✔
29
30	__visiting_iris: set[str] \| None = None	1✔
31
32	def simple_render_document(self) -> str:	1✔
33	return json.dumps(	1✔
34	self.render_jsonld(self.response_data, self.response_focus.single_iri()),
35	indent=2,
36	sort_keys=True,
37	)
38
39	def render_jsonld(	1✔
40	self,
41	rdfgraph: rdf.RdfGraph,
42	focus_iri: str,
43	with_context: bool = False,
44	) -> JsonObject:
45	with self.iri_shorthand.track_used_shorts() as _used_shorts:	1✔
46	_rendered = self.rdfobject_as_jsonld(focus_iri, rdfgraph.tripledict)	1✔
47	if with_context:	1✔
48	_rendered['@context'] = {	×
49	_shorthand_name: self.iri_shorthand.expand_iri(_shorthand_name)
50	for _shorthand_name in _used_shorts
51	}
52	return _rendered	1✔
53
54	def literal_as_jsonld(self, rdfliteral: rdf.Literal) -> JsonObject:	1✔
55	if not rdfliteral.datatype_iris or rdfliteral.datatype_iris == {RDF.string}:	1✔
56	return {'@value': rdfliteral.unicode_value}	1✔
57	if RDF.JSON in rdfliteral.datatype_iris:	1✔
58	# NOTE: does not reset jsonld context (is that a problem?)
59	return json.loads(rdfliteral.unicode_value)	1✔
60	_language_tag = rdfliteral.language	1✔
61	if _language_tag: # standard language tag	1✔
62	return {	1✔
63	'@value': rdfliteral.unicode_value,
64	'@language': _language_tag,
65	}
66	# datatype iri (or non-standard language iri)
67	_datatype_iris = [	1✔
68	self.iri_shorthand.compact_iri(_datatype_iri)
69	for _datatype_iri in rdfliteral.datatype_iris
70	]
71	return {	1✔
72	'@value': rdfliteral.unicode_value,
73	'@type': (
74	_datatype_iris
75	if len(_datatype_iris) != 1
76	else _datatype_iris[0]
77	),
78	}
79
80	def rdfobject_as_jsonld(	1✔
81	self,
82	rdfobject: rdf.RdfObject,
83	tripledict: rdf.RdfTripleDictionary \| None = None,
84	) -> JsonObject:
85	if isinstance(rdfobject, str):	1✔
86	return self.iri_as_jsonld(rdfobject, tripledict)	1✔
87	elif isinstance(rdfobject, frozenset):	1✔
88	if (RDF.type, RDF.Seq) in rdfobject:	1✔
89	# TODO: jsonld has lists but not sequences -- switch to lists?
90	return {'@list': [	1✔
91	self.rdfobject_as_jsonld(_sequence_obj, tripledict)
92	for _sequence_obj in rdf.sequence_objects_in_order(rdfobject)
93	]}
94	return self.blanknode_as_jsonld(rdfobject, tripledict)	1✔
95	elif isinstance(rdfobject, rdf.Literal):	1✔
96	return self.literal_as_jsonld(rdfobject)	1✔
97	elif isinstance(rdfobject, (float, int, datetime.date)):	×
98	return self.literal_as_jsonld(rdf.literal(rdfobject))	×
99	raise trove_exceptions.UnsupportedRdfObject(rdfobject)	×
100
101	def blanknode_as_jsonld(	1✔
102	self,
103	blanknode: rdf.RdfBlanknode,
104	tripledict: rdf.RdfTripleDictionary \| None = None,
105	) -> JsonObject:
106	_twopledict = rdf.twopledict_from_twopleset(blanknode)	1✔
107	_jsonld = {}	1✔
108	for _pred, _objectset in _twopledict.items():	1✔
109	if _objectset:	1✔
110	_key = self.iri_shorthand.compact_iri(_pred)	1✔
111	_jsonld[_key] = self._list_or_single_value(_pred, [	1✔
112	self.rdfobject_as_jsonld(_obj, tripledict)
113	for _obj in _objectset
114	])
115	return _jsonld	1✔
116
117	def iri_as_jsonld(	1✔
118	self,
119	iri: str,
120	tripledict: rdf.RdfTripleDictionary \| None = None,
121	) -> JsonObject:
122	if (not tripledict) or (iri not in tripledict) or self.__already_visiting(iri):	1✔
123	return {'@id': self.iri_shorthand.compact_iri(iri)}	1✔
124	with self.__visiting(iri):	1✔
125	_nested_obj = (	1✔
126	{}
127	if iri.startswith('_:') # HACK: non-blank blank nodes (stop that)
128	else {'@id': self.iri_shorthand.compact_iri(iri)}
129	)
130	for _pred, _objectset in tripledict[iri].items():	1✔
131	if _objectset:	1✔
132	_nested_obj[self.iri_shorthand.compact_iri(_pred)] = self._list_or_single_value(	1✔
133	_pred,
134	[ # indirect recursion:
135	self.rdfobject_as_jsonld(_obj, tripledict)
136	for _obj in _objectset
137	],
138	)
139	return _nested_obj	1✔
140
141	def _list_or_single_value(self, predicate_iri: str, objectlist: list[JsonValue]) -> JsonValue:	1✔
142	_only_one_object = (	1✔
143	(predicate_iri, RDF.type, OWL.FunctionalProperty) in self.thesaurus
144	)
145	if _only_one_object:	1✔
146	if len(objectlist) > 1:	1✔
147	raise trove_exceptions.OwlObjection((	×
148	f'expected at most one object for <{predicate_iri}>'
149	f' (got {objectlist})'
150	))
151	try:	1✔
152	(_only_obj,) = objectlist	1✔
153	except ValueError:	×
154	return None	×
155	else:
156	return _only_obj	1✔
157	if predicate_iri in _PREDICATES_OF_FLEXIBLE_CARDINALITY and len(objectlist) == 1:	1✔
158	return objectlist[0]	×
159	return sorted(objectlist, key=_naive_sort_key)	1✔
160
161	@contextlib.contextmanager	1✔
162	def __visiting(self, iri: str) -> Iterator[None]:	1✔
163	if self.__visiting_iris is None:	1✔
164	self.__visiting_iris = set()	1✔
165	self.__visiting_iris.add(iri)	1✔
166	yield	1✔
167	self.__visiting_iris.discard(iri)	1✔
168
169	def __already_visiting(self, iri: str) -> bool:	1✔
170	return bool(self.__visiting_iris and (iri in self.__visiting_iris))	1✔
171
172
173	def _naive_sort_key(jsonable_obj: Any) -> tuple[int, str]:	1✔
174	_json = json.dumps(jsonable_obj)	1✔
175	return len(_json), _json	1✔

CenterForOpenScience / SHARE / 15909158736

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous