18707793088

Committed 22 Oct 2025 06:47AM UTC coverage: 86.825% (+0.006%) from 86.819%

Build # 18707793088

Build Type

Pull #1065

github

Committed by

web-flow

Commit Message

Merge 30515e0e4 into d4f265455

Pull Request Pull Request #1065: feat: allow data connectors to be skipped when launching sessions

Run Details

36 of 59 new or added lines in 4 files covered. (61.02%)

5 existing lines in 3 files now uncovered.

22723 of 26171 relevant lines covered (86.83%)

1.52 hits per line

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

70.67

/components/renku_data_services/data_connectors/doi/metadata.py

"""Metadata handling for DOIs."""

import httpx
from pydantic import ValidationError as PydanticValidationError

from renku_data_services.data_connectors.doi import models
from renku_data_services.storage.rclone import RCloneDOIMetadata


async def get_dataset_metadata(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata | None:
    """Retrieve DOI metadata."""
    if rclone_metadata.provider == "invenio" or rclone_metadata.provider == "zenodo":
        return await _get_dataset_metadata_invenio(rclone_metadata=rclone_metadata)
    if rclone_metadata.provider == "dataverse":
        return await _get_dataset_metadata_dataverse(rclone_metadata=rclone_metadata)
    return None


async def _get_dataset_metadata_invenio(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata | None:
    """Retrieve DOI metadata from the InvenioRDM API."""
    metadata_url = rclone_metadata.metadata_url
    if not metadata_url:
        return None

    async with httpx.AsyncClient(timeout=5) as client:
        try:
            res = await client.get(url=metadata_url, follow_redirects=True, headers=[("accept", "application/json")])
            if res.status_code >= 400:
                return None
            record = models.InvenioRecord.model_validate_json(res.content)
        except httpx.HTTPError:
            return None
        except PydanticValidationError:
            return None

    name = ""
    description = ""
    keywords = []
    if record.metadata is not None:
        name = record.metadata.title or ""
        description = record.metadata.description or ""
        keywords = record.metadata.keywords or []
    return models.DOIMetadata(name=name, description=description, keywords=keywords)


async def _get_dataset_metadata_dataverse(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata | None:
    """Retrieve DOI metadata from the Dataverse API."""
    metadata_url = rclone_metadata.metadata_url
    if not metadata_url:
        return None

    async with httpx.AsyncClient(timeout=5) as client:
        try:
            res = await client.get(url=metadata_url, follow_redirects=True, headers=[("accept", "application/json")])
            if res.status_code >= 400:
                return None
            response = models.DataverseDatasetResponse.model_validate_json(res.content)
        except httpx.HTTPError:
            return None
        except PydanticValidationError:
            return None

    if response.status != "OK":
        return None

    name = ""
    description = ""
    keywords: list[str] = []
    if (
        response.data is not None
        and response.data.latest_version is not None
        and response.data.latest_version.metadata_blocks is not None
        and response.data.latest_version.metadata_blocks.citation is not None
    ):
        for field in response.data.latest_version.metadata_blocks.citation.fields:
            if field.type_name == "title" and field.type_class == "primitive" and not field.multiple:
                name = str(field.value)
            if (
                field.type_name == "dsDescription"
                and field.type_class == "compound"
                and field.multiple
                and isinstance(field.value, list)
                and field.value
            ):
                try:
                    description_field = models.DataverseMetadataBlockCitationField.model_validate(
                        field.value[0].get("dsDescriptionValue", dict())
                    )
                    if (
                        description_field.type_name == "dsDescriptionValue"
                        and description_field.type_class == "primitive"
                        and not description_field.multiple
                    ):
                        description = str(description_field.value)
                except AttributeError:
                    pass
                except PydanticValidationError:
                    pass
            if (
                field.type_name == "keyword"
                and field.type_class == "compound"
                and field.multiple
                and isinstance(field.value, list)
            ):
                for value in field.value:
                    try:
                        kw_field = models.DataverseMetadataBlockCitationField.model_validate(
                            value.get("keywordValue", dict())
                        )
                        if (
                            kw_field.type_name == "keywordValue"
                            and kw_field.type_class == "primitive"
                            and not kw_field.multiple
                        ):
                            keywords.append(str(kw_field.value))
                    except AttributeError:
                        pass
                    except PydanticValidationError:
                        pass
    return models.DOIMetadata(name=name, description=description, keywords=keywords)

1	"""Metadata handling for DOIs."""
2
3	import httpx	2✔
4	from pydantic import ValidationError as PydanticValidationError	2✔
5
6	from renku_data_services.data_connectors.doi import models	2✔
7	from renku_data_services.storage.rclone import RCloneDOIMetadata	2✔
8
9
10	async def get_dataset_metadata(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata \| None:	2✔
11	"""Retrieve DOI metadata."""
12	if rclone_metadata.provider == "invenio" or rclone_metadata.provider == "zenodo":	1✔
13	return await _get_dataset_metadata_invenio(rclone_metadata=rclone_metadata)	1✔
14	if rclone_metadata.provider == "dataverse":	1✔
15	return await _get_dataset_metadata_dataverse(rclone_metadata=rclone_metadata)	1✔
16	return None	×
17
18
19	async def _get_dataset_metadata_invenio(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata \| None:	2✔
20	"""Retrieve DOI metadata from the InvenioRDM API."""
21	metadata_url = rclone_metadata.metadata_url	1✔
22	if not metadata_url:	1✔
23	return None	×
24
25	async with httpx.AsyncClient(timeout=5) as client:	1✔
26	try:	1✔
27	res = await client.get(url=metadata_url, follow_redirects=True, headers=[("accept", "application/json")])	1✔
28	if res.status_code >= 400:	1✔
29	return None	×
30	record = models.InvenioRecord.model_validate_json(res.content)	1✔
UNCOV 31	except httpx.HTTPError:	×
UNCOV 32	return None	×
33	except PydanticValidationError:	×
34	return None	×
35
36	name = ""	1✔
37	description = ""	1✔
38	keywords = []	1✔
39	if record.metadata is not None:	1✔
40	name = record.metadata.title or ""	1✔
41	description = record.metadata.description or ""	1✔
42	keywords = record.metadata.keywords or []	1✔
43	return models.DOIMetadata(name=name, description=description, keywords=keywords)	1✔
44
45
46	async def _get_dataset_metadata_dataverse(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata \| None:	2✔
47	"""Retrieve DOI metadata from the Dataverse API."""
48	metadata_url = rclone_metadata.metadata_url	1✔
49	if not metadata_url:	1✔
50	return None	×
51
52	async with httpx.AsyncClient(timeout=5) as client:	1✔
53	try:	1✔
54	res = await client.get(url=metadata_url, follow_redirects=True, headers=[("accept", "application/json")])	1✔
55	if res.status_code >= 400:	1✔
56	return None	×
57	response = models.DataverseDatasetResponse.model_validate_json(res.content)	1✔
58	except httpx.HTTPError:	×
59	return None	×
60	except PydanticValidationError:	×
61	return None	×
62
63	if response.status != "OK":	1✔
64	return None	×
65
66	name = ""	1✔
67	description = ""	1✔
68	keywords: list[str] = []	1✔
69	if (	1✔
70	response.data is not None
71	and response.data.latest_version is not None
72	and response.data.latest_version.metadata_blocks is not None
73	and response.data.latest_version.metadata_blocks.citation is not None
74	):
75	for field in response.data.latest_version.metadata_blocks.citation.fields:	1✔
76	if field.type_name == "title" and field.type_class == "primitive" and not field.multiple:	1✔
77	name = str(field.value)	1✔
78	if (	1✔
79	field.type_name == "dsDescription"
80	and field.type_class == "compound"
81	and field.multiple
82	and isinstance(field.value, list)
83	and field.value
84	):
85	try:	1✔
86	description_field = models.DataverseMetadataBlockCitationField.model_validate(	1✔
87	field.value[0].get("dsDescriptionValue", dict())
88	)
89	if (	1✔
90	description_field.type_name == "dsDescriptionValue"
91	and description_field.type_class == "primitive"
92	and not description_field.multiple
93	):
94	description = str(description_field.value)	1✔
95	except AttributeError:	×
96	pass	×
97	except PydanticValidationError:	×
98	pass	×
99	if (	1✔
100	field.type_name == "keyword"
101	and field.type_class == "compound"
102	and field.multiple
103	and isinstance(field.value, list)
104	):
105	for value in field.value:	1✔
106	try:	1✔
107	kw_field = models.DataverseMetadataBlockCitationField.model_validate(	1✔
108	value.get("keywordValue", dict())
109	)
110	if (	1✔
111	kw_field.type_name == "keywordValue"
112	and kw_field.type_class == "primitive"
113	and not kw_field.multiple
114	):
115	keywords.append(str(kw_field.value))	1✔
116	except AttributeError:	×
117	pass	×
118	except PydanticValidationError:	×
119	pass	×
120	return models.DOIMetadata(name=name, description=description, keywords=keywords)	1✔

SwissDataScienceCenter / renku-data-services / 18707793088

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