14889044750

Committed 07 May 2025 05:00PM UTC coverage: 81.259% (+2.1%) from 79.141%

Build # 14889044750

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #867 from CenterForOpenScience/release/25.3.0

Release/25.3.0 share cleanupgrade (milestone 3: enhance)

Run Details

493 of 781 new or added lines in 34 files covered. (63.12%)

14 existing lines in 5 files now uncovered.

6196 of 7625 relevant lines covered (81.26%)

1.62 hits per line

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

92.82

/trove/render/jsonapi.py

import base64
from collections import defaultdict
import contextlib
import dataclasses
import datetime
import itertools
import json
import time
from typing import Iterable, Union

from primitive_metadata import primitive_rdf

from trove import exceptions as trove_exceptions
from trove.vocab.jsonapi import (
    JSONAPI_MEMBERNAME,
    JSONAPI_RELATIONSHIP,
    JSONAPI_ATTRIBUTE,
    JSONAPI_LINK_OBJECT,
)
from trove.vocab import mediatypes
from trove.vocab.namespaces import (
    OSFMAP,
    OWL,
    RDF,
    TROVE,
    XSD,
)
from trove.vocab.trove import trove_indexcard_namespace
from ._base import BaseRenderer


# a jsonapi resource may pull rdf data using an iri or blank node
# (using conventions from py for rdf as python primitives)
_IriOrBlanknode = Union[str, frozenset]


def _resource_ids_defaultdict():
    _prefix = str(time.time_ns())
    _ints = itertools.count()

    def _iter_ids():
        while True:
            _id = next(_ints)
            yield f'{_prefix}-{_id}'

    _ids = _iter_ids()
    return defaultdict(lambda: next(_ids))


