chore: cleanup code, add tests

Author: olevski

components/renku_data_services/data_connectors/core.py

@@ -15,9 +15,11 @@
     NamespacePath,
     ProjectPath,
 )
-from renku_data_services.data_connectors import apispec, models, schema_org_dataset
+from renku_data_services.data_connectors import apispec, models
 from renku_data_services.data_connectors.constants import ALLOWED_GLOBAL_DATA_CONNECTOR_PROVIDERS
-from renku_data_services.data_connectors.doi.metadata import get_dataset_metadata
+from renku_data_services.data_connectors.doi import schema_org
+from renku_data_services.data_connectors.doi.metadata import create_envidat_metadata_url, get_dataset_metadata
+from renku_data_services.data_connectors.doi.models import DOI, SchemaOrgDataset
 from renku_data_services.storage import models as storage_models
 from renku_data_services.storage.constants import ENVIDAT_V1_PROVIDER
 from renku_data_services.storage.rclone import RCloneValidator
@@ -129,29 +131,37 @@ async def prevalidate_unsaved_global_data_connector(
     if not storage.readonly:
         raise errors.ValidationError(message="Global data connectors must be read-only")
 
-    match storage.storage_type:
-        case "doi":
-            rclone_metadata = await validator.get_doi_metadata(configuration=storage.configuration)
-
-            doi_uri = f"doi:{rclone_metadata.doi}"
-            slug = base_models.Slug.from_name(doi_uri).value
-
-            # Override provider in storage config
-            storage.configuration["provider"] = rclone_metadata.provider
-        case x if x == ENVIDAT_V1_PROVIDER:
-            if not isinstance(body.storage, apispec.CloudStorageCorePost):
-                raise errors.ValidationError()
-            doi = body.storage.configuration.get("doi")
-            if not doi:
-                raise errors.ValidationError()
-            doi_uri = f"doi:{doi}"
-            slug = base_models.Slug.from_name(doi_uri).value
-        case x:
+    rclone_metadata = await validator.get_doi_metadata(configuration=storage.configuration)
+    if rclone_metadata:
+        doi_uri = f"doi:{rclone_metadata.doi}"
+
+        # Override provider in storage config
+        storage.configuration["provider"] = rclone_metadata.provider
+        doi = DOI(rclone_metadata.doi)
+    else:
+        # The storage is not supported by rclone
+        if not isinstance(body.storage, apispec.CloudStorageCorePost):
+            raise errors.ValidationError(
+                message="When the data connector is not supported by rclone we cannot parse a storage URL."
+            )
+        # Try to see if we have a different type not directly supported by rclone - from envidat for example
+        doi_str = body.storage.configuration.get("doi")
+        if not isinstance(doi_str, str):
+            raise errors.ValidationError(message="A doi could not be found in the storage configuration.")
+        doi = DOI(doi_str)
+        host = await doi.resolve_host()
+        if not host:
+            raise errors.ValidationError(message=f"The provided doi {doi} cannot be resolved.")
+        doi_uri = f"doi:{doi}"
+        if host not in ["envidat.ch", "www.envidat.ch"]:
             raise errors.ValidationError(
-                message=f"Only {ALLOWED_GLOBAL_DATA_CONNECTOR_PROVIDERS} storage type is allowed "
-                "for global data connectors"
+                message="The doi for the global data connector resolved to an unsupported host"
             )
+        # Set the storage type and re-validate
+        body.storage.storage_type = ENVIDAT_V1_PROVIDER
+        storage = await validate_unsaved_storage(body.storage, validator=validator)
 
+    slug = base_models.Slug.from_name(doi_uri).value
     return models.UnsavedGlobalDataConnector(
         name=doi_uri,
         slug=slug,
@@ -160,6 +170,7 @@ async def prevalidate_unsaved_global_data_connector(
         storage=storage,
         description=None,
         keywords=[],
+        doi=doi,
     )
 
 
@@ -181,7 +192,12 @@ async def validate_unsaved_global_data_connector(
     # Fetch DOI metadata
     if data_connector.storage.storage_type == "doi":
         rclone_metadata = await validator.get_doi_metadata(configuration=data_connector.storage.configuration)
-        metadata = await get_dataset_metadata(rclone_metadata=rclone_metadata)
+        if not rclone_metadata:
+            raise errors.ValidationError()
+        metadata = await get_dataset_metadata(data_connector.storage.storage_type, rclone_metadata.metadata_url)
+    elif data_connector.storage.storage_type == ENVIDAT_V1_PROVIDER:
+        metadata_url = create_envidat_metadata_url(data_connector.doi)
+        metadata = await get_dataset_metadata(data_connector.storage.storage_type, metadata_url)
     else:
         metadata = None
 
@@ -228,6 +244,7 @@ async def validate_unsaved_global_data_connector(
         storage=storage,
         description=description or None,
         keywords=keywords,
+        doi=data_connector.doi,
     )
 
 
@@ -371,8 +388,7 @@ async def convert_envidat_v1_data_connector_to_s3(
         raise errors.ValidationError()
     if len(doi) == 0:
         raise errors.ValidationError()
-    doi = doi.removeprefix("https://")
-    doi = doi.removeprefix("http://")
+    doi = DOI(doi)
 
     new_config = payload.model_copy(deep=True)
     new_config.configuration = {}
@@ -386,12 +402,45 @@ async def convert_envidat_v1_data_connector_to_s3(
         res = await clnt.get(envidat_url, params=query_params, headers=headers)
         if res.status_code != 200:
             raise errors.ProgrammingError()
-    dataset = schema_org_dataset.Dataset.model_validate_strings(res.text)
-    s3_config = schema_org_dataset.get_rclone_config(
+    dataset = SchemaOrgDataset.model_validate_strings(res.text)
+    s3_config = schema_org.get_rclone_config(
         dataset,
-        schema_org_dataset.DatasetProvider.envidat,
+        schema_org.DatasetProvider.envidat,
     )
     new_config.configuration = dict(s3_config.rclone_config)
     new_config.source_path = s3_config.path
     new_config.storage_type = "s3"
     return new_config
+
+
+# async def get_metadata(
+#     configuration: storage_models.RCloneConfig | dict[str, Any], validator: RCloneValidator
+# ) -> RCloneDOIMetadata | None:
+#     """Get metadata for the dataset."""
+#     if isinstance(configuration, storage_models.RCloneConfig):
+#         return await validator.get_doi_metadata(configuration)
+#     doi = configuration.get("doi")
+#     if not doi:
+#         return None
+#     parsed_doi = urlparse(doi)
+#     if parsed_doi.scheme.decode() not in ["http", "https"]:
+#         doi = urlunparse(parsed_doi._replace(scheme=b"https")).decode()
+#     clnt = httpx.AsyncClient(follow_redirects=True)
+#     async with clnt:
+#         res = await clnt.get(doi)
+#     if res.status_code != 200:
+#         return None
+#     match res.url.host:
+#         case "www.envidat.ch":
+#
+#
+# async def get_envidat_metadata(doi: DOI) -> dict | None:
+#     """Get metadata about the envidat dataset, the doi should not be a url."""
+#     clnt = httpx.AsyncClient()
+#     url = "https://envidat.ch/converters-api/internal-dataset/convert/jsonld"
+#     params = {"query": doi}
+#     async with clnt:
+#         res = clnt.get(url, params=params)
+#     if res.status_code != 200:
+#         return None
+#

components/renku_data_services/data_connectors/doi/metadata.py

@@ -1,27 +1,27 @@
 """Metadata handling for DOIs."""
 
+from urllib.parse import urlencode
+
 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
+from renku_data_services.storage.constants import ENVIDAT_V1_PROVIDER
 
 
-async def get_dataset_metadata(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata | None:
+async def get_dataset_metadata(provider: str, metadata_url: str) -> 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)
+    if provider == "invenio" or provider == "zenodo":
+        return await _get_dataset_metadata_invenio(metadata_url)
+    if provider == "dataverse":
+        return await _get_dataset_metadata_dataverse(metadata_url)
+    if provider == ENVIDAT_V1_PROVIDER:
+        return await _get_envidat_metadata(metadata_url)
     return None
 
 
-async def _get_dataset_metadata_invenio(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata | None:
+async def _get_dataset_metadata_invenio(metadata_url: str) -> 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")])
@@ -43,11 +43,8 @@ async def _get_dataset_metadata_invenio(rclone_metadata: RCloneDOIMetadata) -> m
     return models.DOIMetadata(name=name, description=description, keywords=keywords)
 
 
-async def _get_dataset_metadata_dataverse(rclone_metadata: RCloneDOIMetadata) -> models.DOIMetadata | None:
+async def _get_dataset_metadata_dataverse(metadata_url: str) -> 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:
@@ -118,3 +115,30 @@ async def _get_dataset_metadata_dataverse(rclone_metadata: RCloneDOIMetadata) ->
                     except PydanticValidationError:
                         pass
     return models.DOIMetadata(name=name, description=description, keywords=keywords)
+
+
+def create_envidat_metadata_url(doi: models.DOI) -> str:
+    """Create the metadata url for envidat from a DOI."""
+    url = "https://envidat.ch/converters-api/internal-dataset/convert/jsonld"
+    params = urlencode({"query": doi})
+    return f"{url}?{params}"
+
+
+async def _get_envidat_metadata(metadata_url: str) -> models.DOIMetadata | None:
+    """Get metadata about the envidat dataset."""
+    clnt = httpx.AsyncClient(follow_redirects=True, timeout=5)
+    headers = {"accept": "application/json"}
+    async with clnt:
+        try:
+            res = await clnt.get(metadata_url, headers=headers)
+        except httpx.HTTPError:
+            return None
+    if res.status_code != 200:
+        return None
+    try:
+        parsed_metadata = models.SchemaOrgDataset.model_validate_json(res.text)
+    except PydanticValidationError:
+        return None
+    return models.DOIMetadata(
+        name=parsed_metadata.name, description=parsed_metadata.description or "", keywords=parsed_metadata.keywords
+    )

components/renku_data_services/data_connectors/doi/models.py

@@ -1,9 +1,58 @@
 """Models for DOIs."""
 
+import re
 from dataclasses import dataclass
-from typing import Any
-
-from pydantic import BaseModel, Field
+from typing import Any, Self
+from urllib.parse import urlparse
+
+import httpx
+from pydantic import BaseModel, ConfigDict, Field
+
+from renku_data_services.errors import errors
+
+
+class DOI(str):
+    """A doi for a dataset or a similar resource."""
+
+    __regex = re.compile(r"^10\.\d{4,9}/\S+$", re.IGNORECASE)
+
+    def __new__(cls, doi: str) -> Self:
+        """Create a new doi.
+
+        A few cases possible:
+        doi:10.16904/12
+        10.16904/12
+        https://www.doi.org/10.16904/12
+        http://www.doi.org/10.16904/12
+        http://doi.org/10.16904/12
+        """
+        doi_parsed = urlparse(doi)
+        doi_clean = doi
+        if doi_parsed.netloc in ["www.doi.org", "doi.org"]:
+            if doi_parsed.scheme not in ["https", "http"]:
+                raise errors.ValidationError(
+                    message=f"Received the right doi.org host but an unexpected scheme {doi_parsed} for doi {doi}."
+                )
+            doi_clean = doi_parsed.path.strip("/")
+        if doi.startswith("doi:"):
+            doi_clean = doi[4:]
+        if not doi_clean or not DOI.__regex.match(doi_clean):
+            raise errors.ValidationError(message=f"The provided value {doi} is not a valid doi.")
+        return super().__new__(cls, doi_clean)
+
+    @property
+    def url(self) -> str:
+        """Return a proper URL from the doi."""
+        return f"https://doi.org/{self}"
+
+    async def resolve_host(self) -> str | None:
+        """Resolves the DOI and returns the hostname of the url where the redirect leads."""
+        clnt = httpx.AsyncClient(follow_redirects=True)
+        async with clnt:
+            res = await clnt.get(self.url)
+        if res.status_code != 200:
+            return None
+        return res.url.host
 
 
 @dataclass(frozen=True, eq=True, kw_only=True)
@@ -67,3 +116,26 @@ class DataverseDatasetResponse(BaseModel):
 
     status: str = Field()
     data: DataverseDataset | None = Field()
+
+
+class SchemaOrgDistribution(BaseModel):
+    """The distribution field of a schema.org dataset."""
+
+    model_config = ConfigDict(extra="ignore")
+    type: str = Field(alias="@type")
+    content_url: str = Field(alias="contentUrl")
+
+
+class SchemaOrgDataset(BaseModel):
+    """A very limited and partial spec of a schema.org Dataset used by Scicat and Envidat."""
+
+    model_config = ConfigDict(extra="ignore")
+    distribution: list[SchemaOrgDistribution] = Field(default_factory=list)
+    name: str = Field()
+    description: str | None = None
+    raw_keywords: str = Field(alias="keywords", default="")
+
+    @property
+    def keywords(self) -> list[str]:
+        """Split the single keywords string into a list."""
+        return self.raw_keywords.split()

…es/data_connectors/schema_org_dataset.py …rvices/data_connectors/doi/schema_org.pycomponents/renku_data_services/data_connectors/schema_org_dataset.py renamed to components/renku_data_services/data_connectors/doi/schema_org.py components/renku_data_services/data_connectors/schema_org_dataset.py renamed to components/renku_data_services/data_connectors/doi/schema_org.py

@@ -4,27 +4,10 @@
 from enum import StrEnum
 from urllib.parse import parse_qs, urlparse
 
-from pydantic import BaseModel, ConfigDict, Field
-
+from renku_data_services.data_connectors.doi.models import SchemaOrgDataset
 from renku_data_services.errors import errors
 
 
-class Distribution(BaseModel):
-    """The distribution field of a schema.org dataset."""
-
-    model_config = ConfigDict(extra="ignore")
-    type: str
-    contentUrl: str
-    name: str
-
-
-class Dataset(BaseModel):
-    """A very limited and partial spec of a schema.org Dataset used by Scicat and Envidat."""
-
-    model_config = ConfigDict(extra="ignore")
-    distribution: list[Distribution] = Field(default_factory=list)
-
-
 class DatasetProvider(StrEnum):
     """The provider for the dataset."""
 
@@ -45,7 +28,7 @@ def path(self) -> str:
         return f"{self.bucket}/{self.prefix}"
 
 
-def get_rclone_config(dataset: Dataset, provider: DatasetProvider) -> S3Config:
+def get_rclone_config(dataset: SchemaOrgDataset, provider: DatasetProvider) -> S3Config:
     """Parse the dataset into an rclone configuration."""
     match provider:
         case DatasetProvider.envidat:
@@ -55,10 +38,10 @@ def get_rclone_config(dataset: Dataset, provider: DatasetProvider) -> S3Config:
             raise errors.ValidationError(message=f"Got an unknown dataset provider {x}")
 
 
-def __get_rclone_s3_config_envidat(dataset: Dataset) -> S3Config:
+def __get_rclone_s3_config_envidat(dataset: SchemaOrgDataset) -> S3Config:
     """Get the S3 rclone configuration and source path from a dataset returned by envidat."""
     # NOTE: The folks from Envidat assure us that the first entity in the list is the one we want
-    url = dataset.distribution[0].contentUrl
+    url = dataset.distribution[0].content_url
     # NOTE: The folks from Envidat assure us that the URL has the following format
     # http://<bucket-name>.<s3 domain>/?prefix=<path to files>
     url_parsed = urlparse(url)

components/renku_data_services/data_connectors/models.py

@@ -14,6 +14,7 @@
     NamespacePath,
     ProjectPath,
 )
+from renku_data_services.data_connectors.doi.models import DOI
 from renku_data_services.namespace.models import GroupNamespace, ProjectNamespace, UserNamespace
 from renku_data_services.utils.etag import compute_etag_from_fields
 
@@ -97,6 +98,7 @@ class UnsavedGlobalDataConnector(BaseDataConnector):
     """Global data connector model."""
 
     namespace: None = None
+    doi: DOI
 
 
 @dataclass(frozen=True, eq=True, kw_only=True)