feat(local files): Added local folder files

.mypy.ini

@@ -1,3 +1,6 @@
 [mypy]
-files = your_package, tests
-ignore_missing_imports = True
+mypy_path = src
+ignore_missing_imports = True
+strict = True
+namespace_packages = True
+explicit_package_bases = True

docs/README.md

@@ -2,7 +2,7 @@
 
 The purpose of this package is to make a simple way to interface with both cloud, local and in-memory file storage to support a development and production workflow for web applications needing file storage. The idea is to have a common inteface where the backend can be changed without making code changes by setting env variables. In this way local development and testing can use in-memory or local file storage, eventually run tests with a cloud file storage provider and use cloud file storage in production.
 
-Simply use the factory method to get a file manager instance. You can control which file manager will be used by setting the env variable NUMEROUS-FILES-BACKEND to either IN-MEMORY to use the memory based file system or AWS_S3 to use an S3 bucket on AWS.
+Simply use the factory method to get a file manager instance. You can control which file manager will be used by setting the env variable NUMEROUS_FILES_BACKEND to either IN-MEMORY to use the memory based file system, LOCAL to use a local folder, or AWS_S3 to use an S3 bucket on AWS.
 
 ## Installation
 

docs/local_file_manager.md

@@ -0,0 +1,3 @@
+### Local File Manager
+
+::: numerous.files.local.FileManager

mkdocs.yml

@@ -7,6 +7,7 @@ nav:
     - The File Manager Interface: file_manager.md
     - AWS S3 File Manager: aws_s3_file_manager.md
     - Memory File Manager: memory_file_manager.md
+    - Local File Manager: local_file_manager.md
 
 theme: readthedocs
 plugins:

src/numerous/files/__init__.py

@@ -4,10 +4,11 @@
 import os
 
 from numerous.files.aws_s3 import FileManager as AwsS3
+from numerous.files.local import FileManager as LocalFolder
 from numerous.files.memory import FileManager as Memory
 
 