@dataclasses.dataclass
class RdfJsonapiRenderer(BaseRenderer):
    '''render rdf data into jsonapi resources, guided by a given rdf vocabulary

    the given vocab describes how rdf predicates and classes in the data should
    map to jsonapi fields and resource objects in the rendered output, using
    `prefix jsonapi: <https://jsonapi.org/format/1.1/#>` and linked anchors in
    the jsonapi spec to represent jsonapi concepts:
      - jsonapi member name:
          `<iri> jsonapi:document-member-names "foo"@en`
      - jsonapi attribute:
          `<predicate_iri> rdf:type jsonapi:document-resource-object-attributes`
      - jsonapi relationship:
          `<predicate_iri> rdf:type jsonapi:document-resource-object-relationships`
      - to-one relationship or single-value attribute:
          `<predicate_iri> rdf:type owl:FunctionalProperty`

    note: does not support relationship links (or many other jsonapi features)
    '''
    MEDIATYPE = mediatypes.JSONAPI
    INDEXCARD_DERIVER_IRI = TROVE['derive/osfmap_json']

    _identifier_object_cache: dict = dataclasses.field(default_factory=dict)
    _id_namespace_set: Iterable[primitive_rdf.IriNamespace] = (trove_indexcard_namespace(),)
    __to_include: set[primitive_rdf.RdfObject] | None = None
    __assigned_blanknode_resource_ids: defaultdict[frozenset, str] = dataclasses.field(
        default_factory=_resource_ids_defaultdict,
        repr=False,
    )

    # override BaseRenderer
    @classmethod
    def get_deriver_iri(cls, card_blending: bool):
        return (None if card_blending else super().get_deriver_iri(card_blending))

    def simple_render_document(self) -> str:
        return json.dumps(
            self.render_dict(self.response_focus.single_iri()),
            indent=2,  # TODO: pretty-print query param?
        )

    def render_dict(self, primary_iris: Union[str, Iterable[str]]) -> dict:
        _primary_data: dict | list | None = None
        _included_data = []
        with self._contained__to_include() as _to_include:
            if isinstance(primary_iris, str):
                _already_included = {primary_iris}
                _primary_data = self.render_resource_object(primary_iris)
            else:
                _already_included = set(primary_iris)
                _primary_data = [
                    self.render_resource_object(_iri)
                    for _iri in primary_iris
                ]
            while _to_include:
                _next = _to_include.pop()
                if _next not in _already_included:
                    _already_included.add(_next)
                    _included_data.append(self.render_resource_object(_next))
        _document = {'data': _primary_data}
        if _included_data:
            _document['included'] = _included_data
        return _document

    def render_resource_object(self, iri_or_blanknode: _IriOrBlanknode) -> dict:
        _resource_object = {**self.render_identifier_object(iri_or_blanknode)}
        _twopledict = (
            (self.response_data.tripledict.get(iri_or_blanknode) or {})
            if isinstance(iri_or_blanknode, str)
            else primitive_rdf.twopledict_from_twopleset(iri_or_blanknode)
        )
        for _pred, _obj_set in _twopledict.items():
            if _pred != RDF.type:
                self._render_field(_pred, _obj_set, into=_resource_object)
        if isinstance(iri_or_blanknode, str):
            _resource_object.setdefault('links', {})['self'] = iri_or_blanknode
        return _resource_object

    def render_identifier_object(self, iri_or_blanknode: _IriOrBlanknode):
        try:
            return self._identifier_object_cache[iri_or_blanknode]
        except KeyError:
            if isinstance(iri_or_blanknode, str):
                _id_obj = {
                    '@id': self.iri_shorthand.compact_iri(iri_or_blanknode),
                }
                _type_iris = list(self.response_data.q(iri_or_blanknode, RDF.type))
                if _type_iris:
                    _id_obj = {
                        'id': self._resource_id_for_iri(iri_or_blanknode),
                        'type': self._single_typename(_type_iris),
                    }
            elif isinstance(iri_or_blanknode, frozenset):
                _type_iris = [
                    _obj
                    for _pred, _obj in iri_or_blanknode
                    if _pred == RDF.type
                ]
                _id_obj = {
                    'id': self._resource_id_for_blanknode(iri_or_blanknode),
                    'type': self._single_typename(_type_iris),
                }
            else:
                raise trove_exceptions.ExpectedIriOrBlanknode(f'expected str or frozenset (got {iri_or_blanknode})')
            self._identifier_object_cache[iri_or_blanknode] = _id_obj
            return _id_obj

    def _single_typename(self, type_iris: list[str]):
        if not type_iris:
            return ''
        if len(type_iris) == 1:
            return self._membername_for_iri(type_iris[0])
        # choose one predictably, preferring osfmap and trove
        for _namespace in (OSFMAP, TROVE):
            _type_iris = sorted(_iri for _iri in type_iris if _iri in _namespace)
            if _type_iris:
                return self._membername_for_iri(_type_iris[0])
        return self._membername_for_iri(sorted(type_iris)[0])

    def _membername_for_iri(self, iri: str):
        try:
            _membername = next(self.thesaurus.q(iri, JSONAPI_MEMBERNAME))
        except StopIteration:
            pass
        else:
            if isinstance(_membername, primitive_rdf.Literal):
                return _membername.unicode_value
            raise trove_exceptions.ExpectedLiteralObject((iri, JSONAPI_MEMBERNAME, _membername))
        return self.iri_shorthand.compact_iri(iri)

    def _resource_id_for_blanknode(self, blanknode: frozenset, /):
        return self.__assigned_blanknode_resource_ids[blanknode]

    def _resource_id_for_iri(self, iri: str):
        for _iri_namespace in self._id_namespace_set:
            if iri in _iri_namespace:
                return primitive_rdf.iri_minus_namespace(iri, namespace=_iri_namespace)
        # check for a shorthand
        _compact = self.iri_shorthand.compact_iri(iri)
        if _compact != iri:
            return _compact
        # as fallback, encode the iri into a valid jsonapi member name
        return base64.urlsafe_b64encode(iri.encode()).decode()

    def _render_field(self, predicate_iri, object_set, *, into: dict):
        _is_relationship = (predicate_iri, RDF.type, JSONAPI_RELATIONSHIP) in self.thesaurus
        _is_attribute = (predicate_iri, RDF.type, JSONAPI_ATTRIBUTE) in self.thesaurus
        _field_key = self._membername_for_iri(predicate_iri)
        _doc_key = 'meta'  # unless configured for jsonapi, default to unstructured 'meta'
        if ':' not in _field_key:
            if _is_relationship:
                _doc_key = 'relationships'
            elif _is_attribute:
                _doc_key = 'attributes'
        if _is_relationship:
            _fieldvalue = self._render_relationship_object(predicate_iri, object_set)
        else:
            _fieldvalue = self._one_or_many(predicate_iri, self._attribute_datalist(object_set))
        # update the given `into` resource object
        into.setdefault(_doc_key, {})[_field_key] = _fieldvalue

    def _one_or_many(self, predicate_iri: str, datalist: list):
        _only_one = (predicate_iri, RDF.type, OWL.FunctionalProperty) in self.thesaurus
        if _only_one:
            if len(datalist) > 1:
                raise trove_exceptions.OwlObjection(f'multiple objects for to-one relation <{predicate_iri}>: {datalist}')
            return (datalist[0] if datalist else None)
        return datalist

    def _attribute_datalist(self, object_set):
        return [
            self._render_attribute_datum(_obj)
            for _obj in object_set
        ]

    def _render_relationship_object(self, predicate_iri, object_set):
        _data = []
        _links = {}
        for _obj in object_set:
            if isinstance(_obj, frozenset):
                if (RDF.type, RDF.Seq) in _obj:
                    for _seq_obj in primitive_rdf.sequence_objects_in_order(_obj):
                        _data.append(self.render_identifier_object(_seq_obj))
                        self._pls_include(_seq_obj)
                elif (RDF.type, JSONAPI_LINK_OBJECT) in _obj:
                    _key, _link_obj = self._render_link_object(_obj)
                    _links[_key] = _link_obj
                else:
                    _data.append(self.render_identifier_object(_obj))
                    self._pls_include(_obj)
            else:
                assert isinstance(_obj, str)
                _data.append(self.render_identifier_object(_obj))
                self._pls_include(_obj)
        _relationship_obj = {
            'data': self._one_or_many(predicate_iri, _data),
        }
        if _links:
            _relationship_obj['links'] = _links
        return _relationship_obj

    def _render_link_object(self, link_obj: frozenset):
        _membername = next(
            _obj.unicode_value
            for _pred, _obj in link_obj
            if _pred == JSONAPI_MEMBERNAME
        )
        _rendered_link = {
            'href': next(
                _obj
                for _pred, _obj in link_obj
                if _pred == RDF.value
            ),
            # TODO:
            # 'rel':
            # 'describedby':
            # 'title':
            # 'type':
            # 'hreflang':
            # 'meta':
        }
        return _membername, _rendered_link

    def _make_object_gen(self, object_set):
        for _obj in object_set:
            if isinstance(_obj, frozenset) and ((RDF.type, RDF.Seq) in _obj):
                yield from primitive_rdf.sequence_objects_in_order(_obj)
            else:
                yield _obj

    @contextlib.contextmanager
    def _contained__to_include(self):
        assert self.__to_include is None
        self.__to_include = set()
        try:
            yield self.__to_include
        finally:
            self.__to_include = None

    def _pls_include(self, item):
        if self.__to_include is not None:
            self.__to_include.add(item)

    def _render_attribute_datum(self, rdfobject: primitive_rdf.RdfObject) -> dict | list | str | float | int:
        if isinstance(rdfobject, frozenset):
            if (RDF.type, RDF.Seq) in rdfobject:
                return [
                    self._render_attribute_datum(_seq_obj)
                    for _seq_obj in primitive_rdf.sequence_objects_in_order(rdfobject)
                ]
            _json_blanknode = {}
            for _pred, _obj_set in primitive_rdf.twopledict_from_twopleset(rdfobject).items():
                _key = self._membername_for_iri(_pred)
                _json_blanknode[_key] = self._one_or_many(_pred, self._attribute_datalist(_obj_set))
            return _json_blanknode
        if isinstance(rdfobject, primitive_rdf.Literal):
            if RDF.JSON in rdfobject.datatype_iris:
                return json.loads(rdfobject.unicode_value)
            if XSD.integer in rdfobject.datatype_iris:
                return int(rdfobject.unicode_value)
            return rdfobject.unicode_value  # TODO: decide how to represent language
        elif isinstance(rdfobject, str):
            return self.render_identifier_object(rdfobject)
        elif isinstance(rdfobject, (float, int)):
            return rdfobject
        elif isinstance(rdfobject, datetime.date):
            # just "YYYY-MM-DD"
            return datetime.date.isoformat(rdfobject)
        raise trove_exceptions.UnsupportedRdfObject(rdfobject)

