change schema loading

renku_notebooks/api/notebooks.py

@@ -27,6 +27,7 @@
 from webargs.flaskparser import use_args
 
 from renku_notebooks.api.classes.user import AnonymousUser
+from renku_notebooks.api.schemas.cloud_storage import RCloneStorage
 from renku_notebooks.util.repository import get_status
 
 from ..config import config
@@ -318,15 +319,20 @@ def launch_notebook(
 
     if cloudstorage:
         gl_project_id = gl_project.id if gl_project is not None else 0
+        storages = []
         try:
             for storage in cloudstorage:
-                storage.init_config(
-                    user=user,
-                    project_id=gl_project_id,
-                    work_dir=server_work_dir.absolute(),
+                storages.append(
+                    RCloneStorage.storage_from_schema(
+                        storage,
+                        user=user,
+                        project_id=gl_project_id,
+                        work_dir=server_work_dir.absolute(),
+                    )
                 )
         except ValidationError as e:
             raise UserInputError(f"Couldn't load cloud storage config: {str(e)}")
+        cloudstorage = storages
         mount_points = set(
             s.mount_folder for s in cloudstorage if s.mount_folder and s.mount_folder != "/"
         )

renku_notebooks/api/schemas/cloud_storage.py

@@ -5,8 +5,6 @@
 
 from marshmallow import EXCLUDE, Schema, ValidationError, fields, validates_schema
 
-from renku_notebooks.errors.programming import ProgrammingError
-
 from ...config import config
 from ..classes.user import User
 
@@ -31,34 +29,41 @@ def validate_storage(self, data, **kwargs):
                 "'storage_id' cannot be used together with 'source_path' or 'target_path'"
             )
 
-    def init_config(self, data: Dict[str, Any], user: User, project_id: int, work_dir: Path):
-        if self.storage_id:
+
+class RCloneStorage:
+    def __init__(
+        self, source_path: str, configuration: Dict[str, Any], readonly: bool, mount_folder: Path
+    ) -> None:
+        config.storage_validator.validate_storage_configuration(configuration)
+        self.configuration = configuration
+        self.source_path = source_path
+        self.mount_folder = mount_folder
+        self.readonly = readonly
+
+    @classmethod
+    def storage_from_schema(cls, data: Dict[str, Any], user: User, project_id: int, work_dir: str):
+        if data.get("storage_id"):
             # Load from storage service
             if user.access_token is None:
                 raise ValidationError("Storage mounting is only supported for logged-in users.")
             if project_id < 1:
                 raise ValidationError("Could not get gitlab project id")
             (
                 configuration,
-                self.source_path,
-                self.target_path,
+                source_path,
+                target_path,
                 readonly,
-            ) = config.storage_validator.get_storage_by_id(user, project_id, self.storage_id)
-            self.configuration = {**configuration, **(self.configuration or {})}
-            self.readonly = self.readonly
+            ) = config.storage_validator.get_storage_by_id(user, project_id, data["storage_id"])
+            configuration = {**configuration, **(configuration or {})}
+            readonly = readonly
         else:
-            self.source_path = data["source_path"]
-            self.target_path = data["target_path"]
-            self.configuration = data["configuration"]
-            self.readonly = self.readonly
-        config.storage_validator.validate_storage_configuration(self.configuration)
-        self._mount_folder = work_dir / self.target_path
+            source_path = data["source_path"]
+            target_path = data["target_path"]
+            configuration = data["configuration"]
+            readonly = data.get("readonly", True)
+        mount_folder = str(work_dir / target_path)
 
-    @property
-    def mount_folder(self):
-        if not self._mount_folder:
-            raise ProgrammingError("mount_folder not set. Ensure init_config was called first.")
-        return self._mount_folder
+        return cls(source_path, configuration, readonly, mount_folder)
 
     def get_manifest_patch(
         self, base_name: str, namespace: str, labels={}, annotations={}