feat: export to envidat (#1329)

Co-authored-by: Tasko Olevski <16360283+olevski@users.noreply.github.com>

Author: mohammad-alisafaee

bases/renku_data_services/data_api/app.py

@@ -276,6 +276,7 @@ def register_all_handlers(app: Sanic, dm: DependencyManager) -> Sanic:
         authenticator=dm.authenticator,
         metrics=dm.metrics,
         zenodo_client=dm.zenodo_client,
+        envidat_client=dm.envidat_client,
         connected_services_repo=dm.connected_services_repo,
         job_client=dm.job_client,
         secret_client=dm.secret_client,

bases/renku_data_services/data_api/dependencies.py

@@ -38,6 +38,7 @@
     DataConnectorRepository,
     DataConnectorSecretRepository,
 )
+from renku_data_services.data_connectors.deposits.envidat import EnvidatClient
 from renku_data_services.data_connectors.deposits.zenodo import ZenodoAPIClient
 from renku_data_services.git.gitlab import DummyGitlabAPI, EmptyGitlabAPI, GitlabAPI
 from renku_data_services.k8s.client_interfaces import K8sClient
@@ -171,6 +172,7 @@ class DependencyManager:
     resource_requests_repo: ResourceRequestsRepo
     resource_usage_service: ResourceUsageService
     zenodo_client: ZenodoAPIClient
+    envidat_client: EnvidatClient
     job_client: DepositUploadJobClient
     secret_client: K8sSecretClient
     internal_token_mint: RenkuSelfTokenMint
@@ -515,6 +517,7 @@ def from_env(cls) -> DependencyManager:
             resource_requests_repo=resource_requests_repo,
             resource_usage_service=resource_usage_service,
             zenodo_client=ZenodoAPIClient(),
+            envidat_client=EnvidatClient(),
             job_client=job_client,
             secret_client=secret_client,
             internal_token_mint=internal_token_mint,

components/renku_data_services/data_connectors/api.spec.yaml

@@ -1195,6 +1195,7 @@ components:
       type: string
       enum:
         - zenodo
+        - envidat
     DepositStatus:
       type: string
       enum:

components/renku_data_services/data_connectors/apispec.py

@@ -141,6 +141,7 @@ class InaccessibleDataConnectorLinks(BaseAPISpec):
 
 class DepositProvider(StrEnum):
     zenodo = "zenodo"
+    envidat = "envidat"
 
 
 class DepositStatus(StrEnum):

components/renku_data_services/data_connectors/blueprints.py

@@ -51,6 +51,7 @@
     DataConnectorRepository,
     DataConnectorSecretRepository,
 )
+from renku_data_services.data_connectors.deposits.envidat import EnvidatClient
 from renku_data_services.data_connectors.deposits.zenodo import ZenodoAPIClient
 from renku_data_services.k8s.client_interfaces import K8sClient, SecretClient
 from renku_data_services.k8s.clients import DepositUploadJobClient
@@ -69,6 +70,7 @@ class DataConnectorsBP(CustomBlueprint):
     job_client: DepositUploadJobClient
     secret_client: SecretClient
     zenodo_client: ZenodoAPIClient
+    envidat_client: EnvidatClient
     connected_services_repo: ConnectedServicesRepository
     data_source_repo: DataSourceRepository
     dc_storage_class: str
@@ -599,19 +601,19 @@ async def __get_zenodo_access_token(self, user: base_models.APIUser) -> str:
             )
         if not provider.connected_user:
             raise errors.UnauthorizedError(
-                message="You need to connect and autheticate with the zenodo provider to do this"
+                message="You need to connect and authenticate with the zenodo provider to do this"
             )
         token_set = await self.connected_services_repo.get_token_set(
             user=user, connection_id=provider.connected_user.connection.id
         )
         if not token_set:
             raise errors.UnauthorizedError(
-                message="You need to connect and autheticate with the zenodo provider to do this"
+                message="You need to connect and authenticate with the zenodo provider to do this"
             )
         access_token = token_set.access_token
         if not access_token:
             raise errors.UnauthorizedError(
-                message="You need to connect and autheticate with the zenodo provider to do this"
+                message="You need to connect and authenticate with the zenodo provider to do this"
             )
         return access_token
 
@@ -626,15 +628,33 @@ async def _post_deposit(
             user: base_models.AuthenticatedAPIUser,
             body: apispec.DepositPost,
         ) -> JSONResponse:
-            existing_deposits, _ = await self.data_connector_repo.get_deposits(
-                user, ULID.from_str(body.data_connector_id)
-            )
+            dc_id = ULID.from_str(body.data_connector_id)
+            await self.data_connector_repo.get_data_connector(user=user, data_connector_id=dc_id)
+            existing_deposits, _ = await self.data_connector_repo.get_deposits(user, dc_id)
             if len(existing_deposits) != 0:
                 raise errors.ValidationError(
                     message="Cannot have more than 1 deposit for the same user and data connector.",
                     detail="Please delete your existing deposit and make a new one afterward.",
                 )