1	import base64	2✔
2	from collections import defaultdict	2✔
3	import contextlib	2✔
4	import dataclasses	2✔
5	import datetime	2✔
6	import itertools	2✔
7	import json	2✔
8	import time	2✔
9	from typing import Iterable, Union	2✔
10
11	from primitive_metadata import primitive_rdf	2✔
12
13	from trove import exceptions as trove_exceptions	2✔
14	from trove.vocab.jsonapi import (	2✔
15	JSONAPI_MEMBERNAME,
16	JSONAPI_RELATIONSHIP,
17	JSONAPI_ATTRIBUTE,
18	JSONAPI_LINK_OBJECT,
19	)
20	from trove.vocab import mediatypes	2✔
21	from trove.vocab.namespaces import (	2✔
22	OSFMAP,
23	OWL,
24	RDF,
25	TROVE,
26	XSD,
27	)
28	from trove.vocab.trove import trove_indexcard_namespace	2✔
29	from ._base import BaseRenderer	2✔
30
31
32	# a jsonapi resource may pull rdf data using an iri or blank node
33	# (using conventions from py for rdf as python primitives)
34	_IriOrBlanknode = Union[str, frozenset]	2✔
35
36
37	def _resource_ids_defaultdict():	2✔
38	_prefix = str(time.time_ns())	2✔
39	_ints = itertools.count()	2✔
40
41	def _iter_ids():	2✔
42	while True:	2✔
43	_id = next(_ints)	2✔
44	yield f'{_prefix}-{_id}'	2✔
45
46	_ids = _iter_ids()	2✔
47	return defaultdict(lambda: next(_ids))	2✔
48
49
50	@dataclasses.dataclass	2✔
51	class RdfJsonapiRenderer(BaseRenderer):	2✔
52	'''render rdf data into jsonapi resources, guided by a given rdf vocabulary
53
54	the given vocab describes how rdf predicates and classes in the data should
55	map to jsonapi fields and resource objects in the rendered output, using
56	`prefix jsonapi: <https://jsonapi.org/format/1.1/#>` and linked anchors in
57	the jsonapi spec to represent jsonapi concepts:
58	- jsonapi member name:
59	`<iri> jsonapi:document-member-names "foo"@en`
60	- jsonapi attribute:
61	`<predicate_iri> rdf:type jsonapi:document-resource-object-attributes`
62	- jsonapi relationship:
63	`<predicate_iri> rdf:type jsonapi:document-resource-object-relationships`
64	- to-one relationship or single-value attribute:
65	`<predicate_iri> rdf:type owl:FunctionalProperty`
66
67	note: does not support relationship links (or many other jsonapi features)
68	'''
69	MEDIATYPE = mediatypes.JSONAPI	2✔
70	INDEXCARD_DERIVER_IRI = TROVE['derive/osfmap_json']	2✔
71
72	_identifier_object_cache: dict = dataclasses.field(default_factory=dict)	2✔
73	_id_namespace_set: Iterable[primitive_rdf.IriNamespace] = (trove_indexcard_namespace(),)	2✔
74	__to_include: set[primitive_rdf.RdfObject] \| None = None	2✔
75	__assigned_blanknode_resource_ids: defaultdict[frozenset, str] = dataclasses.field(	2✔
76	default_factory=_resource_ids_defaultdict,
77	repr=False,
78	)
79
80	# override BaseRenderer
81	@classmethod	2✔
82	def get_deriver_iri(cls, card_blending: bool):	2✔
83	return (None if card_blending else super().get_deriver_iri(card_blending))	2✔
84
85	def simple_render_document(self) -> str:	2✔
86	return json.dumps(	2✔
87	self.render_dict(self.response_focus.single_iri()),
88	indent=2, # TODO: pretty-print query param?
89	)
90
91	def render_dict(self, primary_iris: Union[str, Iterable[str]]) -> dict:	2✔
92	_primary_data: dict \| list \| None = None	2✔
93	_included_data = []	2✔
94	with self._contained__to_include() as _to_include:	2✔
95	if isinstance(primary_iris, str):	2✔
96	_already_included = {primary_iris}	2✔
97	_primary_data = self.render_resource_object(primary_iris)	2✔
98	else:
99	_already_included = set(primary_iris)	×
100	_primary_data = [	×
101	self.render_resource_object(_iri)
102	for _iri in primary_iris
103	]
104	while _to_include:	2✔
105	_next = _to_include.pop()	2✔
106	if _next not in _already_included:	2✔
107	_already_included.add(_next)	2✔
108	_included_data.append(self.render_resource_object(_next))	2✔
109	_document = {'data': _primary_data}	2✔
110	if _included_data:	2✔
111	_document['included'] = _included_data	2✔
112	return _document	2✔
113
114	def render_resource_object(self, iri_or_blanknode: _IriOrBlanknode) -> dict:	2✔
115	_resource_object = {**self.render_identifier_object(iri_or_blanknode)}	2✔
116	_twopledict = (	2✔
117	(self.response_data.tripledict.get(iri_or_blanknode) or {})
118	if isinstance(iri_or_blanknode, str)
119	else primitive_rdf.twopledict_from_twopleset(iri_or_blanknode)
120	)
121	for _pred, _obj_set in _twopledict.items():	2✔
122	if _pred != RDF.type:	2✔
123	self._render_field(_pred, _obj_set, into=_resource_object)	2✔
124	if isinstance(iri_or_blanknode, str):	2✔
125	_resource_object.setdefault('links', {})['self'] = iri_or_blanknode	2✔
126	return _resource_object	2✔
127
128	def render_identifier_object(self, iri_or_blanknode: _IriOrBlanknode):	2✔
129	try:	2✔
130	return self._identifier_object_cache[iri_or_blanknode]	2✔
131	except KeyError:	2✔
132	if isinstance(iri_or_blanknode, str):	2✔
133	_id_obj = {	2✔
134	'@id': self.iri_shorthand.compact_iri(iri_or_blanknode),
135	}
136	_type_iris = list(self.response_data.q(iri_or_blanknode, RDF.type))	2✔
137	if _type_iris:	2✔
138	_id_obj = {	2✔
139	'id': self._resource_id_for_iri(iri_or_blanknode),
140	'type': self._single_typename(_type_iris),
141	}
142	elif isinstance(iri_or_blanknode, frozenset):	2✔
143	_type_iris = [	2✔
144	_obj
145	for _pred, _obj in iri_or_blanknode
146	if _pred == RDF.type
147	]
148	_id_obj = {	2✔
149	'id': self._resource_id_for_blanknode(iri_or_blanknode),
150	'type': self._single_typename(_type_iris),
151	}
152	else:
153	raise trove_exceptions.ExpectedIriOrBlanknode(f'expected str or frozenset (got {iri_or_blanknode})')	×
154	self._identifier_object_cache[iri_or_blanknode] = _id_obj	2✔
155	return _id_obj	2✔
156
157	def _single_typename(self, type_iris: list[str]):	2✔
158	if not type_iris:	2✔
NEW 159	return ''	×
160	if len(type_iris) == 1:	2✔
161	return self._membername_for_iri(type_iris[0])	2✔
162	# choose one predictably, preferring osfmap and trove
163	for _namespace in (OSFMAP, TROVE):	2✔
164	_type_iris = sorted(_iri for _iri in type_iris if _iri in _namespace)	2✔
165	if _type_iris:	2✔
166	return self._membername_for_iri(_type_iris[0])	2✔
167	return self._membername_for_iri(sorted(type_iris)[0])	×
168
169	def _membername_for_iri(self, iri: str):	2✔
170	try:	2✔
171	_membername = next(self.thesaurus.q(iri, JSONAPI_MEMBERNAME))	2✔
172	except StopIteration:	2✔
173	pass	2✔
174	else:
175	if isinstance(_membername, primitive_rdf.Literal):	2✔
176	return _membername.unicode_value	2✔
177	raise trove_exceptions.ExpectedLiteralObject((iri, JSONAPI_MEMBERNAME, _membername))	×
178	return self.iri_shorthand.compact_iri(iri)	2✔
179
180	def _resource_id_for_blanknode(self, blanknode: frozenset, /):	2✔
181	return self.__assigned_blanknode_resource_ids[blanknode]	2✔
182
183	def _resource_id_for_iri(self, iri: str):	2✔
184	for _iri_namespace in self._id_namespace_set:	2✔
185	if iri in _iri_namespace:	2✔
186	return primitive_rdf.iri_minus_namespace(iri, namespace=_iri_namespace)	2✔
187	# check for a shorthand
188	_compact = self.iri_shorthand.compact_iri(iri)	2✔
189	if _compact != iri:	2✔
190	return _compact	2✔
191	# as fallback, encode the iri into a valid jsonapi member name
192	return base64.urlsafe_b64encode(iri.encode()).decode()	2✔
193
194	def _render_field(self, predicate_iri, object_set, *, into: dict):	2✔
195	_is_relationship = (predicate_iri, RDF.type, JSONAPI_RELATIONSHIP) in self.thesaurus	2✔
196	_is_attribute = (predicate_iri, RDF.type, JSONAPI_ATTRIBUTE) in self.thesaurus	2✔
197	_field_key = self._membername_for_iri(predicate_iri)	2✔
198	_doc_key = 'meta' # unless configured for jsonapi, default to unstructured 'meta'	2✔
199	if ':' not in _field_key:	2✔
200	if _is_relationship:	2✔
201	_doc_key = 'relationships'	2✔
202	elif _is_attribute:	2✔
203	_doc_key = 'attributes'	2✔
204	if _is_relationship:	2✔
205	_fieldvalue = self._render_relationship_object(predicate_iri, object_set)	2✔
206	else:
207	_fieldvalue = self._one_or_many(predicate_iri, self._attribute_datalist(object_set))	2✔
208	# update the given `into` resource object
209	into.setdefault(_doc_key, {})[_field_key] = _fieldvalue	2✔
210
211	def _one_or_many(self, predicate_iri: str, datalist: list):	2✔
212	_only_one = (predicate_iri, RDF.type, OWL.FunctionalProperty) in self.thesaurus	2✔
213	if _only_one:	2✔
214	if len(datalist) > 1:	2✔
215	raise trove_exceptions.OwlObjection(f'multiple objects for to-one relation <{predicate_iri}>: {datalist}')	×
216	return (datalist[0] if datalist else None)	2✔
217	return datalist	2✔
218
219	def _attribute_datalist(self, object_set):	2✔
220	return [	2✔
221	self._render_attribute_datum(_obj)
222	for _obj in object_set
223	]
224
225	def _render_relationship_object(self, predicate_iri, object_set):	2✔
226	_data = []	2✔
227	_links = {}	2✔
228	for _obj in object_set:	2✔
229	if isinstance(_obj, frozenset):	2✔
230	if (RDF.type, RDF.Seq) in _obj:	2✔
231	for _seq_obj in primitive_rdf.sequence_objects_in_order(_obj):	2✔
232	_data.append(self.render_identifier_object(_seq_obj))	2✔
233	self._pls_include(_seq_obj)	2✔
234	elif (RDF.type, JSONAPI_LINK_OBJECT) in _obj:	2✔
235	_key, _link_obj = self._render_link_object(_obj)	2✔
236	_links[_key] = _link_obj	2✔
237	else:
238	_data.append(self.render_identifier_object(_obj))	2✔
239	self._pls_include(_obj)	2✔
240	else:
241	assert isinstance(_obj, str)	2✔
242	_data.append(self.render_identifier_object(_obj))	2✔
243	self._pls_include(_obj)	2✔
244	_relationship_obj = {	2✔
245	'data': self._one_or_many(predicate_iri, _data),
246	}
247	if _links:	2✔
248	_relationship_obj['links'] = _links	2✔
249	return _relationship_obj	2✔
250
251	def _render_link_object(self, link_obj: frozenset):	2✔
252	_membername = next(	2✔
253	_obj.unicode_value
254	for _pred, _obj in link_obj
255	if _pred == JSONAPI_MEMBERNAME
256	)
257	_rendered_link = {	2✔
258	'href': next(
259	_obj
260	for _pred, _obj in link_obj
261	if _pred == RDF.value
262	),
263	# TODO:
264	# 'rel':
265	# 'describedby':
266	# 'title':
267	# 'type':
268	# 'hreflang':
269	# 'meta':
270	}
271	return _membername, _rendered_link	2✔
272
273	def _make_object_gen(self, object_set):	2✔
274	for _obj in object_set:	×
275	if isinstance(_obj, frozenset) and ((RDF.type, RDF.Seq) in _obj):	×
276	yield from primitive_rdf.sequence_objects_in_order(_obj)	×
277	else:
278	yield _obj	×
279
280	@contextlib.contextmanager	2✔
281	def _contained__to_include(self):	2✔
282	assert self.__to_include is None	2✔
283	self.__to_include = set()	2✔
284	try:	2✔
285	yield self.__to_include	2✔
286	finally:
287	self.__to_include = None	2✔
288
289	def _pls_include(self, item):	2✔
290	if self.__to_include is not None:	2✔
291	self.__to_include.add(item)	2✔
292
293	def _render_attribute_datum(self, rdfobject: primitive_rdf.RdfObject) -> dict \| list \| str \| float \| int:	2✔
294	if isinstance(rdfobject, frozenset):	2✔
295	if (RDF.type, RDF.Seq) in rdfobject:	2✔
296	return [	2✔
297	self._render_attribute_datum(_seq_obj)
298	for _seq_obj in primitive_rdf.sequence_objects_in_order(rdfobject)
299	]
300	_json_blanknode = {}	2✔
301	for _pred, _obj_set in primitive_rdf.twopledict_from_twopleset(rdfobject).items():	2✔
302	_key = self._membername_for_iri(_pred)	2✔
303	_json_blanknode[_key] = self._one_or_many(_pred, self._attribute_datalist(_obj_set))	2✔
304	return _json_blanknode	2✔
305	if isinstance(rdfobject, primitive_rdf.Literal):	2✔
306	if RDF.JSON in rdfobject.datatype_iris:	2✔
307	return json.loads(rdfobject.unicode_value)	2✔
308	if XSD.integer in rdfobject.datatype_iris:	2✔
309	return int(rdfobject.unicode_value)	2✔
310	return rdfobject.unicode_value # TODO: decide how to represent language	2✔
311	elif isinstance(rdfobject, str):	2✔
312	return self.render_identifier_object(rdfobject)	2✔
313	elif isinstance(rdfobject, (float, int)):	2✔
314	return rdfobject	2✔
315	elif isinstance(rdfobject, datetime.date):	×
316	# just "YYYY-MM-DD"
317	return datetime.date.isoformat(rdfobject)	×
318	raise trove_exceptions.UnsupportedRdfObject(rdfobject)	×

CenterForOpenScience / SHARE / 14889044750

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