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

29.41

/trove/openapi.py

import itertools
import json
from typing import Iterable, Generator, Any, Tuple

from django.conf import settings
from primitive_metadata import primitive_rdf

from share.version import get_shtrove_version
from trove.util.randomness import shuffled
from trove.vocab import mediatypes
from trove.vocab.jsonapi import JSONAPI_MEMBERNAME
from trove.vocab.namespaces import TROVE, RDFS, RDF, DCTERMS
from trove.vocab.trove import TROVE_API_THESAURUS


_OPENAPI_PARAM_LOCATION_BY_RDF_TYPE = {
    TROVE.QueryParameter: 'query',
    TROVE.PathParameter: 'path',
    TROVE.HeaderParameter: 'header',
    # TODO? 'cookie'
}


def get_trove_openapi_json() -> str:
    return json.dumps(get_trove_openapi(), indent=2)


def get_trove_openapi() -> dict[str, Any]:
    '''generate an openapi description of the trove api

    following https://spec.openapis.org/oas/v3.1.0
    '''
    # TODO: language parameter, get translations
    _api_graph = primitive_rdf.RdfGraph(TROVE_API_THESAURUS)
    _path_iris = shuffled(set(_api_graph.q(TROVE.search_api, TROVE.hasPath)))
    _label = next(_api_graph.q(TROVE.search_api, RDFS.label))
    _comment = next(_api_graph.q(TROVE.search_api, RDFS.comment))
    return {
        'openapi': '3.1.0',
        # 'externalDocs': {'url': TROVE.search_api},
        'info': {
            'title': _label.unicode_value,
            'summary': _comment.unicode_value,
            'description': _markdown_description(TROVE.search_api, _api_graph),
            'termsOfService': 'https://github.com/CenterForOpenScience/cos.io/blob/HEAD/TERMS_OF_USE.md',
            'contact': {
                # 'name':
                # 'url': web-browsable version of this
                'email': 'share-support@osf.io',
            },
            # 'license':
            'version': get_shtrove_version(),
        },
        'servers': [{
            'url': settings.SHARE_WEB_URL,
        }],
        'paths': dict(
            _openapi_path(_path_iri, _api_graph)
            for _path_iri in _path_iris
        ),
        'components': {
            'parameters': dict(_openapi_parameters(_path_iris, _api_graph)),
            'examples': dict(_openapi_examples(_path_iris, _api_graph)),
        },
    }


def _openapi_parameters(path_iris: Iterable[str], api_graph: primitive_rdf.RdfGraph) -> Iterable[tuple[str, Any]]:
    _param_iris = set(itertools.chain(*(
        api_graph.q(_path_iri, TROVE.hasParameter)
        for _path_iri in path_iris
    )))
    for _param_iri in shuffled(_param_iris):
        # TODO: better error message on absence
        try:
            _jsonname = next(api_graph.q(_param_iri, JSONAPI_MEMBERNAME))
        except StopIteration:
            raise ValueError(f'no jsonapi membername for {_param_iri}')
        _label = next(api_graph.q(_param_iri, RDFS.label))
        _comment = next(api_graph.q(_param_iri, RDFS.comment))
        _jsonschema = next(api_graph.q(_param_iri, TROVE.jsonSchema), None)
        _required = ((_param_iri, RDF.type, TROVE.RequiredParameter) in api_graph)
        _location = next(
            _OPENAPI_PARAM_LOCATION_BY_RDF_TYPE[_type_iri]
            for _type_iri in api_graph.q(_param_iri, RDF.type)
            if _type_iri in _OPENAPI_PARAM_LOCATION_BY_RDF_TYPE
        )
        yield _jsonname.unicode_value, {
            'name': _label.unicode_value,
            'in': _location,
            'required': _required,
            'summary': _comment.unicode_value,
            'description': _markdown_description(_param_iri, api_graph),
            'schema': (json.loads(_jsonschema.unicode_value) if _jsonschema else None),
        }


