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)

Coverage Stats

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

72.13

/trove/render/_simple_trovesearch.py

from __future__ import annotations
from collections.abc import Generator, Iterator
import json
from typing import Any, TYPE_CHECKING

from primitive_metadata import primitive_rdf as rdf

from trove import exceptions as trove_exceptions
from trove.vocab.jsonapi import JSONAPI_LINK_OBJECT
from trove.vocab.namespaces import TROVE, RDF
from ._base import BaseRenderer
from ._rendering import ProtoRendering, SimpleRendering
if TYPE_CHECKING:
    from trove.util.json import JsonObject


class SimpleTrovesearchRenderer(BaseRenderer):
    '''for "simple" search api responses (including only result metadata)

    (very entangled with trove/trovesearch/trovesearch_gathering.py)
    '''
    PASSIVE_RENDER = False  # knows the properties it cares about
    _page_links: set[str]
    __already_iterated_cards = False

    def simple_unicard_rendering(self, card_iri: str, osfmap_json: JsonObject) -> str:
        raise NotImplementedError

    def simple_multicard_rendering(self, cards: Iterator[tuple[str, JsonObject]]) -> str:
        raise NotImplementedError

    def unicard_rendering(self, card_iri: str, osfmap_json: JsonObject) -> ProtoRendering:
        return SimpleRendering(  # type: ignore[return-value]
            mediatype=self.MEDIATYPE,
            rendered_content=self.simple_unicard_rendering(card_iri, osfmap_json),
        )

    def multicard_rendering(self, card_pages: Iterator[dict[str, JsonObject]]) -> ProtoRendering:
        _cards = (
            (_card_iri, _card_contents)
            for _page in card_pages
            for _card_iri, _card_contents in _page.items()
        )
        return SimpleRendering(  # type: ignore[return-value]
            mediatype=self.MEDIATYPE,
            rendered_content=self.simple_multicard_rendering(_cards),
        )

    def render_document(self) -> ProtoRendering:
        _focustypes = self.response_focus.type_iris
        if (TROVE.Cardsearch in _focustypes) or (TROVE.Valuesearch in _focustypes):
            return self.multicard_rendering(self._iter_card_pages())
        if TROVE.Indexcard in _focustypes:
            return self.unicard_rendering(
                self.response_focus.single_iri(),
                self._get_card_content(self.response_focus.single_iri()),
            )
        raise trove_exceptions.UnsupportedRdfType(_focustypes)

    def _iter_card_pages(self) -> Generator[dict[str, JsonObject]]:
        assert not self.__already_iterated_cards
        self.__already_iterated_cards = True
        self._page_links = set()
        for _page, _page_graph in self.response_gathering.ask_exhaustively(
            TROVE.searchResultPage, focus=self.response_focus
        ):
            if (RDF.type, JSONAPI_LINK_OBJECT) in _page:
                self._page_links.add(_page)
            elif rdf.is_container(_page):
                _cardpage = []
                for _search_result in rdf.container_objects(_page):
                    try:
                        _card = next(
                            _obj
                            for _pred, _obj in _search_result
                            if _pred == TROVE.indexCard
                        )
                    except StopIteration:
                        pass  # skip malformed
                    else:
                        _cardpage.append(_card)
                yield {
                    self._get_card_iri(_card): self._get_card_content(_card, _page_graph)
                    for _card in _cardpage
                }

    def _get_card_iri(self, card: str | rdf.RdfBlanknode) -> str:
        return card if isinstance(card, str) else ''

    def _get_card_content(
        self,
        card: str | rdf.RdfBlanknode,
        graph: rdf.RdfGraph | None = None,
    ) -> Any:
        if isinstance(card, str):
            _card_content = (
                next(self.response_gathering.ask(TROVE.resourceMetadata, focus=card))
                if graph is None
                else next(graph.q(card, TROVE.resourceMetadata), None)
            )
        elif isinstance(card, frozenset):
            _card_content = next(
                _obj
                for _pred, _obj in card
                if _pred == TROVE.resourceMetadata
            )
        else:
            raise trove_exceptions.ExpectedIriOrBlanknode(card)
        if isinstance(_card_content, rdf.QuotedGraph):
            return _card_content
        if isinstance(_card_content, rdf.Literal) and (RDF.JSON in _card_content.datatype_iris):
            return json.loads(_card_content.unicode_value)
        raise ValueError(card)