-            token = await self.__get_zenodo_access_token(user)
+
+            match body.provider:
+                case apispec.DepositProvider.envidat:
+                    # TODO: Should we use the deposit ULID as the directory name!?
+                    # NOTE: Each new Envidat deposit gets its own dir, so a failed deposit cannot be resumed and need to
+                    # be cleaned up manually on Envidat.
+                    original_id = str(ULID())
+                    deposit_api_key: str | None = None
+
+                case apispec.DepositProvider.zenodo:
+                    token = await self.__get_zenodo_access_token(user)
+                    zenodo_dep = await self.zenodo_client.create_deposit(token, body.name)
+                    original_id = str(zenodo_dep.id)
+                    deposit_api_key = token
+
+                case x:
+                    raise errors.ValidationError(
+                        message=f"Received unknown deposit provider {x} when creating deposit."
+                    )
 
             # The closure below allows us to tie the creation of the db entry to the successful
             # creation of the job in kubernetes. I.e. if the k8s creation fails nothing is saved
@@ -654,13 +674,12 @@ async def in_transaction_ops(
                     job_client=self.job_client,
                     data_connector_secret_repo=self.data_connector_secret_repo,
                     data_source_repo=self.data_source_repo,
-                    deposit_api_key=token,
+                    deposit_api_key=deposit_api_key,
                 )
 
-            zenodo_dep = await self.zenodo_client.create_deposit(token, body.name)
-            unsaved_dep = validate_deposit(body, str(zenodo_dep.id))
+            unsaved_dep = validate_deposit(body, original_id)
             saved_dep = await self.data_connector_repo.create_deposit(user, unsaved_dep, in_transaction_ops)
-            return validated_json(apispec.Deposit, serialize_deposit(saved_dep))
+            return validated_json(apispec.Deposit, serialize_deposit(saved_dep, self.deposit_config))
 
         return "/deposits", ["POST"], _post_deposit
 