def _openapi_examples(path_iris: Iterable[str], api_graph: primitive_rdf.RdfGraph) -> Iterable[tuple[str, Any]]:
    # assumes examples are blank nodes (frozenset of twoples)
    _examples = set(itertools.chain(*(
        api_graph.q(_path_iri, TROVE.example)
        for _path_iri in path_iris
    )))
    for _example_blanknode in _examples:
        _example = primitive_rdf.twopledict_from_twopleset(_example_blanknode)
        _label = ''.join(
            _literal.unicode_value
            for _literal in _example.get(RDFS.label, ())
        )
        _comment = ' '.join(
            _literal.unicode_value
            for _literal in _example.get(RDFS.comment, ())
        )
        _description = '\n\n'.join(
            _literal.unicode_value
            for _literal in _example.get(DCTERMS.description, ())
        )
        _value = next(iter(_example[RDF.value]))  # assume literal
        if RDF.JSON in _value.datatype_iris:
            _valuekey = 'value'  # literal json given
            _value = json.loads(_value.unicode_value)
        else:
            _valuekey = 'externalValue'  # link
            _value = _value.unicode_value
        yield _label, {
            'summary': _comment,
            'description': _description,
            _valuekey: _value,
        }


def _openapi_path(path_iri: str, api_graph: primitive_rdf.RdfGraph) -> Tuple[str, Any]:
    # TODO: better error message on absence
    try:
        _path = next(iter(_text(path_iri, TROVE.iriPath, api_graph)))
    except StopIteration:
        raise ValueError(f'could not find trove:iriPath for {path_iri}')
    _label = ' '.join(_text(path_iri, RDFS.label, api_graph))
    _param_labels = shuffled(_text(path_iri, {TROVE.hasParameter: {JSONAPI_MEMBERNAME}}, api_graph))
    _example_labels = shuffled(_text(path_iri, {TROVE.example: {RDFS.label}}, api_graph))
    return _path, {
        'get': {  # TODO (if generalizability): separate metadata by verb
            # 'tags':
            'summary': _label,
            'description': _markdown_description(path_iri, api_graph),
            # 'externalDocs':
            'operationId': path_iri,
            'parameters': [
                {'$ref': f'#/components/parameters/{_param_label}'}
                for _param_label in _param_labels
            ],
            'responses': {
                '200': {
                    'description': 'ok',
                    'content': {
                        mediatypes.JSONAPI: {
                            'examples': [
                                {'$ref': f'#/components/examples/{_example_label}'}
                                for _example_label in _example_labels
                            ],
                        },
                    },
                },
            },
        },
    }


def _concept_markdown_blocks(concept_iri: str, api_graph: primitive_rdf.RdfGraph) -> Generator[str, None, None]:
    for _label in api_graph.q(concept_iri, RDFS.label):
        yield f'## {_label.unicode_value}'
    for _comment in api_graph.q(concept_iri, RDFS.comment):
        yield f'<aside>{_comment.unicode_value}</aside>'
    for _desc in api_graph.q(concept_iri, DCTERMS.description):
        yield _desc.unicode_value


def _text(subj: Any, pred: Any, api_graph: primitive_rdf.RdfGraph) -> Iterable[str]:
    for _obj in api_graph.q(subj, pred):
        yield _obj.unicode_value


def _markdown_description(subj_iri: str, api_graph: primitive_rdf.RdfGraph) -> str:
    return '\n\n'.join((
        *(
            _description.unicode_value
            for _description in api_graph.q(subj_iri, DCTERMS.description)
        ),
        *(
            '\n\n'.join(_concept_markdown_blocks(_concept_iri, api_graph))
            for _concept_iri in api_graph.q(subj_iri, TROVE.usesConcept)
        ),
    ))

