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

Run Details

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

82.35

/trove/render/simple_csv.py

from __future__ import annotations
from collections.abc import (
    Generator,
    Iterator,
    Sequence,
)
import csv
import dataclasses
import functools
import itertools
import logging
from typing import TYPE_CHECKING, ClassVar

from trove.trovesearch.search_params import (
    CardsearchParams,
    ValuesearchParams,
)
from trove.util.iter import iter_unique
from trove.util.json import json_prims
from trove.util.propertypath import Propertypath, GLOB_PATHSTEP
from trove.vocab import mediatypes
from trove.vocab import osfmap
from ._simple_trovesearch import SimpleTrovesearchRenderer
from .rendering import ProtoRendering
from .rendering.streamable import StreamableRendering
if TYPE_CHECKING:
    from trove.util.trove_params import BasicTroveParams
    from trove.util.json import (
        JsonObject,
        JsonPath,
    )

_logger = logging.getLogger(__name__)

type CsvValue = str | int | float | None

_MULTIVALUE_DELIMITER = ' ; '  # possible improvement: smarter in-value delimiting?
_VALUE_KEY_PREFERENCE = ('@value', '@id', 'name', 'prefLabel', 'label')
_ID_JSONPATH = ('@id',)


class TrovesearchSimpleCsvRenderer(SimpleTrovesearchRenderer):
    MEDIATYPE = mediatypes.CSV
    CSV_DIALECT: ClassVar[type[csv.Dialect]] = csv.excel

    def multicard_rendering(self, card_pages: Iterator[Sequence[tuple[str, JsonObject]]]) -> ProtoRendering:
        _doc = TabularDoc(
            card_pages,
            trove_params=getattr(self.response_focus, 'search_params', None),
        )
        return StreamableRendering(
            mediatype=self.MEDIATYPE,
            content_stream=csv_stream(self.CSV_DIALECT, _doc.header(), _doc.rows()),
        )


def csv_stream(
    csv_dialect: type[csv.Dialect],
    header: list[CsvValue],
    rows: Iterator[list[CsvValue]],
) -> Iterator[str]:
    _writer = csv.writer(_Echo(), dialect=csv_dialect)
    yield _writer.writerow(header)
    for _row in rows:
        yield _writer.writerow(_row)


@dataclasses.dataclass
class TabularDoc:
    card_pages: Iterator[Sequence[tuple[str, JsonObject]]]
    trove_params: BasicTroveParams | None = None
    _started: bool = False

    @functools.cached_property
    def column_jsonpaths(self) -> tuple[JsonPath, ...]:
        _column_jsonpaths = (
            _osfmap_jsonpath(_path)
            for _path in self._column_paths()
        )
        return (_ID_JSONPATH, *_column_jsonpaths)

    def _column_paths(self) -> Iterator[Propertypath]:
        _pathlists: list[Sequence[Propertypath]] = []
        if self.trove_params is not None:  # hacks
            if GLOB_PATHSTEP in self.trove_params.attrpaths_by_type:
                _pathlists.append(self.trove_params.attrpaths_by_type[GLOB_PATHSTEP])
            if isinstance(self.trove_params, ValuesearchParams):
                _expected_card_types = set(self.trove_params.valuesearch_type_iris())
            elif isinstance(self.trove_params, CardsearchParams):
                _expected_card_types = set(self.trove_params.cardsearch_type_iris())
            else:
                _expected_card_types = set()
            for _type_iri in sorted(_expected_card_types, key=len):
                try:
                    _pathlist = self.trove_params.attrpaths_by_type[_type_iri]
                except KeyError:
                    pass
                else:
                    _pathlists.append(_pathlist)
        if not _pathlists:
            _pathlists.append(osfmap.DEFAULT_TABULAR_SEARCH_COLUMN_PATHS)
        return iter_unique(itertools.chain.from_iterable(_pathlists))

    def header(self) -> list[CsvValue]:
        return ['.'.join(_path) for _path in self.column_jsonpaths]

    def rows(self) -> Generator[list[CsvValue]]:
        assert not self._started
        self._started = True
        for _page in self.card_pages:
            for _card_iri, _osfmap_json in _page:
                yield self._row_values(_osfmap_json)

    def _row_values(self, osfmap_json: JsonObject) -> list[CsvValue]:
        return [
            self._row_field_value(osfmap_json, _field_path)
            for _field_path in self.column_jsonpaths
        ]

    def _row_field_value(self, osfmap_json: JsonObject, field_path: JsonPath) -> CsvValue:
        _rendered_values = [
            _obj
            for _obj in json_prims(osfmap_json, field_path, _VALUE_KEY_PREFERENCE)
            if _obj is not None
        ]
        if len(_rendered_values) == 1:
            return _rendered_values[0]  # preserve type for single numbers
        # for multiple values, can only be a string
        return _MULTIVALUE_DELIMITER.join(map(str, _rendered_values))


