Feature/storage

fastapi_startkit/src/fastapi_startkit/application.py

@@ -112,6 +112,12 @@ def use_fastapi(self, fastapi: "FastAPI"):
     def use_base_path(self, path: str):
         return self.base_path / path
 
+    def storage_path(self, path: str = "") -> str:
+        return str(self.base_path / "storage" / path)
+
+    def public_path(self, path: str = "") -> str:
+        return str(self.base_path / "public" / path)
+
     def get(self, path: str, **kwargs) -> Callable:
         return self.fastapi.get(path, **kwargs)
 

fastapi_startkit/src/fastapi_startkit/facades/__init__.py

@@ -11,7 +11,6 @@
 from .Config import Config
 from .Loader import Loader
 from .Notification import Notification
-from .Storage import Storage
 from .Dump import Dump
 from .Queue import Queue
 from .Cache import Cache

fastapi_startkit/src/fastapi_startkit/helpers/app.py

@@ -0,0 +1,9 @@
+def storage_path(path: str = "") -> str:
+    """Get the path to the storage directory."""
+    from fastapi_startkit.application import app
+    return app().storage_path(path)
+
+def public_path(path: str = "") -> str:
+    """Get the path to the public directory."""
+    from fastapi_startkit.application import app
+    return app().public_path(path)

fastapi_startkit/src/fastapi_startkit/storage/__init__.py

@@ -0,0 +1 @@
+from .storage import Storage

fastapi_startkit/src/fastapi_startkit/storage/config.py

@@ -0,0 +1,41 @@
+import dataclasses
+from fastapi_startkit.environment import env
+from fastapi_startkit.helpers.app import storage_path
+
+@dataclasses.dataclass
+class StorageConfig:
+    default: str = dataclasses.field(
+        default_factory=lambda: env("FILESYSTEM_DISK", "local")
+    )
+
+    disks: dict = dataclasses.field(
+        default_factory=lambda: {
+            "local": {
+                "driver": "local",
+                "root": storage_path("app/private"),
+                "serve": True,
+                "throw": False,
+                "report": False,
+            },
+            "public": {
+                "driver": "local",
+                "root": storage_path("app/public"),
+                "url": env("APP_URL", "http://localhost").rstrip("/") + "/storage",
+                "visibility": "public",
+                "throw": False,
+                "report": False,
+            },
+            "s3": {
+                "driver": "s3",
+                "key": env("AWS_ACCESS_KEY_ID"),
+                "secret": env("AWS_SECRET_ACCESS_KEY"),
+                "region": env("AWS_DEFAULT_REGION", "us-east-1"),
+                "bucket": env("AWS_BUCKET"),
+                "url": env("AWS_URL"),
+                "endpoint": env("AWS_ENDPOINT"),
+                "use_path_style_endpoint": env("AWS_USE_PATH_STYLE_ENDPOINT", False),
+                "throw": False,
+                "report": False,
+            },
+        }
+    )

fastapi_startkit/src/fastapi_startkit/storage/drivers/__init__.py

@@ -0,0 +1,2 @@
+from .local import LocalDriver
+from .s3 import S3Driver

fastapi_startkit/src/fastapi_startkit/storage/drivers/local.py

@@ -0,0 +1,131 @@
+import os
+import uuid
+from os.path import isfile, join
+from shutil import copyfile, move
+
+from ..FileStream import FileStream
+from ..File import File
+from ...utils.filesystem import get_extension
+
+
+class LocalDriver:
+    def __init__(self, application):
+        self.application = application
+        self.options = {}
+
+    def set_options(self, options):
+        self.options = options
+        return self
+
+    def get_path(self, path):
+        root = self.options.get("root") or self.options.get("path")
+        file_path = os.path.join(root, path)
+        self.make_file_path_if_not_exists(file_path)
+        return file_path
+
+    def get_name(self, path, alias):
+        extension = get_extension(path)
+        return f"{alias}{extension}"
+
+    def put(self, file_path, content):
+        if isinstance(content, (bytes, bytearray)):
+            write_mode = "wb"
+        else:
+            write_mode = "w"
+        with open(self.get_path(os.path.join(file_path)), write_mode) as f:
+            f.write(content)
+        return content
+
+    def put_file(self, file_path, content, name=None):
+        file_name = self.get_name(content.name, name or str(uuid.uuid4()))
+
+        if hasattr(content, "get_content"):
+            content = content.get_content()
+
+        if isinstance(content, str):
+            content = bytes(content, "utf-8")
+
+        with open(self.get_path(os.path.join(file_path, file_name)), "wb") as f:
+            f.write(content)
+
+        return os.path.join(file_path, file_name)
+
+    def get(self, file_path):
+        try:
+            with open(self.get_path(file_path), "r") as f:
+                content = f.read()
+
+            return content
+        except FileNotFoundError:
+            return None
+
+    def exists(self, file_path):
+        return os.path.exists(self.get_path(file_path))
+
+    def missing(self, file_path):
+        return not self.exists(file_path)
+
+    def stream(self, file_path):
+        with open(self.get_path(file_path), "r") as f:
+            content = f
+        return FileStream(content)
+
+    def copy(self, from_file_path, to_file_path):
+        return copyfile(from_file_path, to_file_path)
+
+    def move(self, from_file_path, to_file_path):
+        return move(self.get_path(from_file_path), self.get_path(to_file_path))
+
+    def prepend(self, file_path, content):
+        value = self.get(file_path)
+        content = content + value
+        self.put(file_path, content)
+        return content
+
+    def append(self, file_path, content):
+        with open(self.get_path(file_path), "a") as f:
+            f.write(content)
+        return content
+
+    def delete(self, file_path):
+        return os.remove(self.get_path(file_path))
+
+    def make_directory(self, directory):
+        pass
+
+    def store(self, file, name=None):
+        if name:
+            name = f"{name}{file.extension()}"
+        full_path = self.get_path(name or file.hash_path_name())
+        with open(full_path, "wb") as f:
+            f.write(file.stream())
+
+        return full_path
+
+    def make_file_path_if_not_exists(self, file_path):
+        if not os.path.isfile(file_path):
+            if not os.path.exists(os.path.dirname(file_path)):
+                # Create the path to the model if it does not exist
+                os.makedirs(os.path.dirname(file_path))
+
+            return True
+
+        return False
+
+    def get_files(self, directory=""):
+        file_path = self.get_path(directory)
+        files = []
+        for f in os.listdir(file_path):
+            if not isfile(join(file_path, f)):
+                continue
+
+            files.append(File(self.get(f), f))
+
+        return files
+
+    def download(self, file_path, name=None, force=False):
+        from fastapi.responses import FileResponse
+        return FileResponse(
+            self.get_path(file_path),
+            filename=name or os.path.basename(file_path)
+        )