Files list end point refactor (#1529)

* First refactorization for files

* Completed use cases for list end-point

* Renamed storage to file_storage

* Migrated files to file_storage

* Migrated get files to the new service

* Check if function exists first

* Separated end-points to the new provider

* Improved tests

* Restore original fixtures

* Solved the problem with the external if

* Added Literal type in the FileStorage service

* functions methods refactorization

* use self.username instead of the variable

* run path sanitization only one time

* function title is mandatory for file storage

* fix black

* Updated 403 by 404

* Updated swagger for list

* Makes use of enum with integers

* use is instead of equals

* unify path build

* get_function refactor

gateway/api/services/file_storage.py

@@ -0,0 +1,124 @@
+"""
+This file stores the logic to manage the access to data stores
+"""
+import glob
+import logging
+import os
+from enum import Enum
+
+from django.conf import settings
+
+from utils import sanitize_file_path
+
+
+class WorkingDir(Enum):
+    """
+    This Enum has the values:
+        USER_STORAGE
+        PROVIDER_STORAGE
+
+    Both values are being used to identify in
+        FileStorage service the path to be used
+    """
+
+    USER_STORAGE = 1
+    PROVIDER_STORAGE = 2
+
+
+SUPPORTED_FILE_EXTENSIONS = [".tar", ".h5"]
+
+logger = logging.getLogger("gateway")
+
+
+class FileStorage:  # pylint: disable=too-few-public-methods
+    """
+    The main objective of this class is to manage the access of the users to their storage.
+
+    Attributes:
+        username (str): storgae user's username
+        working_dir (WorkingDir(Enum)): working directory
+        function_title (str): title of the function in case is needed to build the path
+        provider_name (str | None): name of the provider in caseis needed to build the path
+    """
+
+    def __init__(
+        self,
+        username: str,
+        working_dir: WorkingDir,
+        function_title: str,
+        provider_name: str | None,
+    ) -> None:
+        self.file_path = None
+        self.username = username
+
+        if working_dir is WorkingDir.USER_STORAGE:
+            self.file_path = self.__get_user_path(function_title, provider_name)
+        elif working_dir is WorkingDir.PROVIDER_STORAGE:
+            self.file_path = self.__get_provider_path(function_title, provider_name)
+
+    def __get_user_path(self, function_title: str, provider_name: str | None) -> str:
+        """
+        This method returns the path where the user will store its files
+
+        Args:
+            function_title (str): in case the function is from a
+                provider it will identify the function folder
+            provider_name (str | None): in case a provider is provided it will
+                identify the folder for the specific function
+
+        Returns:
+            str: storage path.
+                - In case the function is from a provider that path would
+                    be: username/provider_name/function_title
+                - In case the function is from a user that path would
+                    be: username/
+        """
+        if provider_name is None:
+            path = os.path.join(settings.MEDIA_ROOT, self.username)
+        else:
+            path = os.path.join(
+                settings.MEDIA_ROOT, self.username, provider_name, function_title
+            )
+
+        return sanitize_file_path(path)
+
+    def __get_provider_path(self, function_title: str, provider_name: str) -> str:
+        """
+        This method returns the provider path where the user will store its files
+
+        Args:
+            function_title (str): in case the function is from a provider
+                it will identify the function folder
+            provider_name (str): in case a provider is provided
+                it will identify the folder for the specific function
+
+        Returns:
+            str: storage path following the format provider_name/function_title/
+        """
+        path = os.path.join(settings.MEDIA_ROOT, provider_name, function_title)
+
+        return sanitize_file_path(path)
+
+    def get_files(self) -> list[str]:
+        """
+        This method returns a list of file names following the next rules:
+            - Only files with supported extensions are listed
+            - It returns only files from a user or a provider file storage
+
+        Returns:
+            list[str]: list of file names
+        """
+
+        if not os.path.exists(self.file_path):
+            logger.warning(
+                "Directory %s does not exist for %s.",
+                self.file_path,
+                self.username,
+            )
+            return []
+
+        return [
+            os.path.basename(path)
+            for extension in SUPPORTED_FILE_EXTENSIONS
+            for path in glob.glob(f"{self.file_path}/*{extension}")
+        ]

gateway/api/utils.py

@@ -419,7 +419,7 @@ def create_dependency_allowlist():
     return allowlist
 
 
-def sanitize_name(name: str):
+def sanitize_name(name: str | None):
     """Sanitize name"""
     if name:
         sanitized_name = ""

gateway/api/v1/views/files.py

@@ -19,7 +19,7 @@ class FilesViewSet(views.FilesViewSet):
     permission_classes = [permissions.IsAuthenticated, IsOwner]
 
     @swagger_auto_schema(
-        operation_description="List of available for user files",
+        operation_description="List of available files in the user directory",
         manual_parameters=[
             openapi.Parameter(
                 "provider",
@@ -28,11 +28,41 @@ class FilesViewSet(views.FilesViewSet):
                 type=openapi.TYPE_STRING,
                 required=False,
             ),
+            openapi.Parameter(
+                "function",
+                openapi.IN_QUERY,
+                description="function title",
+                type=openapi.TYPE_STRING,
+                required=True,
+            ),
         ],
     )
     def list(self, request):
         return super().list(request)
 
+    @swagger_auto_schema(
+        operation_description="List of available files in the provider directory",
+        manual_parameters=[
+            openapi.Parameter(
+                "provider",
+                openapi.IN_QUERY,
+                description="provider name",
+                type=openapi.TYPE_STRING,
+                required=True,
+            ),
+            openapi.Parameter(
+                "function",
+                openapi.IN_QUERY,
+                description="function title",
+                type=openapi.TYPE_STRING,
+                required=True,
+            ),
+        ],
+    )
+    @action(methods=["GET"], detail=False, url_path="provider")
+    def provider_list(self, request):
+        return super().provider_list(request)
+
     @swagger_auto_schema(
         operation_description="Download a specific file",
         manual_parameters=[