feat: add envidat data connectors support

olevski · olevski · commit 43d6a9ba0e13 · 2025-11-19T21:20:14.000+01:00
diff --git a/components/renku_data_services/data_connectors/blueprints.py b/components/renku_data_services/data_connectors/blueprints.py
@@ -99,7 +99,7 @@ def post(self) -> BlueprintFactoryResponse:
         async def _post(
             _: Request, user: base_models.APIUser, body: apispec.DataConnectorPost, validator: RCloneValidator
         ) -> JSONResponse:
-            data_connector = validate_unsaved_data_connector(body, validator=validator)
+            data_connector = await validate_unsaved_data_connector(body, validator=validator)
             result = await self.data_connector_repo.insert_namespaced_data_connector(
                 user=user, data_connector=data_connector
             )
diff --git a/components/renku_data_services/data_connectors/constants.py b/components/renku_data_services/data_connectors/constants.py
@@ -0,0 +1,7 @@
+"""Constants for data connectors."""
+
+from typing import Final
+
+from renku_data_services.storage.constants import ENVIDAT_V1_PROVIDER
+
+ALLOWED_GLOBAL_DATA_CONNECTOR_PROVIDERS: Final[list[str]] = ["doi", ENVIDAT_V1_PROVIDER]
diff --git a/components/renku_data_services/data_connectors/core.py b/components/renku_data_services/data_connectors/core.py
@@ -6,6 +6,7 @@
 from html.parser import HTMLParser
 from typing import Any
 
+import httpx
 from pydantic import ValidationError as PydanticValidationError
 
 from renku_data_services import base_models, errors
@@ -14,9 +15,11 @@
     NamespacePath,
     ProjectPath,
 )
-from renku_data_services.data_connectors import apispec, models
+from renku_data_services.data_connectors import apispec, models, schema_org_dataset
+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.storage import models as storage_models
+from renku_data_services.storage.constants import ENVIDAT_V1_PROVIDER
 from renku_data_services.storage.rclone import RCloneValidator
 
 