1	from __future__ import annotations	1✔
2	from collections.abc import Generator, Iterator	1✔
3	import json	1✔
4	from typing import Any, TYPE_CHECKING	1✔
5
6	from primitive_metadata import primitive_rdf as rdf	1✔
7
8	from trove import exceptions as trove_exceptions	1✔
9	from trove.vocab.jsonapi import JSONAPI_LINK_OBJECT	1✔
10	from trove.vocab.namespaces import TROVE, RDF	1✔
11	from ._base import BaseRenderer	1✔
12	from ._rendering import ProtoRendering, SimpleRendering	1✔
13	if TYPE_CHECKING:	1✔
NEW 14	from trove.util.json import JsonObject	×
15
16
17	class SimpleTrovesearchRenderer(BaseRenderer):	1✔
18	'''for "simple" search api responses (including only result metadata)
19
20	(very entangled with trove/trovesearch/trovesearch_gathering.py)
21	'''
22	PASSIVE_RENDER = False # knows the properties it cares about	1✔
23	_page_links: set[str]	1✔
24	__already_iterated_cards = False	1✔
25
26	def simple_unicard_rendering(self, card_iri: str, osfmap_json: JsonObject) -> str:	1✔
27	raise NotImplementedError	×
28
29	def simple_multicard_rendering(self, cards: Iterator[tuple[str, JsonObject]]) -> str:	1✔
30	raise NotImplementedError	×
31
32	def unicard_rendering(self, card_iri: str, osfmap_json: JsonObject) -> ProtoRendering:	1✔
33	return SimpleRendering( # type: ignore[return-value]	×
34	mediatype=self.MEDIATYPE,
35	rendered_content=self.simple_unicard_rendering(card_iri, osfmap_json),
36	)
37
38	def multicard_rendering(self, card_pages: Iterator[dict[str, JsonObject]]) -> ProtoRendering:	1✔
39	_cards = (	×
40	(_card_iri, _card_contents)
41	for _page in card_pages
42	for _card_iri, _card_contents in _page.items()
43	)
44	return SimpleRendering( # type: ignore[return-value]	×
45	mediatype=self.MEDIATYPE,
46	rendered_content=self.simple_multicard_rendering(_cards),
47	)
48
49	def render_document(self) -> ProtoRendering:	1✔
50	_focustypes = self.response_focus.type_iris	1✔
51	if (TROVE.Cardsearch in _focustypes) or (TROVE.Valuesearch in _focustypes):	1✔
52	return self.multicard_rendering(self._iter_card_pages())	1✔
53	if TROVE.Indexcard in _focustypes:	×
54	return self.unicard_rendering(	×
55	self.response_focus.single_iri(),
56	self._get_card_content(self.response_focus.single_iri()),
57	)
58	raise trove_exceptions.UnsupportedRdfType(_focustypes)	×
59
60	def _iter_card_pages(self) -> Generator[dict[str, JsonObject]]:	1✔
61	assert not self.__already_iterated_cards	1✔
62	self.__already_iterated_cards = True	1✔
63	self._page_links = set()	1✔
64	for _page, _page_graph in self.response_gathering.ask_exhaustively(	1✔
65	TROVE.searchResultPage, focus=self.response_focus
66	):
67	if (RDF.type, JSONAPI_LINK_OBJECT) in _page:	1✔
68	self._page_links.add(_page)	×
69	elif rdf.is_container(_page):	1✔
70	_cardpage = []	1✔
71	for _search_result in rdf.container_objects(_page):	1✔
72	try:	1✔
73	_card = next(	1✔
74	_obj
75	for _pred, _obj in _search_result
76	if _pred == TROVE.indexCard
77	)
78	except StopIteration:	×
79	pass # skip malformed	×
80	else:
81	_cardpage.append(_card)	1✔
82	yield {	1✔
83	self._get_card_iri(_card): self._get_card_content(_card, _page_graph)
84	for _card in _cardpage
85	}
86
87	def _get_card_iri(self, card: str \| rdf.RdfBlanknode) -> str:	1✔
88	return card if isinstance(card, str) else ''	1✔
89
90	def _get_card_content(	1✔
91	self,
92	card: str \| rdf.RdfBlanknode,
93	graph: rdf.RdfGraph \| None = None,
94	) -> Any:
95	if isinstance(card, str):	1✔
96	_card_content = (	1✔
97	next(self.response_gathering.ask(TROVE.resourceMetadata, focus=card))
98	if graph is None
99	else next(graph.q(card, TROVE.resourceMetadata), None)
100	)
101	elif isinstance(card, frozenset):	×
102	_card_content = next(	×
103	_obj
104	for _pred, _obj in card
105	if _pred == TROVE.resourceMetadata
106	)
107	else:
108	raise trove_exceptions.ExpectedIriOrBlanknode(card)	×
109	if isinstance(_card_content, rdf.QuotedGraph):	1✔
110	return _card_content	×
111	if isinstance(_card_content, rdf.Literal) and (RDF.JSON in _card_content.datatype_iris):	1✔
112	return json.loads(_card_content.unicode_value)	1✔
113	raise ValueError(card)	×

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