def _osfmap_jsonpath(iri_path: Propertypath) -> JsonPath:
    _shorthand = osfmap.osfmap_json_shorthand()
    return tuple(
        _shorthand.compact_iri(_pathstep)
        for _pathstep in iri_path
    )


class _Echo:
    '''a write-only file-like object, to convince `csv.csvwriter.writerow` to return strings

    from https://docs.djangoproject.com/en/5.1/howto/outputting-csv/#streaming-large-csv-files
    '''
    def write(self, line: str) -> str:
        return line

1	from __future__ import annotations	1✔
2	from collections.abc import (	1✔
3	Generator,
4	Iterator,
5	Sequence,
6	)
7	import csv	1✔
8	import dataclasses	1✔
9	import functools	1✔
10	import itertools	1✔
11	import logging	1✔
12	from typing import TYPE_CHECKING, ClassVar	1✔
13
14	from trove.trovesearch.search_params import (	1✔
15	CardsearchParams,
16	ValuesearchParams,
17	)
18	from trove.util.iter import iter_unique	1✔
19	from trove.util.json import json_prims	1✔
20	from trove.util.propertypath import Propertypath, GLOB_PATHSTEP	1✔
21	from trove.vocab import mediatypes	1✔
22	from trove.vocab import osfmap	1✔
23	from ._simple_trovesearch import SimpleTrovesearchRenderer	1✔
24	from .rendering import ProtoRendering	1✔
25	from .rendering.streamable import StreamableRendering	1✔
26	if TYPE_CHECKING:	1✔
27	from trove.util.trove_params import BasicTroveParams	×
NEW 28	from trove.util.json import (	×
29	JsonObject,
30	JsonPath,
31	)
32
33	_logger = logging.getLogger(__name__)	1✔
34
35	type CsvValue = str \| int \| float \| None	1✔
36
37	_MULTIVALUE_DELIMITER = ' ; ' # possible improvement: smarter in-value delimiting?	1✔
38	_VALUE_KEY_PREFERENCE = ('@value', '@id', 'name', 'prefLabel', 'label')	1✔
39	_ID_JSONPATH = ('@id',)	1✔
40
41
42	class TrovesearchSimpleCsvRenderer(SimpleTrovesearchRenderer):	1✔
43	MEDIATYPE = mediatypes.CSV	1✔
44	CSV_DIALECT: ClassVar[type[csv.Dialect]] = csv.excel	1✔
45
46	def multicard_rendering(self, card_pages: Iterator[Sequence[tuple[str, JsonObject]]]) -> ProtoRendering:	1✔
47	_doc = TabularDoc(	1✔
48	card_pages,
49	trove_params=getattr(self.response_focus, 'search_params', None),
50	)
51	return StreamableRendering(	1✔
52	mediatype=self.MEDIATYPE,
53	content_stream=csv_stream(self.CSV_DIALECT, _doc.header(), _doc.rows()),
54	)
55
56
57	def csv_stream(	1✔
58	csv_dialect: type[csv.Dialect],
59	header: list[CsvValue],
60	rows: Iterator[list[CsvValue]],
61	) -> Iterator[str]:
62	_writer = csv.writer(_Echo(), dialect=csv_dialect)	1✔
63	yield _writer.writerow(header)	1✔
64	for _row in rows:	1✔
65	yield _writer.writerow(_row)	1✔
66
67
68	@dataclasses.dataclass	1✔
69	class TabularDoc:	1✔
70	card_pages: Iterator[Sequence[tuple[str, JsonObject]]]	1✔
71	trove_params: BasicTroveParams \| None = None	1✔
72	_started: bool = False	1✔
73
74	@functools.cached_property	1✔
75	def column_jsonpaths(self) -> tuple[JsonPath, ...]:	1✔
76	_column_jsonpaths = (	1✔
77	_osfmap_jsonpath(_path)
78	for _path in self._column_paths()
79	)
80	return (_ID_JSONPATH, *_column_jsonpaths)	1✔
81
82	def _column_paths(self) -> Iterator[Propertypath]:	1✔
83	_pathlists: list[Sequence[Propertypath]] = []	1✔
84	if self.trove_params is not None: # hacks	1✔
85	if GLOB_PATHSTEP in self.trove_params.attrpaths_by_type:	×
86	_pathlists.append(self.trove_params.attrpaths_by_type[GLOB_PATHSTEP])	×
87	if isinstance(self.trove_params, ValuesearchParams):	×
88	_expected_card_types = set(self.trove_params.valuesearch_type_iris())	×
89	elif isinstance(self.trove_params, CardsearchParams):	×
90	_expected_card_types = set(self.trove_params.cardsearch_type_iris())	×
91	else:
92	_expected_card_types = set()	×
93	for _type_iri in sorted(_expected_card_types, key=len):	×
94	try:	×
95	_pathlist = self.trove_params.attrpaths_by_type[_type_iri]	×
96	except KeyError:	×
97	pass	×
98	else:
99	_pathlists.append(_pathlist)	×
100	if not _pathlists:	1✔
101	_pathlists.append(osfmap.DEFAULT_TABULAR_SEARCH_COLUMN_PATHS)	1✔
102	return iter_unique(itertools.chain.from_iterable(_pathlists))	1✔
103
104	def header(self) -> list[CsvValue]:	1✔
105	return ['.'.join(_path) for _path in self.column_jsonpaths]	1✔
106
107	def rows(self) -> Generator[list[CsvValue]]:	1✔
108	assert not self._started	1✔
109	self._started = True	1✔
110	for _page in self.card_pages:	1✔
111	for _card_iri, _osfmap_json in _page:	1✔
112	yield self._row_values(_osfmap_json)	1✔
113
114	def _row_values(self, osfmap_json: JsonObject) -> list[CsvValue]:	1✔
115	return [	1✔
116	self._row_field_value(osfmap_json, _field_path)
117	for _field_path in self.column_jsonpaths
118	]
119
120	def _row_field_value(self, osfmap_json: JsonObject, field_path: JsonPath) -> CsvValue:	1✔
121	_rendered_values = [	1✔
122	_obj
123	for _obj in json_prims(osfmap_json, field_path, _VALUE_KEY_PREFERENCE)
124	if _obj is not None
125	]
126	if len(_rendered_values) == 1:	1✔
127	return _rendered_values[0] # preserve type for single numbers	1✔
128	# for multiple values, can only be a string
129	return _MULTIVALUE_DELIMITER.join(map(str, _rendered_values))	1✔
130
131
132	def _osfmap_jsonpath(iri_path: Propertypath) -> JsonPath:	1✔
133	_shorthand = osfmap.osfmap_json_shorthand()	1✔
134	return tuple(	1✔
135	_shorthand.compact_iri(_pathstep)
136	for _pathstep in iri_path
137	)
138
139
140	class _Echo:	1✔
141	'''a write-only file-like object, to convince `csv.csvwriter.writerow` to return strings
142
143	from https://docs.djangoproject.com/en/5.1/howto/outputting-csv/#streaming-large-csv-files
144	'''
145	def write(self, line: str) -> str:	1✔
146	return line	1✔

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