18715077147

Committed 22 Oct 2025 11:45AM UTC coverage: 86.653% (-0.1%) from 86.802%

Build # 18715077147

Build Type

Pull #945

github

Committed by

web-flow

Commit Message

Merge 93c973951 into 5cc2b39de

Pull Request Pull Request #945: feat: add support for OpenBIS datasets

Run Details

70 of 139 new or added lines in 12 files covered. (50.36%)

4 existing lines in 3 files now uncovered.

22781 of 26290 relevant lines covered (86.65%)

1.52 hits per line

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

88.11

/components/renku_data_services/data_connectors/blueprints.py

"""Data connectors blueprint."""

from dataclasses import dataclass
from datetime import datetime
from typing import Any

from sanic import Request
from sanic.response import HTTPResponse, JSONResponse
from sanic_ext import validate
from ulid import ULID

from renku_data_services import base_models, errors
from renku_data_services.base_api.auth import (
    authenticate,
    only_authenticated,
)
from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint
from renku_data_services.base_api.etag import extract_if_none_match, if_match_required
from renku_data_services.base_api.misc import validate_body_root_model, validate_query
from renku_data_services.base_api.pagination import PaginationRequest, paginate
from renku_data_services.base_models.core import (
    DataConnectorInProjectPath,
    DataConnectorPath,
    NamespacePath,
    ProjectPath,
    Slug,
)
from renku_data_services.base_models.metrics import MetricsService
from renku_data_services.base_models.validation import validate_and_dump, validated_json
from renku_data_services.data_connectors import apispec, models
from renku_data_services.data_connectors.core import (
    dump_storage_with_sensitive_fields,
    prevalidate_unsaved_global_data_connector,
    validate_data_connector_patch,
    validate_data_connector_secrets_patch,
    validate_unsaved_data_connector,
)
from renku_data_services.data_connectors.db import (
    DataConnectorRepository,
    DataConnectorSecretRepository,
)
from renku_data_services.storage.rclone import RCloneValidator
from renku_data_services.utils.core import get_openbis_pat


