Register local files (#816)

* Create storage.md

* add stubs for uploading local files

* file remove and download endpoints

* fix pydantic model

* give endpoint a distinct name

* add important parentheses

* run codegen

* hide share button if not Minio storage

* Hide version history tab

* hide version chip if local, include in details

* Update DatasetsService.ts

* ignore version when downloading metadata for local files

* add additional check for v0

* remove logs

* merge and code updates

* codegen

* codegen

---------

Co-authored-by: Chen Wang <[email protected]>

Author: max-zilla

backend/app/config.py

@@ -36,7 +36,10 @@ class Settings(BaseSettings):
     MINIO_EXPIRES: int = 3600  # seconds
     MINIO_SECURE: str = "False"  # http vs https
 
-    # keycloak server
+    # Files in the listed directories can be added to Clowder without copying them elsewhere
+    LOCAL_WHITELIST: List[str] = []
+
+    # Keycloak server
     auth_base = "http://localhost:8080"
     auth_realm = "clowder"
     auth_client_id = "clowder2-backend"

backend/app/models/files.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from enum import Enum
 from typing import Optional, List
 
 from beanie import Document, View, PydanticObjectId
@@ -9,6 +10,16 @@
 from app.models.users import UserOut
 
 
+class StorageType(str, Enum):
+    """Depending on the StorageType,the file may need different properties such as local path or URL.
+    Also, some StorageTypes do not support versioning or anonymous sharing."""
+
+    MINIO = "minio"
+    LOCAL = "local"
+    REMOTE = "remote"
+    AWS = "aws"
+
+
 class ContentType(BaseModel):
     """This model describes the content type of any type of file(File or Visualization data) uploaded to Clowder. A typical example is "text/plain" for .txt.
     In Clowder v1 extractors, "text/*" syntax is acceptable for wildcard matches. To support this, the content type is
@@ -42,6 +53,12 @@ class FileIn(FileBase):
     pass
 
 
+class LocalFileIn(BaseModel):
+    """Used when adding a file from a local disk."""
+
+    path: str
+
+
 class FileDB(Document, FileBase):
     creator: UserOut
     created: datetime = Field(default_factory=datetime.utcnow)
@@ -54,10 +71,16 @@ class FileDB(Document, FileBase):
     bytes: int = 0
     content_type: ContentType = ContentType()
     thumbnail_id: Optional[PydanticObjectId] = None
+    storage_type: StorageType = StorageType.MINIO
+    storage_path: Optional[str]  # store URL or file path depending on storage_type
 
     class Settings:
         name = "files"
 
+    class Config:
+        # required for Enum to properly work
+        use_enum_values = True
+
 
 class FileDBViewList(View, FileBase):
     id: PydanticObjectId = Field(None, alias="_id")  # necessary for Views

backend/app/routers/datasets.py

@@ -47,15 +47,15 @@
     DatasetDBViewList,
     DatasetStatus,
 )
-from app.models.files import FileOut, FileDB, FileDBViewList
+from app.models.files import FileOut, FileDB, FileDBViewList, LocalFileIn, StorageType
 from app.models.folders import FolderOut, FolderIn, FolderDB, FolderDBViewList
 from app.models.metadata import MetadataDB
 from app.models.pyobjectid import PyObjectId
 from app.models.thumbnails import ThumbnailDB
 from app.models.users import UserOut
 from app.rabbitmq.listeners import submit_dataset_job
 from app.routers.authentication import get_admin
-from app.routers.files import add_file_entry, remove_file_entry
+from app.routers.files import add_file_entry, add_local_file_entry, remove_file_entry
 from app.search.connect import (
     delete_document_by_id,
 )
@@ -446,6 +446,10 @@ async def _delete_nested_folders(parent_folder_id):
     raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")
 
 
+# new endpoint for /{dataset_id}/local_files
+# new endpoint for /{dataset_id}/remote_files
+
+
 @router.post("/{dataset_id}/files", response_model=FileOut)
 async def save_file(
     dataset_id: str,
@@ -534,6 +538,61 @@ async def save_files(
         raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")
 
 
+@router.post("/{dataset_id}/local_files", response_model=FileOut)
+async def save_local_file(
+    localfile_in: LocalFileIn,
+    dataset_id: str,
+    folder_id: Optional[str] = None,
+    user=Depends(get_current_user),
+    es=Depends(dependencies.get_elasticsearchclient),
+    rabbitmq_client: BlockingChannel = Depends(dependencies.get_rabbitmq),
+    allow: bool = Depends(Authorization("uploader")),
+):
+    if (dataset := await DatasetDB.get(PydanticObjectId(dataset_id))) is not None:
+        if user is None:
+            raise HTTPException(
+                status_code=401, detail=f"User not found. Session might have expired."
+            )
+
+        # Check to make sure file is cleared to proceed
+        cleared = False
+        for wpath in settings.LOCAL_WHITELIST:
+            if localfile_in.path.startswith(wpath):
+                cleared = True
+        if not cleared:
+            raise HTTPException(
+                status_code=500,
+                detail=f"File is not located in a whitelisted directory.",
+            )
+
+        (dirname, filename) = os.path.split(localfile_in.path)
+        new_file = FileDB(
+            name=filename,
+            creator=user,
+            dataset_id=dataset.id,
+            storage_type=StorageType.LOCAL,
+            storage_path=localfile_in.path,
+            bytes=os.path.getsize(localfile_in.path),
+        )
+
+        if folder_id is not None:
+            if (folder := await FolderDB.get(PydanticObjectId(folder_id))) is not None:
+                new_file.folder_id = folder.id
+            else:
+                raise HTTPException(
+                    status_code=404, detail=f"Folder {folder_id} not found"
+                )
+
+        await add_local_file_entry(
+            new_file,
+            user,
+            es,
+            rabbitmq_client,
+        )
+        return new_file.dict()
+    raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")
+
+
 @router.post("/createFromZip", response_model=DatasetOut)
 async def create_dataset_from_zip(
     user=Depends(get_current_user),

backend/app/routers/files.py

@@ -16,7 +16,7 @@
     File,
     UploadFile,
 )
-from fastapi.responses import StreamingResponse
+from fastapi.responses import StreamingResponse, FileResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from minio import Minio
 from pika.adapters.blocking_connection import BlockingChannel
@@ -25,15 +25,10 @@
 from app.config import settings
 from app.deps.authorization_deps import FileAuthorization
 from app.keycloak_auth import get_current_user, get_token
-from app.models.files import (
-    FileOut,
-    FileVersion,
-    FileDB,
-    FileVersionDB,
-)
+from app.models.files import FileOut, FileVersion, FileDB, FileVersionDB, StorageType
 from app.models.metadata import MetadataDB
-from app.models.users import UserOut
 from app.models.thumbnails import ThumbnailDB
+from app.models.users import UserOut
 from app.rabbitmq.listeners import submit_file_job, EventListenerJobDB
 from app.routers.feeds import check_feed_listeners
 from app.routers.utils import get_content_type
@@ -155,17 +150,41 @@ async def add_file_entry(
     )
 
 
+async def add_local_file_entry(
+    new_file: FileDB,
+    user: UserOut,
+    es: Elasticsearch,
+    rabbitmq_client: BlockingChannel,
+    content_type: Optional[str] = None,
+):
+    """Insert FileDB object into MongoDB (makes Clowder ID). Bytes are not stored in DB and versioning not supported
+    for local files."""
+
+    content_type_obj = get_content_type(new_file.name, content_type)
+    new_file.content_type = content_type_obj
+    await new_file.insert()
+
+    # Add entry to the file index
+    await index_file(es, FileOut(**new_file.dict()))
+
+    # TODO - timing issue here, check_feed_listeners needs to happen asynchronously.
+    time.sleep(1)
+
+    # Submit file job to any qualifying feeds
+    await check_feed_listeners(
+        es,
+        FileOut(**new_file.dict()),
+        user,
+        rabbitmq_client,
+    )
+
+
 # TODO: Move this to MongoDB middle layer
 async def remove_file_entry(
     file_id: Union[str, ObjectId], fs: Minio, es: Elasticsearch
 ):
     """Remove FileDB object into MongoDB, Minio, and associated metadata and version information."""
     # TODO: Deleting individual versions will require updating version_id in mongo, or deleting entire document
-
-    # Check all connection and abort if any one of them is not available
-    if fs is None or es is None:
-        raise HTTPException(status_code=503, detail="Service not available")
-        return
     fs.remove_object(settings.MINIO_BUCKET_NAME, str(file_id))
     # delete from elasticsearch
     delete_document_by_id(es, settings.elasticsearch_index, str(file_id))
@@ -175,6 +194,18 @@ async def remove_file_entry(
     await FileVersionDB.find(FileVersionDB.file_id == ObjectId(file_id)).delete()
 
 
+async def remove_local_file_entry(file_id: Union[str, ObjectId], es: Elasticsearch):
+    """Remove FileDB object into MongoDB, Minio, and associated metadata and version information."""
+    # TODO: Deleting individual versions will require updating version_id in mongo, or deleting entire document
+    # delete from elasticsearch
+    delete_document_by_id(es, settings.elasticsearch_index, str(file_id))
+    if (file := await FileDB.get(PydanticObjectId(file_id))) is not None:
+        # TODO: delete from disk - should this be allowed if Clowder didn't originally write the file?
+        # os.path.remove(file.storage_path)
+        await file.delete()
+    await MetadataDB.find(MetadataDB.resource.resource_id == ObjectId(file_id)).delete()
+
+
 @router.put("/{file_id}", response_model=FileOut)
 async def update_file(
     file_id: str,
@@ -275,33 +306,53 @@ async def download_file(
 ):
     # If file exists in MongoDB, download from Minio
     if (file := await FileDB.get(PydanticObjectId(file_id))) is not None:
-        if version is not None:
-            # Version is specified, so get the minio ID from versions table if possible
-            file_vers = await FileVersionDB.find_one(
-                FileVersionDB.file_id == ObjectId(file_id),
-                FileVersionDB.version_num == version,
-            )
-            if file_vers is not None:
-                vers = FileVersion(**file_vers.dict())
-                content = fs.get_object(
-                    settings.MINIO_BUCKET_NAME, file_id, version_id=vers.version_id
+        if file.storage_type == StorageType.MINIO:
+            if version is not None:
+                # Version is specified, so get the minio ID from versions table if possible
+                file_vers = await FileVersionDB.find_one(
+                    FileVersionDB.file_id == ObjectId(file_id),
+                    FileVersionDB.version_num == version,
                 )
+                if file_vers is not None:
+                    vers = FileVersion(**file_vers.dict())
+                    content = fs.get_object(
+                        settings.MINIO_BUCKET_NAME, file_id, version_id=vers.version_id
+                    )
+                else:
+                    raise HTTPException(
+                        status_code=404,
+                        detail=f"File {file_id} version {version} not found",
+                    )
             else:
-                raise HTTPException(
-                    status_code=404,
-                    detail=f"File {file_id} version {version} not found",
-                )
+                # If no version specified, get latest version directly
+                content = fs.get_object(settings.MINIO_BUCKET_NAME, file_id)
+
+            # Get content type & open file stream
+            response = StreamingResponse(
+                content.stream(settings.MINIO_UPLOAD_CHUNK_SIZE)
+            )
+            response.headers["Content-Disposition"] = (
+                "attachment; filename=%s" % file.name
+            )
+
+        elif file.storage_type == StorageType.LOCAL:
+            response = FileResponse(
+                path=file.storage_path,
+                filename=file.name,
+                media_type=file.content_type.content_type,
+            )
+
         else:
-            # If no version specified, get latest version directly
-            content = fs.get_object(settings.MINIO_BUCKET_NAME, file_id)
+            raise HTTPException(
+                status_code=400, detail=f"Unable to download {file_id}."
+            )
+
+        if response:
+            if increment:
+                # Increment download count
+                await file.update(Inc({FileDB.downloads: 1}))
+            return response
 
-        # Get content type & open file stream
-        response = StreamingResponse(content.stream(settings.MINIO_UPLOAD_CHUNK_SIZE))
-        response.headers["Content-Disposition"] = "attachment; filename=%s" % file.name
-        if increment:
-            # Increment download count
-            await file.update(Inc({FileDB.downloads: 1}))
-        return response
     else:
         raise HTTPException(status_code=404, detail=f"File {file_id} not found")
 
@@ -365,8 +416,11 @@ async def delete_file(
     es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
     allow: bool = Depends(FileAuthorization("editor")),
 ):
-    if (await FileDB.get(PydanticObjectId(file_id))) is not None:
-        await remove_file_entry(file_id, fs, es)
+    if (file := await FileDB.get(PydanticObjectId(file_id))) is not None:
+        if file.storage_type == StorageType.LOCAL:
+            await remove_local_file_entry(file_id, es)
+        else:
+            await remove_file_entry(file_id, fs, es)
         return {"deleted": file_id}
     else:
         raise HTTPException(status_code=404, detail=f"File {file_id} not found")
@@ -415,13 +469,13 @@ async def get_file_versions(
     limit: int = 20,
     allow: bool = Depends(FileAuthorization("viewer")),
 ):
-    file = await FileDB.get(PydanticObjectId(file_id))
-    if file is not None:
+    if (file := await FileDB.get(PydanticObjectId(file_id))) is not None:
         mongo_versions = []
-        async for ver in FileVersionDB.find(
-            FileVersionDB.file_id == ObjectId(file_id)
-        ).sort(-FileVersionDB.created).skip(skip).limit(limit):
-            mongo_versions.append(FileVersion(**ver.dict()))
+        if file.storage_type == StorageType.MINIO:
+            async for ver in FileVersionDB.find(
+                FileVersionDB.file_id == ObjectId(file_id)
+            ).sort(-FileVersionDB.created).skip(skip).limit(limit):
+                mongo_versions.append(FileVersion(**ver.dict()))
         return mongo_versions
 
     raise HTTPException(status_code=404, detail=f"File {file_id} not found")

backend/app/routers/metadata_files.py

@@ -54,7 +54,7 @@ async def _build_metadata_db_obj(
     if version is None:
         # Validate specified version, or use latest by default
         file_version = metadata_in.file_version
-        if file_version is not None:
+        if file_version is not None and file_version > 0:
             if (
                 await FileVersionDB.find_one(
                     FileVersionDB.file_id == file.id,
@@ -117,10 +117,10 @@ async def add_file_metadata(
     """
     if (file := await FileDB.get(PydanticObjectId(file_id))) is not None:
         current_file_version = file.version_num
-        # if metadata does not already specify a file version
-        # change metadata_in file version to match the current file version
+        # if metadata does not already specify a version, change metadata_in version to match current file version
         if metadata_in.file_version is None:
             metadata_in.file_version = current_file_version
+
         # If dataset already has metadata using this definition, don't allow duplication
         definition = metadata_in.definition
         if definition is not None:
@@ -340,7 +340,7 @@ async def get_file_metadata(
 
         # Validate specified version, or use latest by default
         if not all_versions:
-            if version is not None:
+            if version is not None and version > 0:
                 if (
                     await FileVersionDB.find_one(
                         FileVersionDB.file_id == ObjectId(file_id),

backend/app/routers/utils.py

@@ -8,7 +8,7 @@ def get_content_type(
     filename: str,
     content_type: Optional[str] = None,
 ):
-    """Returns ContentType object given a content_tyoe, also guessed the content_type if none is provided
+    """Returns ContentType object given a content_type, also guessed the content_type if none is provided
 
     Arguments:
         content_type: content_type of a file to be uploaded