1	import itertools	1✔
2	import json	1✔
3	from typing import Iterable, Generator, Any, Tuple	1✔
4
5	from django.conf import settings	1✔
6	from primitive_metadata import primitive_rdf	1✔
7
8	from share.version import get_shtrove_version	1✔
9	from trove.util.randomness import shuffled	1✔
10	from trove.vocab import mediatypes	1✔
11	from trove.vocab.jsonapi import JSONAPI_MEMBERNAME	1✔
12	from trove.vocab.namespaces import TROVE, RDFS, RDF, DCTERMS	1✔
13	from trove.vocab.trove import TROVE_API_THESAURUS	1✔
14
15
16	_OPENAPI_PARAM_LOCATION_BY_RDF_TYPE = {	1✔
17	TROVE.QueryParameter: 'query',
18	TROVE.PathParameter: 'path',
19	TROVE.HeaderParameter: 'header',
20	# TODO? 'cookie'
21	}
22
23
24	def get_trove_openapi_json() -> str:	1✔
25	return json.dumps(get_trove_openapi(), indent=2)	×
26
27
28	def get_trove_openapi() -> dict[str, Any]:	1✔
29	'''generate an openapi description of the trove api
30
31	following https://spec.openapis.org/oas/v3.1.0
32	'''
33	# TODO: language parameter, get translations
34	_api_graph = primitive_rdf.RdfGraph(TROVE_API_THESAURUS)	×
35	_path_iris = shuffled(set(_api_graph.q(TROVE.search_api, TROVE.hasPath)))	×
36	_label = next(_api_graph.q(TROVE.search_api, RDFS.label))	×
37	_comment = next(_api_graph.q(TROVE.search_api, RDFS.comment))	×
38	return {	×
39	'openapi': '3.1.0',
40	# 'externalDocs': {'url': TROVE.search_api},
41	'info': {
42	'title': _label.unicode_value,
43	'summary': _comment.unicode_value,
44	'description': _markdown_description(TROVE.search_api, _api_graph),
45	'termsOfService': 'https://github.com/CenterForOpenScience/cos.io/blob/HEAD/TERMS_OF_USE.md',
46	'contact': {
47	# 'name':
48	# 'url': web-browsable version of this
49	'email': 'share-support@osf.io',
50	},
51	# 'license':
52	'version': get_shtrove_version(),
53	},
54	'servers': [{
55	'url': settings.SHARE_WEB_URL,
56	}],
57	'paths': dict(
58	_openapi_path(_path_iri, _api_graph)
59	for _path_iri in _path_iris
60	),
61	'components': {
62	'parameters': dict(_openapi_parameters(_path_iris, _api_graph)),
63	'examples': dict(_openapi_examples(_path_iris, _api_graph)),
64	},
65	}
66
67
68	def _openapi_parameters(path_iris: Iterable[str], api_graph: primitive_rdf.RdfGraph) -> Iterable[tuple[str, Any]]:	1✔
69	_param_iris = set(itertools.chain(*(	×
70	api_graph.q(_path_iri, TROVE.hasParameter)
71	for _path_iri in path_iris
72	)))
73	for _param_iri in shuffled(_param_iris):	×
74	# TODO: better error message on absence
75	try:	×
76	_jsonname = next(api_graph.q(_param_iri, JSONAPI_MEMBERNAME))	×
77	except StopIteration:	×
78	raise ValueError(f'no jsonapi membername for {_param_iri}')	×
79	_label = next(api_graph.q(_param_iri, RDFS.label))	×
80	_comment = next(api_graph.q(_param_iri, RDFS.comment))	×
81	_jsonschema = next(api_graph.q(_param_iri, TROVE.jsonSchema), None)	×
82	_required = ((_param_iri, RDF.type, TROVE.RequiredParameter) in api_graph)	×
83	_location = next(	×
84	_OPENAPI_PARAM_LOCATION_BY_RDF_TYPE[_type_iri]
85	for _type_iri in api_graph.q(_param_iri, RDF.type)
86	if _type_iri in _OPENAPI_PARAM_LOCATION_BY_RDF_TYPE
87	)
88	yield _jsonname.unicode_value, {	×
89	'name': _label.unicode_value,
90	'in': _location,
91	'required': _required,
92	'summary': _comment.unicode_value,
93	'description': _markdown_description(_param_iri, api_graph),
94	'schema': (json.loads(_jsonschema.unicode_value) if _jsonschema else None),
95	}
96
97
98	def _openapi_examples(path_iris: Iterable[str], api_graph: primitive_rdf.RdfGraph) -> Iterable[tuple[str, Any]]:	1✔
99	# assumes examples are blank nodes (frozenset of twoples)
100	_examples = set(itertools.chain(*(	×
101	api_graph.q(_path_iri, TROVE.example)
102	for _path_iri in path_iris
103	)))
104	for _example_blanknode in _examples:	×
105	_example = primitive_rdf.twopledict_from_twopleset(_example_blanknode)	×
106	_label = ''.join(	×
107	_literal.unicode_value
108	for _literal in _example.get(RDFS.label, ())
109	)
110	_comment = ' '.join(	×
111	_literal.unicode_value
112	for _literal in _example.get(RDFS.comment, ())
113	)
114	_description = '\n\n'.join(	×
115	_literal.unicode_value
116	for _literal in _example.get(DCTERMS.description, ())
117	)
118	_value = next(iter(_example[RDF.value])) # assume literal	×
119	if RDF.JSON in _value.datatype_iris:	×
120	_valuekey = 'value' # literal json given	×
121	_value = json.loads(_value.unicode_value)	×
122	else:
123	_valuekey = 'externalValue' # link	×
124	_value = _value.unicode_value	×
125	yield _label, {	×
126	'summary': _comment,
127	'description': _description,
128	_valuekey: _value,
129	}
130
131
132	def _openapi_path(path_iri: str, api_graph: primitive_rdf.RdfGraph) -> Tuple[str, Any]:	1✔
133	# TODO: better error message on absence
134	try:	×
NEW 135	_path = next(iter(_text(path_iri, TROVE.iriPath, api_graph)))	×
136	except StopIteration:	×
137	raise ValueError(f'could not find trove:iriPath for {path_iri}')	×
138	_label = ' '.join(_text(path_iri, RDFS.label, api_graph))	×
139	_param_labels = shuffled(_text(path_iri, {TROVE.hasParameter: {JSONAPI_MEMBERNAME}}, api_graph))	×
140	_example_labels = shuffled(_text(path_iri, {TROVE.example: {RDFS.label}}, api_graph))	×
141	return _path, {	×
142	'get': { # TODO (if generalizability): separate metadata by verb
143	# 'tags':
144	'summary': _label,
145	'description': _markdown_description(path_iri, api_graph),
146	# 'externalDocs':
147	'operationId': path_iri,
148	'parameters': [
149	{'$ref': f'#/components/parameters/{_param_label}'}
150	for _param_label in _param_labels
151	],
152	'responses': {
153	'200': {
154	'description': 'ok',
155	'content': {
156	mediatypes.JSONAPI: {
157	'examples': [
158	{'$ref': f'#/components/examples/{_example_label}'}
159	for _example_label in _example_labels
160	],
161	},
162	},
163	},
164	},
165	},
166	}
167
168
169	def _concept_markdown_blocks(concept_iri: str, api_graph: primitive_rdf.RdfGraph) -> Generator[str, None, None]:	1✔
170	for _label in api_graph.q(concept_iri, RDFS.label):	×
171	yield f'## {_label.unicode_value}'	×
172	for _comment in api_graph.q(concept_iri, RDFS.comment):	×
173	yield f'<aside>{_comment.unicode_value}</aside>'	×
174	for _desc in api_graph.q(concept_iri, DCTERMS.description):	×
175	yield _desc.unicode_value	×
176
177
178	def _text(subj: Any, pred: Any, api_graph: primitive_rdf.RdfGraph) -> Iterable[str]:	1✔
179	for _obj in api_graph.q(subj, pred):	×
180	yield _obj.unicode_value	×
181
182
183	def _markdown_description(subj_iri: str, api_graph: primitive_rdf.RdfGraph) -> str:	1✔
184	return '\n\n'.join((	×
185	*(
186	_description.unicode_value
187	for _description in api_graph.q(subj_iri, DCTERMS.description)
188	),
189	*(
190	'\n\n'.join(_concept_markdown_blocks(_concept_iri, api_graph))
191	for _concept_iri in api_graph.q(subj_iri, TROVE.usesConcept)
192	),
193	))

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