18378636854

Committed 09 Oct 2025 01:53PM UTC coverage: 84.286% (+2.6%) from 81.693%

Build # 18378636854

Build Type

Pull #886

github

Committed by

web-flow

Commit Message

Merge b8734c780 into 8f08e7f39

Pull Request Pull Request #886: Feat/9046 shtrove rss

Coverage Stats

509 of 600 new or added lines in 38 files covered. (84.83%)

1 existing line in 1 file now uncovered.

6565 of 7789 relevant lines covered (84.29%)

0.84 hits per line

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

77.78

/trove/render/simple_json.py

from __future__ import annotations
import json
import re
import typing

from primitive_metadata import primitive_rdf as rdf

from trove.vocab.jsonapi import (
    JSONAPI_LINK_OBJECT,
    JSONAPI_MEMBERNAME,
)
from trove.vocab import mediatypes
from trove.vocab.namespaces import TROVE, RDF
from .rendering import (
    ProtoRendering,
    SimpleRendering,
)
from .rendering.streamable import StreamableRendering
from ._simple_trovesearch import SimpleTrovesearchRenderer
if typing.TYPE_CHECKING:
    from collections.abc import (
        Generator,
        Iterator,
        Sequence,
    )
    from trove.util.json import JsonObject


class TrovesearchSimpleJsonRenderer(SimpleTrovesearchRenderer):
    '''for "simple json" search api -- very entangled with trove/trovesearch/trovesearch_gathering.py
    '''
    MEDIATYPE = mediatypes.JSON

    def unicard_rendering(self, card_iri: str, osfmap_json: JsonObject) -> ProtoRendering:
        return SimpleRendering(
            mediatype=self.MEDIATYPE,
            rendered_content=json.dumps({
                'data': self._render_card_content(card_iri, osfmap_json),
                'links': self._render_links(),
                'meta': self._render_meta(),
            }, indent=2),
        )

    def multicard_rendering(self, card_pages: Iterator[Sequence[tuple[str, JsonObject]]]) -> ProtoRendering:
        return StreamableRendering(
            mediatype=self.MEDIATYPE,
            content_stream=self._stream_json(card_pages),
        )

    def _stream_json(self, card_pages: Iterator[Sequence[tuple[str, JsonObject]]]) -> Generator[str]:
        _prefix = '{"data": ['
        yield _prefix
        _datum_prefix = None
        for _page in card_pages:
            for _card_iri, _osfmap_json in _page:
                if _datum_prefix is not None:
                    yield _datum_prefix
                yield json.dumps(self._render_card_content(_card_iri, _osfmap_json), indent=2)
                _datum_prefix = ','
        _nondata = json.dumps({
            'meta': self._render_meta(),
            'links': self._render_links(),
        })
        yield re.sub(
            '^{',  # replace the opening {
            '],',  # ...with a closing ] for the "data" list
            _nondata,
            count=1,
        )

    def _render_card_content(self, card_iri: str, osfmap_json: JsonObject) -> JsonObject:
        self._add_twople(osfmap_json, 'foaf:isPrimaryTopicOf', card_iri)
        return osfmap_json

    def _render_meta(self) -> dict[str, int | str]:
        _meta: dict[str, int | str] = {}
        try:
            _total = next(self.response_gathering.ask(
                TROVE.totalResultCount,
                focus=self.response_focus,
            ))
            if isinstance(_total, int):
                _meta['total'] = _total
            elif isinstance(_total, rdf.Literal):
                _meta['total'] = int(_total.unicode_value)
            elif _total == TROVE['ten-thousands-and-more']:
                _meta['total'] = 'trove:ten-thousands-and-more'
        except StopIteration:
            pass
        return _meta

    def _render_links(self) -> JsonObject:
        _links = {}
        for _pagelink in self._page_links:
            _twopledict = rdf.twopledict_from_twopleset(_pagelink)
            if JSONAPI_LINK_OBJECT in _twopledict.get(RDF.type, ()):
                (_membername,) = _twopledict[JSONAPI_MEMBERNAME]
                (_link_url,) = _twopledict[RDF.value]
                _links[_membername.unicode_value] = _link_url
        return _links

    def _add_twople(self, json_dict: JsonObject, predicate_iri: str, object_iri: str) -> None:
        _obj_ref: JsonObject = {'@id': object_iri}
        _obj_list = json_dict.setdefault(predicate_iri, [])
        if isinstance(_obj_list, list):
            _obj_list.append(_obj_ref)
        else:
            json_dict[predicate_iri] = [_obj_list, _obj_ref]