-def file_manager_factory(**kwargs:dict[str,str]) -> AwsS3|Memory:
+def file_manager_factory(**kwargs:dict[str,str]) -> AwsS3|Memory|LocalFolder:
     """
     Use the factory for creating a file manager.
 
@@ -66,5 +67,10 @@ def file_manager_factory(**kwargs:dict[str,str]) -> AwsS3|Memory:
         return AwsS3(bucket=env_params["bucket"],
                      base_prefix=env_params["base_prefix"], credentials=credentials)
 
+    if file_manager == "LOCAL":
+        # Get a temporary folder from the operating system.
+
+        return LocalFolder(workfolder=str(kwargs.get("workfolder", "./tmp")))
+
     err_str = f"Unknown file manager type {file_manager}"
     raise ValueError(err_str)

src/numerous/files/aws_s3.py

@@ -2,7 +2,6 @@
 from typing import Any, Dict, List
 
 import boto3
-
 from numerous.files.file_manager import FileManager as FileManagerInterface
 from numerous.files.file_manager import StrOrPath
 

src/numerous/files/local.py

@@ -0,0 +1,119 @@
+"""Local Folder File manager."""
+import shutil
+from pathlib import Path
+
+from numerous.files.file_manager import FileManager as FileManagerInterface
+from numerous.files.file_manager import StrOrPath
+
+
+class FileManager(FileManagerInterface):
+
+    """
+    Local Folder File manager.
+
+    This class provides an local folder implementation of the FileManagerInterface.
+
+    Use this class for testing or when you want to store files in a local folder only.
+    """
+
+    def __init__(self, workfolder: StrOrPath="./tmp") -> None:
+        """
+        Initialize the LocalFolderFileManager.
+
+        Args:
+            workfolder: Path to the working folder.
+
+        """
+        self._workfolder = Path(workfolder)
+
+        # Ensure the workfolder exists.
+        self._workfolder.mkdir(parents=True, exist_ok=True)
+
+    def put(self, src: StrOrPath, dst: StrOrPath ) -> None:
+        """
+        Upload a file to a path.
+
+        Args:
+            src: Source path.
+            dst: Destination path.
+
+        """
+        _path = self._workfolder / Path(dst)
+        # Ensure the parent folder exists.
+        _path.parent.mkdir(parents=True, exist_ok=True)
+
+        with Path.open(Path(src), "rb") as f:
+            shutil.copyfileobj(f, Path.open(_path, "wb"))
+
+    def remove(self, path: StrOrPath) -> None:
+        """
+        Remove a file at a path.
+
+        Args:
+            path: Path to file.
+
+        """
+        _path = self._workfolder / Path(path)
+        Path(_path).unlink()
+
+    def list(self, path: StrOrPath|None) -> list[str]:
+        """
+        List files at a path.
+
+        Args:
+            path: Path to list files at.
+
+        """
+        _path = self._workfolder if path is None else self._workfolder / Path(path)
+
+        # Make all paths relative to the workfolder.
+        return [str(p.relative_to(self._workfolder))
+                for p in _path.rglob("*") if p.is_file()]
+
+    def move(self, src: StrOrPath, dst: StrOrPath) -> None:
+        """
+        Move a file from a source to a destination.
+
+        Args:
+            src: Source path.
+            dst: Destination path.
+
+        """
+        _src = self._workfolder / Path(src)
+        _dst = self._workfolder / Path(dst)
+        # Ensure the parent folder exists.
+        _dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.move(_src, _dst)
+
+
+    def copy(self, src: StrOrPath, dst: StrOrPath) -> None:
+        """
+        Copy a file from a source to a destination.
+
+        Args:
+            src: Source path.
+            dst: Destination path.
+
+        """
+        _src = self._workfolder / Path(src)
+        _dst = self._workfolder / Path(dst)
+        # Ensure the parent folder exists.
+        _dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy(_src, _dst)
+
+    def get(self, src: StrOrPath, dest: StrOrPath) -> None:
+        """
+        Download a file from a source to a destination.
+
+        Args:
+            src: Source path.
+            dest: Destination path.
+
+        """
+        _src = self._workfolder / Path(src)
+        # Ensure the parent folder exists.
+        Path(dest).parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy(_src, dest)
+
+
+

tests/test_local_filemanager.py

@@ -0,0 +1,96 @@
+from pathlib import Path
+from typing import Generator
+
+import pytest
+from numerous.files.local import FileManager
+
+
+@pytest.fixture()
+def test_file_create() -> Generator[Path, None, None]:
+
+    path_to_file = Path("test.txt")
+
+    with Path.open(path_to_file, "w") as f:
+        f.write("Hello World!")
+    yield path_to_file
+    Path.unlink(path_to_file)
+
+
+def test_file_manager_create() -> None:
+
+    file_manager = FileManager() # noqa: F841
+
+def test_file_put(test_file_create: str) -> None:
+
+    file_manager = FileManager()
+
+    file_manager.put(test_file_create, "tests/test_memory_filemanager.py")
+
+def test_file_remove(test_file_create: str) -> None:
+
+    file_manager = FileManager()
+
+    upload_path = "tests/test_memory_filemanager.py"
+
+    file_manager.put(test_file_create, upload_path)
+    file_manager.remove(upload_path)
+
+def test_file_list(test_file_create: str) -> None:
+
+    file_manager = FileManager()
+
+    upload_path = "tests/test_memory_filemanager.py"
+
+    file_manager.put(test_file_create, upload_path)
+    results = file_manager.list("tests/")
+    # Replace \\ with / for Windows compatibility
+    results = [r.replace("\\", "/") for r in results]
+    assert upload_path in results
+
+def test_file_move(test_file_create: str) -> None:
+
+    file_manager = FileManager()
+
+    upload_path = "tests/test_memory_filemanager.py"
+
+    file_manager.put(test_file_create, upload_path)
+    file_manager.move(upload_path, "tests/test_memory_filemanager2.py")
+
+    results = file_manager.list("tests/")
+    # Replace \\ with / for Windows compatibility
+    results = [r.replace("\\", "/") for r in results]
+
+    assert "tests/test_memory_filemanager2.py" in results
+    assert upload_path not in results
+
+def test_file_copy(test_file_create: str) -> None:
+
+    file_manager = FileManager()
+
+    upload_path = "tests/test_memory_filemanager.py"
+
+    file_manager.put(test_file_create, upload_path)
+    file_manager.copy(upload_path, "tests/test_memory_filemanager2.py")
+
+    file_manager.list("tests/")
+
+    results = file_manager.list("tests/")
+    # Replace \\ with / for Windows compatibility
+    results = [r.replace("\\", "/") for r in results]
+
+    assert "tests/test_memory_filemanager2.py" in results
+    assert upload_path in results
+
+def test_file_get(test_file_create: str) -> None:
+
+    file_manager = FileManager()
+
+    upload_path = "tests/test_memory_filemanager.py"
+
+    file_manager.put(test_file_create, upload_path)
+    file_manager.get(upload_path, "test_memory_filemanager.py")
+
+    with Path.open(Path("test_memory_filemanager.py")) as f:
+        assert f.read() == "Hello World!"
+
+    Path.unlink(Path("test_memory_filemanager.py"))