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

79.03

/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 StreamableRendering, ProtoRendering
from ._simple_trovesearch import SimpleTrovesearchRenderer
if typing.TYPE_CHECKING:
    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
    INDEXCARD_DERIVER_IRI = TROVE['derive/osfmap_json']

    def simple_unicard_rendering(self, card_iri: str, osfmap_json: dict[str, typing.Any]) -> str:
        return 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: typing.Iterator[dict[str, dict[str, typing.Any]]]) -> ProtoRendering:
        return StreamableRendering(  # type: ignore[return-value]
            mediatype=self.MEDIATYPE,
            content_stream=self._stream_json(card_pages),
        )

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

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