1	from __future__ import annotations	1✔
2	import json	1✔
3	import re	1✔
4	import typing	1✔
5
6	from primitive_metadata import primitive_rdf as rdf	1✔
7
8	from trove.vocab.jsonapi import (	1✔
9	JSONAPI_LINK_OBJECT,
10	JSONAPI_MEMBERNAME,
11	)
12	from trove.vocab import mediatypes	1✔
13	from trove.vocab.namespaces import TROVE, RDF	1✔
14	from .rendering import (	1✔
15	ProtoRendering,
16	SimpleRendering,
17	)
18	from .rendering.streamable import StreamableRendering	1✔
19	from ._simple_trovesearch import SimpleTrovesearchRenderer	1✔
20	if typing.TYPE_CHECKING:	1✔
NEW 21	from collections.abc import (	×
22	Generator,
23	Iterator,
24	Sequence,
25	)
UNCOV 26	from trove.util.json import JsonObject	×
27
28
29	class TrovesearchSimpleJsonRenderer(SimpleTrovesearchRenderer):	1✔
30	'''for "simple json" search api -- very entangled with trove/trovesearch/trovesearch_gathering.py
31	'''
32	MEDIATYPE = mediatypes.JSON	1✔
33
34	def unicard_rendering(self, card_iri: str, osfmap_json: JsonObject) -> ProtoRendering:	1✔
NEW 35	return SimpleRendering(	×
36	mediatype=self.MEDIATYPE,
37	rendered_content=json.dumps({
38	'data': self._render_card_content(card_iri, osfmap_json),
39	'links': self._render_links(),
40	'meta': self._render_meta(),
41	}, indent=2),
42	)
43
44	def multicard_rendering(self, card_pages: Iterator[Sequence[tuple[str, JsonObject]]]) -> ProtoRendering:	1✔
45	return StreamableRendering(	1✔
46	mediatype=self.MEDIATYPE,
47	content_stream=self._stream_json(card_pages),
48	)
49
50	def _stream_json(self, card_pages: Iterator[Sequence[tuple[str, JsonObject]]]) -> Generator[str]:	1✔
51	_prefix = '{"data": ['	1✔
52	yield _prefix	1✔
53	_datum_prefix = None	1✔
54	for _page in card_pages:	1✔
55	for _card_iri, _osfmap_json in _page:	1✔
56	if _datum_prefix is not None:	1✔
57	yield _datum_prefix	1✔
58	yield json.dumps(self._render_card_content(_card_iri, _osfmap_json), indent=2)	1✔
59	_datum_prefix = ','	1✔
60	_nondata = json.dumps({	1✔
61	'meta': self._render_meta(),
62	'links': self._render_links(),
63	})
64	yield re.sub(	1✔
65	'^{', # replace the opening {
66	'],', # ...with a closing ] for the "data" list
67	_nondata,
68	count=1,
69	)
70
71	def _render_card_content(self, card_iri: str, osfmap_json: JsonObject) -> JsonObject:	1✔
72	self._add_twople(osfmap_json, 'foaf:isPrimaryTopicOf', card_iri)	1✔
73	return osfmap_json	1✔
74
75	def _render_meta(self) -> dict[str, int \| str]:	1✔
76	_meta: dict[str, int \| str] = {}	1✔
77	try:	1✔
78	_total = next(self.response_gathering.ask(	1✔
79	TROVE.totalResultCount,
80	focus=self.response_focus,
81	))
82	if isinstance(_total, int):	1✔
83	_meta['total'] = _total	×
84	elif isinstance(_total, rdf.Literal):	1✔
85	_meta['total'] = int(_total.unicode_value)	1✔
86	elif _total == TROVE['ten-thousands-and-more']:	×
87	_meta['total'] = 'trove:ten-thousands-and-more'	×
88	except StopIteration:	×
89	pass	×
90	return _meta	1✔
91
92	def _render_links(self) -> JsonObject:	1✔
93	_links = {}	1✔
94	for _pagelink in self._page_links:	1✔
95	_twopledict = rdf.twopledict_from_twopleset(_pagelink)	×
96	if JSONAPI_LINK_OBJECT in _twopledict.get(RDF.type, ()):	×
97	(_membername,) = _twopledict[JSONAPI_MEMBERNAME]	×
98	(_link_url,) = _twopledict[RDF.value]	×
99	_links[_membername.unicode_value] = _link_url	×
100	return _links	1✔
101
102	def _add_twople(self, json_dict: JsonObject, predicate_iri: str, object_iri: str) -> None:	1✔
103	_obj_ref: JsonObject = {'@id': object_iri}	1✔
104	_obj_list = json_dict.setdefault(predicate_iri, [])	1✔
105	if isinstance(_obj_list, list):	1✔
106	_obj_list.append(_obj_ref)	1✔
107	else:
108	json_dict[predicate_iri] = [_obj_list, _obj_ref]	×

CenterForOpenScience / SHARE / 18378636854

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