@dataclass(kw_only=True)
class DataConnectorsBP(CustomBlueprint):
    """Handlers for manipulating data connectors."""

    data_connector_repo: DataConnectorRepository
    data_connector_secret_repo: DataConnectorSecretRepository
    authenticator: base_models.Authenticator
    metrics: MetricsService

    def get_all(self) -> BlueprintFactoryResponse:
        """List data connectors."""

        @authenticate(self.authenticator)
        @validate_query(query=apispec.DataConnectorsGetQuery)
        @paginate
        async def _get_all(
            _: Request,
            user: base_models.APIUser,
            pagination: PaginationRequest,
            query: apispec.DataConnectorsGetQuery,
            validator: RCloneValidator,
        ) -> tuple[list[dict[str, Any]], int]:
            ns_segments = query.namespace.split("/")
            ns: None | NamespacePath | ProjectPath
            if len(ns_segments) == 0 or (len(ns_segments) == 1 and len(ns_segments[0]) == 0):
                ns = None
            elif len(ns_segments) == 1 and len(ns_segments[0]) > 0:
                ns = NamespacePath.from_strings(*ns_segments)
            elif len(ns_segments) == 2:
                ns = ProjectPath.from_strings(*ns_segments)
            else:
                raise errors.ValidationError(
                    message="Got an unexpected number of path segments for the data connector namespace"
                    " in the request query parameter, expected 0, 1 or 2"
                )
            data_connectors, total_num = await self.data_connector_repo.get_data_connectors(
                user=user, pagination=pagination, namespace=ns
            )
            return [
                validate_and_dump(
                    apispec.DataConnector,
                    self._dump_data_connector(dc, validator=validator),
                )
                for dc in data_connectors
            ], total_num

        return "/data_connectors", ["GET"], _get_all

    def post(self) -> BlueprintFactoryResponse:
        """Create a new data connector."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate(json=apispec.DataConnectorPost)
        async def _post(
            _: Request, user: base_models.APIUser, body: apispec.DataConnectorPost, validator: RCloneValidator
        ) -> JSONResponse:
            data_connector = validate_unsaved_data_connector(body, validator=validator)
            result = await self.data_connector_repo.insert_namespaced_data_connector(
                user=user, data_connector=data_connector
            )
            await self.metrics.data_connector_created(user)
            return validated_json(
                apispec.DataConnector,
                self._dump_data_connector(result, validator=validator),
                status=201,
            )

        return "/data_connectors", ["POST"], _post

    def post_global(self) -> BlueprintFactoryResponse:
        """Create a new global data connector."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate(json=apispec.GlobalDataConnectorPost)
        async def _post_global(
            _: Request, user: base_models.APIUser, body: apispec.GlobalDataConnectorPost, validator: RCloneValidator
        ) -> JSONResponse:
            data_connector = await prevalidate_unsaved_global_data_connector(body, validator=validator)
            result, inserted = await self.data_connector_repo.insert_global_data_connector(
                user=user, data_connector=data_connector, validator=validator
            )
            return validated_json(
                apispec.DataConnector,
                self._dump_data_connector(result, validator=validator),
                status=201 if inserted else 200,
            )

        return "/data_connectors/global", ["POST"], _post_global

    def get_one(self) -> BlueprintFactoryResponse:
        """Get a specific data connector."""

        @authenticate(self.authenticator)
        @extract_if_none_match
        async def _get_one(
            _: Request, user: base_models.APIUser, data_connector_id: ULID, etag: str | None, validator: RCloneValidator
        ) -> HTTPResponse:
            data_connector = await self.data_connector_repo.get_data_connector(
                user=user, data_connector_id=data_connector_id
            )

            if data_connector.etag == etag:
                return HTTPResponse(status=304)

            headers = {"ETag": data_connector.etag}
            return validated_json(
                apispec.DataConnector,
                self._dump_data_connector(data_connector, validator=validator),
                headers=headers,
            )

        return "/data_connectors/<data_connector_id:ulid>", ["GET"], _get_one

    def get_one_global_by_slug(self) -> BlueprintFactoryResponse:
        """Get a specific global data connector by slug."""

        @authenticate(self.authenticator)
        @extract_if_none_match
        async def _get_one_global_by_slug(
            _: Request,
            user: base_models.APIUser,
            slug: Slug,
            etag: str | None,
            validator: RCloneValidator,
        ) -> HTTPResponse:
            data_connector = await self.data_connector_repo.get_global_data_connector_by_slug(user=user, slug=slug)

            if data_connector.etag == etag:
                return HTTPResponse(status=304)

            headers = {"ETag": data_connector.etag}
            return validated_json(
                apispec.DataConnector,
                self._dump_data_connector(data_connector, validator=validator),
                headers=headers,
            )

        return "/data_connectors/global/<slug:renku_slug>", ["GET"], _get_one_global_by_slug

    def get_one_by_slug(self) -> BlueprintFactoryResponse:
        """Get a specific data connector by namespace/entity slug."""

        @authenticate(self.authenticator)
        @extract_if_none_match
        async def _get_one_by_slug(
            _: Request,
            user: base_models.APIUser,
            namespace: str,
            slug: Slug,
            etag: str | None,
            validator: RCloneValidator,
        ) -> HTTPResponse:
            data_connector = await self.data_connector_repo.get_data_connector_by_slug(
                user=user,
                path=DataConnectorPath.from_strings(namespace, slug.value),
            )

            if data_connector.etag == etag:
                return HTTPResponse(status=304)

            headers = {"ETag": data_connector.etag}
            return validated_json(
                apispec.DataConnector,
                self._dump_data_connector(data_connector, validator=validator),
                headers=headers,
            )

        return "/namespaces/<namespace>/data_connectors/<slug:renku_slug>", ["GET"], _get_one_by_slug

    def get_one_by_slug_from_project_namespace(self) -> BlueprintFactoryResponse:
        """Get a specific data connector by namespace/project_slug/dc_slug slug."""

        @authenticate(self.authenticator)
        @extract_if_none_match
        async def _get_one_from_project_namespace(
            _: Request,
            user: base_models.APIUser,
            ns_slug: Slug,
            project_slug: Slug,
            dc_slug: Slug,
            etag: str | None,
            validator: RCloneValidator,
        ) -> HTTPResponse:
            dc_path = DataConnectorInProjectPath.from_strings(
                ns_slug.value,
                project_slug.value,
                dc_slug.value,
            )
            data_connector = await self.data_connector_repo.get_data_connector_by_slug(user=user, path=dc_path)

            if data_connector.etag == etag:
                return HTTPResponse(status=304)

            headers = {"ETag": data_connector.etag}
            return validated_json(
                apispec.DataConnector,
                self._dump_data_connector(data_connector, validator=validator),
                headers=headers,
            )

        return (
            "/namespaces/<ns_slug:renku_slug>/projects/<project_slug:renku_slug>/data_connectors/<dc_slug:renku_slug>",
            ["GET"],
            _get_one_from_project_namespace,
        )

    def patch(self) -> BlueprintFactoryResponse:
        """Partially update a data connector."""

        @authenticate(self.authenticator)
        @only_authenticated
        @if_match_required
        @validate(json=apispec.DataConnectorPatch)
        async def _patch(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
            body: apispec.DataConnectorPatch,
            etag: str,
            validator: RCloneValidator,
        ) -> JSONResponse:
            existing_dc = await self.data_connector_repo.get_data_connector(
                user=user, data_connector_id=data_connector_id
            )
            dc_patch = validate_data_connector_patch(existing_dc, body, validator=validator)
            data_connector_update = await self.data_connector_repo.update_data_connector(
                user=user, data_connector_id=data_connector_id, patch=dc_patch, etag=etag
            )

            return validated_json(
                apispec.DataConnector,
                self._dump_data_connector(data_connector_update.new, validator=validator),
            )

        return "/data_connectors/<data_connector_id:ulid>", ["PATCH"], _patch

    def delete(self) -> BlueprintFactoryResponse:
        """Delete a data connector."""

        @authenticate(self.authenticator)
        @only_authenticated
        async def _delete(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
        ) -> HTTPResponse:
            await self.data_connector_repo.delete_data_connector(user=user, data_connector_id=data_connector_id)
            return HTTPResponse(status=204)

        return "/data_connectors/<data_connector_id:ulid>", ["DELETE"], _delete

    def get_permissions(self) -> BlueprintFactoryResponse:
        """Get the permissions of the current user on the data connector."""

        @authenticate(self.authenticator)
        async def _get_permissions(_: Request, user: base_models.APIUser, data_connector_id: ULID) -> JSONResponse:
            permissions = await self.data_connector_repo.get_data_connector_permissions(
                user=user, data_connector_id=data_connector_id
            )
            return validated_json(apispec.DataConnectorPermissions, permissions)

        return "/data_connectors/<data_connector_id:ulid>/permissions", ["GET"], _get_permissions

    def get_all_project_links(self) -> BlueprintFactoryResponse:
        """List all links from a given data connector to projects."""

        @authenticate(self.authenticator)
        async def _get_all_project_links(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
        ) -> JSONResponse:
            links = await self.data_connector_repo.get_links_from(user=user, data_connector_id=data_connector_id)
            return validated_json(
                apispec.DataConnectorToProjectLinksList,
                [self._dump_data_connector_to_project_link(link) for link in links],
            )

        return "/data_connectors/<data_connector_id:ulid>/project_links", ["GET"], _get_all_project_links

    def post_project_link(self) -> BlueprintFactoryResponse:
        """Create a new link from a data connector to a project."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate(json=apispec.DataConnectorToProjectLinkPost)
        async def _post_project_link(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
            body: apispec.DataConnectorToProjectLinkPost,
        ) -> JSONResponse:
            unsaved_link = models.UnsavedDataConnectorToProjectLink(
                data_connector_id=data_connector_id,
                project_id=ULID.from_str(body.project_id),
            )
            link = await self.data_connector_repo.insert_link(user=user, link=unsaved_link)
            await self.metrics.data_connector_linked(user)
            return validated_json(
                apispec.DataConnectorToProjectLink, self._dump_data_connector_to_project_link(link), status=201
            )

        return "/data_connectors/<data_connector_id:ulid>/project_links", ["POST"], _post_project_link

    def delete_project_link(self) -> BlueprintFactoryResponse:
        """Delete a link from a data connector to a project."""

        @authenticate(self.authenticator)
        @only_authenticated
        async def _delete_project_link(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
            link_id: ULID,
        ) -> HTTPResponse:
            await self.data_connector_repo.delete_link(user=user, data_connector_id=data_connector_id, link_id=link_id)
            return HTTPResponse(status=204)

        return (
            "/data_connectors/<data_connector_id:ulid>/project_links/<link_id:ulid>",
            ["DELETE"],
            _delete_project_link,
        )

    def get_all_data_connectors_links_to_project(self) -> BlueprintFactoryResponse:
        """List all links from data connectors to a given project."""

        @authenticate(self.authenticator)
        async def _get_all_data_connectors_links_to_project(
            _: Request,
            user: base_models.APIUser,
            project_id: ULID,
        ) -> JSONResponse:
            links = await self.data_connector_repo.get_links_to(user=user, project_id=project_id)
            return validated_json(
                apispec.DataConnectorToProjectLinksList,
                [self._dump_data_connector_to_project_link(link) for link in links],
            )

        return "/projects/<project_id:ulid>/data_connector_links", ["GET"], _get_all_data_connectors_links_to_project

    def get_inaccessible_data_connectors_links_to_project(self) -> BlueprintFactoryResponse:
        """The number of data connector links in a given project the user has no access to."""

        @authenticate(self.authenticator)
        async def _get_inaccessible_data_connectors_links_to_project(
            _: Request,
            user: base_models.APIUser,
            project_id: ULID,
        ) -> JSONResponse:
            link_ids = await self.data_connector_repo.get_inaccessible_links_to_project(
                user=user, project_id=project_id
            )
            return validated_json(apispec.InaccessibleDataConnectorLinks, {"count": len(link_ids)})

        return (
            "/projects/<project_id:ulid>/inaccessible_data_connector_links",
            ["GET"],
            _get_inaccessible_data_connectors_links_to_project,
        )

    def get_secrets(self) -> BlueprintFactoryResponse:
        """List all saved secrets for a data connector."""

        @authenticate(self.authenticator)
        @only_authenticated
        async def _get_secrets(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
        ) -> JSONResponse:
            secrets = await self.data_connector_secret_repo.get_data_connector_secrets(
                user=user, data_connector_id=data_connector_id
            )
            data_connector = await self.data_connector_repo.get_data_connector(
                user=user, data_connector_id=data_connector_id
            )
            return validated_json(
                apispec.DataConnectorSecretsList,
                [
                    self._dump_data_connector_secret(secret)
                    for secret in self._adjust_secrets(secrets, data_connector.storage)
                ],
            )

        return "/data_connectors/<data_connector_id:ulid>/secrets", ["GET"], _get_secrets

    def patch_secrets(self) -> BlueprintFactoryResponse:
        """Create, update or delete saved secrets for a data connector."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate_body_root_model(json=apispec.DataConnectorSecretPatchList)
        async def _patch_secrets(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
            body: apispec.DataConnectorSecretPatchList,
            validator: RCloneValidator,
        ) -> JSONResponse:
            unsaved_secrets = validate_data_connector_secrets_patch(put=body)
            data_connector = await self.data_connector_repo.get_data_connector(
                user=user, data_connector_id=data_connector_id
            )
            storage = data_connector.storage
            provider = validator.providers[storage.storage_type]
            sensitive_lookup = [o.name for o in provider.options if o.sensitive]
            for secret in unsaved_secrets:
                if secret.name in sensitive_lookup:
                    continue
                raise errors.ValidationError(
                    message=f"The '{secret.name}' property is not marked sensitive and can not be saved in the secret "
                    f"storage."
                )
            expiration_timestamp = None

            if storage.storage_type == "openbis":

                async def openbis_transform_session_token_to_pat() -> (
                    tuple[list[models.DataConnectorSecretUpdate], datetime]
                ):
                    if len(unsaved_secrets) == 1 and unsaved_secrets[0].name == "session_token":
                        if unsaved_secrets[0].value is not None:
                            try:
                                openbis_pat = await get_openbis_pat(
                                    storage.configuration["host"], unsaved_secrets[0].value
                                )
                                return (
                                    [models.DataConnectorSecretUpdate(name="pass", value=openbis_pat[0])],
                                    openbis_pat[1],
                                )
                            except Exception as e:
                                raise errors.ProgrammingError(message=str(e)) from e
                        raise errors.ValidationError(message="The openBIS session token must be a string value.")

                    raise errors.ValidationError(message="The openBIS storage has only one secret: session_token")

                (
                    unsaved_secrets,
                    expiration_timestamp,
                ) = await openbis_transform_session_token_to_pat()

            secrets = await self.data_connector_secret_repo.patch_data_connector_secrets(
                user=user,
                data_connector_id=data_connector_id,
                secrets=unsaved_secrets,
                expiration_timestamp=expiration_timestamp,
            )
            return validated_json(
                apispec.DataConnectorSecretsList,
                [self._dump_data_connector_secret(secret) for secret in self._adjust_secrets(secrets, storage)],
            )

        return "/data_connectors/<data_connector_id:ulid>/secrets", ["PATCH"], _patch_secrets

    def delete_secrets(self) -> BlueprintFactoryResponse:
        """Delete all saved secrets for a data connector."""

        @authenticate(self.authenticator)
        @only_authenticated
        async def _delete_secrets(
            _: Request,
            user: base_models.APIUser,
            data_connector_id: ULID,
        ) -> HTTPResponse:
            await self.data_connector_secret_repo.delete_data_connector_secrets(
                user=user, data_connector_id=data_connector_id
            )
            return HTTPResponse(status=204)

        return "/data_connectors/<data_connector_id:ulid>/secrets", ["DELETE"], _delete_secrets

    @staticmethod
    def _dump_data_connector(
        data_connector: models.DataConnector | models.GlobalDataConnector, validator: RCloneValidator
    ) -> dict[str, Any]:
        """Dumps a data connector for API responses."""
        storage = dump_storage_with_sensitive_fields(data_connector.storage, validator=validator)
        if data_connector.namespace is None:
            return dict(
                id=str(data_connector.id),
                name=data_connector.name,
                slug=data_connector.slug,
                storage=storage,
                creation_date=data_connector.creation_date,
                created_by=data_connector.created_by,
                visibility=data_connector.visibility.value,
                description=data_connector.description,
                etag=data_connector.etag,
                keywords=data_connector.keywords or [],
            )
        return dict(
            id=str(data_connector.id),
            name=data_connector.name,
            namespace=data_connector.namespace.path.serialize(),
            slug=data_connector.slug,
            storage=storage,
            # secrets=,
            creation_date=data_connector.creation_date,
            created_by=data_connector.created_by,
            visibility=data_connector.visibility.value,
            description=data_connector.description,
            etag=data_connector.etag,
            keywords=data_connector.keywords or [],
        )

    @staticmethod
    def _dump_data_connector_to_project_link(link: models.DataConnectorToProjectLink) -> dict[str, Any]:
        """Dumps a link from a data connector to a project for API responses."""
        return dict(
            id=str(link.id),
            data_connector_id=str(link.data_connector_id),
            project_id=str(link.project_id),
            creation_date=link.creation_date,
            created_by=link.created_by,
        )

    @staticmethod
    def _adjust_secrets(
        secrets: list[models.DataConnectorSecret], storage: models.CloudStorageCore
    ) -> list[models.DataConnectorSecret]:
        if storage.storage_type == "openbis":
            for i, secret in enumerate(secrets):
                if secret.name == "pass":
                    secrets[i] = models.DataConnectorSecret(
                        name="session_token",
                        user_id=secret.user_id,
                        data_connector_id=secret.data_connector_id,
                        secret_id=secret.secret_id,
                    )
                    break
        return secrets

    @staticmethod
    def _dump_data_connector_secret(secret: models.DataConnectorSecret) -> dict[str, Any]:
        """Dumps a data connector secret for API responses."""
        return dict(
            name=secret.name,
            secret_id=str(secret.secret_id),
        )

SwissDataScienceCenter / renku-data-services / 18715077147

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