@@ -41,7 +44,7 @@ def dump_storage_with_sensitive_fields(
     return body
 
 
-def validate_unsaved_storage(
+async def validate_unsaved_storage(
     storage: apispec.CloudStorageCorePost | apispec.CloudStorageUrlV2, validator: RCloneValidator
 ) -> models.CloudStorageCore:
     """Validate the storage configuration of an unsaved data connector."""
@@ -59,6 +62,10 @@ def validate_unsaved_storage(
         )
         configuration = cloud_storage.configuration.config
         source_path = cloud_storage.source_path
+    elif storage.storage_type == ENVIDAT_V1_PROVIDER:
+        converted_storage = await convert_envidat_v1_data_connector_to_s3(storage)
+        configuration = converted_storage.configuration
+        source_path = converted_storage.source_path
     else:
         configuration = storage.configuration
         source_path = storage.source_path
@@ -74,13 +81,13 @@ def validate_unsaved_storage(
     )
 
 
-def validate_unsaved_data_connector(
+async def validate_unsaved_data_connector(
     body: apispec.DataConnectorPost, validator: RCloneValidator
 ) -> models.UnsavedDataConnector:
     """Validate an unsaved data connector."""
 
     keywords = [kw.root for kw in body.keywords] if body.keywords is not None else []
-    storage = validate_unsaved_storage(body.storage, validator=validator)
+    storage = await validate_unsaved_storage(body.storage, validator=validator)
 
     if body.namespace is None:
         raise NotImplementedError("Missing namespace not supported")
@@ -113,20 +120,37 @@ async def prevalidate_unsaved_global_data_connector(
 ) -> models.UnsavedGlobalDataConnector:
     """Pre-validate an unsaved data connector."""
 
-    storage = validate_unsaved_storage(body.storage, validator=validator)
+    storage = await validate_unsaved_storage(body.storage, validator=validator)
     # TODO: allow admins to create global data connectors, e.g. s3://giab
-    if storage.storage_type != "doi":
-        raise errors.ValidationError(message="Only doi storage type is allowed for global data connectors")
+    if storage.storage_type not in ALLOWED_GLOBAL_DATA_CONNECTOR_PROVIDERS:
+        raise errors.ValidationError(
+            message=f"Only {ALLOWED_GLOBAL_DATA_CONNECTOR_PROVIDERS} storage type is allowed for global data connectors"
+        )
     if not storage.readonly:
         raise errors.ValidationError(message="Global data connectors must be read-only")
 
-    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
+    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:
+            raise errors.ValidationError(
+                message=f"Only {ALLOWED_GLOBAL_DATA_CONNECTOR_PROVIDERS} storage type is allowed "
+                "for global data connectors"
+            )
 
     return models.UnsavedGlobalDataConnector(
         name=doi_uri,
@@ -155,8 +179,11 @@ async def validate_unsaved_global_data_connector(
         )
 
     # Fetch DOI metadata
-    rclone_metadata = await validator.get_doi_metadata(configuration=data_connector.storage.configuration)
-    metadata = await get_dataset_metadata(rclone_metadata=rclone_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)
+    else:
+        metadata = None
 
     name = data_connector.name
     description = ""
@@ -325,3 +352,46 @@ def text(self) -> str:
 
     def handle_data(self, data: str) -> None:
         self._text += data
+
+
+async def convert_envidat_v1_data_connector_to_s3(
+    payload: apispec.CloudStorageCorePost,
+) -> apispec.CloudStorageCorePost:
+    """Converts a doi-like configuration for Envidat to S3.
+
+    If the paylaod that is passed in is not of the expected type nothing is changed
+    and the same payload that was passed in is returned.
+    """
+    config = payload.configuration
+    if config.get("type") != ENVIDAT_V1_PROVIDER:
+        return payload
+
+    doi = config.get("doi")
+    if not isinstance(doi, str):
+        raise errors.ValidationError()
+    if len(doi) == 0:
+        raise errors.ValidationError()
+    doi = doi.removeprefix("https://")
+    doi = doi.removeprefix("http://")
+
+    new_config = payload.model_copy(deep=True)
+    new_config.configuration = {}
+
+    envidat_url = "https://envidat.ch/converters-api/internal-dataset/convert/jsonld"
+    query_params = {"query": doi}
+    headers = {"accept": "application/json"}
+
+    clnt = httpx.AsyncClient(follow_redirects=True)
+    async with clnt:
+        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,
+        schema_org_dataset.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
diff --git a/components/renku_data_services/data_connectors/schema_org_dataset.py b/components/renku_data_services/data_connectors/schema_org_dataset.py
@@ -0,0 +1,92 @@
+"""This is used by envidat and scicat to provide information about their datasets."""
+
+from dataclasses import dataclass
+from enum import StrEnum
+from urllib.parse import parse_qs, urlparse
+
+from pydantic import BaseModel, ConfigDict, Field
+
+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."""
+
+    envidat = "envidat"
+
+
+@dataclass
+class S3Config:
+    """Configuration for a location on S3 storage."""
+
+    rclone_config: dict[str, str]
+    bucket: str
+    prefix: str
+
+    @property
+    def path(self) -> str:
+        """Return the path including the bucket name and the prefix."""
+        return f"{self.bucket}/{self.prefix}"
+
+
+def get_rclone_config(dataset: Dataset, provider: DatasetProvider) -> S3Config:
+    """Parse the dataset into an rclone configuration."""
+    match provider:
+        case DatasetProvider.envidat:
+            return __get_rclone_s3_config_envidat(dataset)
+        # TODO: Add scicat here
+        case x:
+            raise errors.ValidationError(message=f"Got an unknown dataset provider {x}")
+
+
+def __get_rclone_s3_config_envidat(dataset: Dataset) -> 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
+    # 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)
+    if not url_parsed.scheme:
+        raise errors.ValidationError(message="A scheme like http or https is needed for the S3 url.")
+    if not url_parsed.netloc:
+        raise errors.ValidationError(message="A hostname is needed for the S3 url.")
+    if not url_parsed.query:
+        raise errors.ValidationError(message="A query parameter with the path is needed for the S3 url.")
+    query_params = parse_qs(url_parsed.query)
+    prefix_list = query_params.get("prefix")
+    if prefix_list is None or len(prefix_list) == 0:
+        raise errors.ValidationError(message="The query paramter in the S3 url should container the 'prefix' key.")
+    prefix = prefix_list[0]
+    host_split = url_parsed.netloc.split(".")
+    if len(host_split) < 2:
+        raise errors.ValidationError(
+            message="The envidat s3 url is expected to have a host name with at least two parts."
+        )
+    s3_host = ".".join(host_split[1:])
+    bucket = host_split[0]
+    prefix = "/" + prefix.strip("/")
+    return S3Config(
+        {
+            "type": "s3",
+            "provider": "Other",
+            "endpoint": f"{url_parsed.scheme}://{s3_host}",
+        },
+        bucket.strip("/"),
+        prefix,
+    )
diff --git a/components/renku_data_services/storage/constants.py b/components/renku_data_services/storage/constants.py
@@ -0,0 +1,6 @@
+"""Constants for storage."""
+
+from typing import Final
+
+ENVIDAT_V1_PROVIDER: Final[str] = "envidat_v1"
+SCICAT_V1_PROVIDER: Final[str] = "scicat_v1"
diff --git a/components/renku_data_services/storage/rclone_patches.py b/components/renku_data_services/storage/rclone_patches.py
@@ -1,9 +1,11 @@
 """Patches to apply to phe rclone storage schema."""
 
+from collections.abc import Callable
 from copy import deepcopy
-from typing import Any, Final
+from typing import Any, Final, cast
 
 from renku_data_services import errors
+from renku_data_services.storage.constants import ENVIDAT_V1_PROVIDER, SCICAT_V1_PROVIDER
 
 BANNED_STORAGE: Final[set[str]] = {
     "alias",
@@ -261,6 +263,28 @@ def __patch_schema_remove_banned_sftp_options(spec: list[dict[str, Any]]) -> Non
     sftp["Options"] = options
 
 
+def __add_custom_doi_s3_provider(name: str, description: str, prefix: str) -> Callable[[list[dict[str, Any]]], None]:
+    """This is used to add envidata and scicat as providers.
+
+    However this is not a real provider in Rclone. The data service has to intercept the request
+    and convert this provider to the proper S3 configuration where the data can be found.
+    """
+
+    def __patch(spec: list[dict[str, Any]]) -> None:
+        doi_original = find_storage(spec, "doi")
+        doi_new = deepcopy(doi_original)
+        doi_new["Description"] = description
+        doi_new["Name"] = name
+        doi_new["Prefix"] = prefix
+        doi_new_options = cast(list[dict[str, Any]], doi_new.get("Options", []))
+        provider_ind = next((i for i, opt in enumerate(doi_new_options) if opt.get("Name") == "provider"), None)
+        if provider_ind is not None:
+            doi_new_options.pop(provider_ind)
+        spec.append(doi_new)
+
+    return __patch
+
+
 def apply_patches(spec: list[dict[str, Any]]) -> None:
     """Apply patches to RClone schema."""
     patches = [
@@ -271,6 +295,8 @@ def apply_patches(spec: list[dict[str, Any]]) -> None:
         __patch_schema_remove_oauth_propeties,
         __patch_polybox_storage,
         __patch_switchdrive_storage,
+        __add_custom_doi_s3_provider("Envidat", "Envidat data provider", ENVIDAT_V1_PROVIDER),
+        __add_custom_doi_s3_provider("SciCat", "SciCat data provider", SCICAT_V1_PROVIDER),
         __patch_schema_remove_banned_sftp_options,
     ]
 

Original file line number	Diff line number	Diff line change
`@@ -99,7 +99,7 @@ def post(self) -> BlueprintFactoryResponse:`
`99`	`99`	`async def _post(`
`100`	`100`	`_: Request, user: base_models.APIUser, body: apispec.DataConnectorPost, validator: RCloneValidator`
`101`	`101`	`) -> JSONResponse:`
`102`		`- data_connector = validate_unsaved_data_connector(body, validator=validator)`
	`102`	`+ data_connector = await validate_unsaved_data_connector(body, validator=validator)`
`103`	`103`	`result = await self.data_connector_repo.insert_namespaced_data_connector(`
`104`	`104`	`user=user, data_connector=data_connector`
`105`	`105`	`)`