@@ -684,7 +703,7 @@ async def _get_deposit(
                 return HTTPResponse(status=304)
 
             headers = {"ETag": saved_dep.etag}
-            return validated_json(apispec.Deposit, serialize_deposit(saved_dep), headers=headers)
+            return validated_json(apispec.Deposit, serialize_deposit(saved_dep, self.deposit_config), headers=headers)
 
         return "/deposits/<deposit_id:ulid>", ["GET"], _get_deposit
 
@@ -713,23 +732,35 @@ async def _patch_deposit(
             if patch.status:
                 validate_deposit_status_change(saved_dep.deposit.status, patch.status)
             if patch.status == models.DepositStatus.complete:
-                token = await self.__get_zenodo_access_token(user)
-                zenodo_dep = await self.zenodo_client.get_deposit(token, saved_dep.deposit.original_id)
-                if not zenodo_dep:
-                    raise errors.MissingResourceError(
-                        message=f"The deposit with id {saved_dep.deposit.original_id} "
-                        "cannot be found from the provider."
-                    )
-                if not zenodo_dep.submitted:
-                    raise errors.ValidationError(
-                        message="The deposit needs to be completed and published first before being completed."
-                    )
+                match saved_dep.deposit.source:
+                    case models.DepositSource.envidat:
+                        envidat_status = await self.envidat_client.get_deposit_status(saved_dep.deposit.original_id)
+                        if not envidat_status.is_published:
+                            raise errors.ValidationError(
+                                message="The deposit needs to be published on Envidat before being marked complete."
+                            )
+                    case models.DepositSource.zenodo:
+                        token = await self.__get_zenodo_access_token(user)
+                        zenodo_dep = await self.zenodo_client.get_deposit(token, saved_dep.deposit.original_id)
+                        if not zenodo_dep:
+                            raise errors.MissingResourceError(
+                                message=f"The Zenodo deposit with id {saved_dep.deposit.original_id} cannot be found."
+                            )
+                        if not zenodo_dep.submitted:
+                            raise errors.ValidationError(
+                                message="The Zenodo deposit needs to be completed and published first "
+                                "before being completed."
+                            )
+                    case x:
+                        raise errors.ValidationError(
+                            message=f"Received unknown deposit source {x} when pathing deposit with ID {deposit_id}."
+                        )
             saved_dep = await self.data_connector_repo.update_deposit(user, deposit_id, patch, etag=etag)
             if patch.status == models.DepositStatus.complete:
                 # If the deposit is being completed then we delete it
                 # We leave it to the user to create a new data connector and link it
                 await self.data_connector_repo.delete_deposit(user, saved_dep.deposit.id)
-            return validated_json(apispec.Deposit, serialize_deposit(saved_dep))
+            return validated_json(apispec.Deposit, serialize_deposit(saved_dep, self.deposit_config))
 
         return "/deposits/<deposit_id:ulid>", ["PATCH"], _patch_deposit
 
@@ -741,14 +772,22 @@ def post_deposit_job(self) -> BlueprintFactoryResponse:
         async def _post_deposit_job(
             request: Request, user: base_models.AuthenticatedAPIUser, deposit_id: ULID
         ) -> HTTPResponse:
-            token = await self.__get_zenodo_access_token(user)
             saved_dep = await update_deposit_status(
                 user,
                 job=deposit_id,
                 dc_repo=self.data_connector_repo,
                 job_client=self.job_client,
                 namespace=self.deposit_config.namespace,
             )
+            if saved_dep.deposit.source == models.DepositSource.envidat:
+                raise errors.ValidationError(
+                    message="Envidat deposits cannot be retried. Please delete this deposit and create a new one.",
+                    detail="Each Envidat deposit uploads to a unique S3 directory identified by the deposit ID. "
+                    "Starting over with a new deposit ensures a clean upload without mixing partial files.",
+                )
+            if saved_dep.deposit.status == models.DepositStatus.in_progress:
+                raise errors.ValidationError(message="Cannot rerun a deposit job that is currently in progress.")
+            token = await self.__get_zenodo_access_token(user)
             await self.job_client.delete(saved_dep.to_meta(user.id, self.deposit_config.namespace))
             new_job_name = "deposit-" + str(ULID()).lower()
             saved_dep = await self.data_connector_repo.update_deposit(
@@ -817,7 +856,9 @@ async def _get_deposits(
                 job_client=self.job_client,
                 namespace=self.deposit_config.namespace,
             )
-            return [validate_and_dump(apispec.Deposit, serialize_deposit(i)) for i in deposits], total_num
+            return [
+                validate_and_dump(apispec.Deposit, serialize_deposit(i, self.deposit_config)) for i in deposits
+            ], total_num
 
         return "/deposits", ["GET"], _get_deposits
 
@@ -838,7 +879,9 @@ async def _get_dc_deposits(
                 job_client=self.job_client,
                 namespace=self.deposit_config.namespace,
             )
-            return [validate_and_dump(apispec.Deposit, serialize_deposit(i)) for i in deposits], total_num
+            return [
+                validate_and_dump(apispec.Deposit, serialize_deposit(i, self.deposit_config)) for i in deposits
+            ], total_num
 
         return "/data_connectors/<data_connector_id:ulid>/deposits", ["GET"], _get_dc_deposits
 

components/renku_data_services/data_connectors/config.py

@@ -11,6 +11,7 @@
 from kubernetes.client import ApiClient, V1Toleration
 
 from renku_data_services.app_config import logging
+from renku_data_services.errors import errors
 from renku_data_services.k8s.constants import DEFAULT_K8S_CLUSTER, ClusterId
 
 logger = logging.getLogger(__name__)
@@ -24,6 +25,7 @@ class DepositConfig:
     namespace: str
     renku_url: str
     zenodo_url: str
+    envidat: EnvidatConfig
     node_selector: dict[str, str] | None = None
     tolerations: list[V1Toleration] | None = None
     cluster_id: Final[ClusterId] = DEFAULT_K8S_CLUSTER
@@ -64,4 +66,47 @@ def from_env(cls, renku_url: str) -> DepositConfig:
             namespace=os.environ["KUBERNETES_NAMESPACE"],
             cluster_id=DEFAULT_K8S_CLUSTER,
             zenodo_url=os.environ.get("ZENODO_URL", "https://zenodo.org").rstrip("/"),
+            envidat=EnvidatConfig.from_env(),
         )
+
+
+@dataclass
+class EnvidatConfig:
+    """Configuration for envidat data exports and imports."""
+
+    exports_enabled: bool
+    url: str
+    rclone_image: str
+    s3_endpoint: str
+    s3_bucket: str
+    s3_access_key_id: str
+    s3_secret_access_key: str
+
+    @classmethod
+    def from_env(cls) -> EnvidatConfig:
+        """Generate the config from environment variables."""
+        exports_enabled = os.environ.get("ENVIDAT_EXPORTS_ENABLED", "false").lower() == "true"
+        output = cls(
+            exports_enabled=exports_enabled,
+            url=os.environ.get("ENVIDAT_URL", "https://www.envidat.ch").rstrip("/"),
+            rclone_image=os.environ.get("ENVIDAT_RCLONE_IMAGE", "rclone/rclone:1"),
+            s3_endpoint=os.environ.get("ENVIDAT_S3_ENDPOINT", ""),
+            s3_bucket=os.environ.get("ENVIDAT_S3_BUCKET", ""),
+            s3_access_key_id=os.environ.get("ENVIDAT_S3_ACCESS_KEY_ID", ""),
+            s3_secret_access_key=os.environ.get("ENVIDAT_S3_SECRET_ACCESS_KEY", ""),
+        )
+        if exports_enabled and any(
+            [
+                i == ""
+                for i in [
+                    output.s3_endpoint,
+                    output.s3_bucket,
+                    output.s3_access_key_id,
+                    output.s3_secret_access_key,
+                ]
+            ]
+        ):
+            raise errors.ConfigurationError(
+                message="Envidat exports are enabled but not all required parameters are provided."
+            )
+        return output