storage provider added + full tests coverage

Author: Tomasz Szymański

storages/backends/amazon_s3.py

@@ -1,7 +1,6 @@
 import logging
 from datetime import datetime
-from os import environ
-from typing import AnyStr, Optional
+from typing import AnyStr
 
 import boto3
 from boto3 import Session
@@ -10,23 +9,30 @@
 from botocore.response import StreamingBody
 
 from storages.backends.base import Storage
+from storages.exceptions import ImproperlyConfiguredError
 
 
 class AmazonS3Storage(Storage):
     _SERVICE_NAME = "s3"
 
     def __init__(
         self,
-        aws_access_key_id: Optional[str],
-        aws_secret_access_key: Optional[str],
-        bucket_name: Optional[str],
+        aws_access_key_id: str,
+        aws_secret_access_key: str,
+        bucket_name: str,
     ):
-        if aws_access_key_id is None:
-            raise ValueError("aws_access_key_id can not be None.")
-        if aws_secret_access_key is None:
-            raise ValueError("aws_secret_access_key can not be None.")
-        if bucket_name is None:
-            raise ValueError("bucket_name can not be None.")
+        if not aws_access_key_id:
+            raise ImproperlyConfiguredError(
+                name="aws_access_key_id", value=aws_access_key_id
+            )
+        if not aws_secret_access_key:
+            raise ImproperlyConfiguredError(
+                name="aws_secret_access_key", value=aws_secret_access_key
+            )
+        if not bucket_name:
+            raise ImproperlyConfiguredError(
+                name="bucket_name", value=bucket_name
+            )
         self._session: Session = boto3.Session(
             aws_access_key_id=aws_access_key_id,
             aws_secret_access_key=aws_secret_access_key,
@@ -75,10 +81,3 @@ def get_access_time(self, name: str) -> datetime:
         raise NotImplementedError(
             "S3 storage does not provide access time info."
         )
-
-
-amazon_s3_storage = AmazonS3Storage(
-    aws_access_key_id=environ.get("AWS_ACCESS_KEY_ID"),
-    aws_secret_access_key=environ.get("AWS_SECRET_ACCESS_KEY"),
-    bucket_name=environ.get("AWS_S3_BUCKET_NAME"),
-)

storages/backends/base.py

@@ -38,3 +38,9 @@ def get_modified_time(self, name: str) -> datetime:
     @abstractmethod
     def get_access_time(self, name: str) -> datetime:
         pass  # pragma: no cover
+
+
+class StorageBuilder(ABC):
+    @abstractmethod
+    def build(self) -> Storage:
+        pass  # pragma: no cover

storages/backends/file_system.py

@@ -4,12 +4,13 @@
 from typing import AnyStr, Union
 
 from storages.backends.base import Storage
+from storages.exceptions import ImproperlyConfiguredError
 
 
 class FileSystemStorage(Storage):
     def __init__(self, path: str):
         if not path:
-            raise ValueError("path can not be empty")
+            raise ImproperlyConfiguredError(name="path", value=path)
         self._base_path = path
 
     def _path(self, name: str) -> str:

storages/exceptions.py

@@ -0,0 +1,17 @@
+from typing import Any
+
+
+class StoragesError(Exception):
+    pass
+
+
+class ImproperlyConfiguredError(StoragesError):
+    def __init__(self, name: str, value: Any):
+        super().__init__(
+            f"The '{name}' setting has improper value of: '{value}'"
+        )
+
+
+class MissingEnvironmentVariableError(StoragesError):
+    def __init__(self, name: str):
+        super().__init__(f"The environment variable '{name}' is not defined")

storages/provider.py

@@ -0,0 +1,81 @@
+import os
+from importlib import import_module
+from inspect import Signature
+from os import environ
+from typing import Any, Dict, Generator, Iterable, Tuple, Type
+
+from storages.backends.base import Storage
+from storages.exceptions import MissingEnvironmentVariableError
+
+
+class StorageConstructorArgumentsExtractor:
+    _DEFAULT_IGNORED_ARGUMENTS = ("self", "cls")
+
+    @staticmethod
+    def extract(
+        storage_backend_class: Type[Storage],
+        ignored_arguments: Tuple[str, ...] = _DEFAULT_IGNORED_ARGUMENTS,
+    ) -> Generator[str, None, None]:
+        signature = Signature.from_callable(storage_backend_class.__init__)
+        return (
+            parameter.name
+            for _, parameter in signature.parameters.items()
+            if parameter.kind is parameter.POSITIONAL_OR_KEYWORD
+            and parameter.name not in ignored_arguments
+        )
+
+
+class EnvironmentVariablesCollector:
+    _DEFAULT_PREFIX = "STORAGES_"
+
+    @classmethod
+    def collect(
+        cls, names: Iterable[str], prefix: str = _DEFAULT_PREFIX
+    ) -> Dict[str, Any]:
+        values = {}
+        for name in names:
+            environment_variable_name = f"{prefix}{name.upper()}"
+            value = os.environ.get(environment_variable_name)
+            if value is None:
+                raise MissingEnvironmentVariableError(
+                    name=environment_variable_name
+                )
+            values[name] = value
+        return values
+
+
+class DynamicStorageLoader:
+    _PATH_DELIMITER = "."
+
+    @classmethod
+    def load_class(cls, path: str) -> Type[Storage]:
+        module_name, class_name = cls._get_module_and_class_name_from_path(
+            path=path
+        )
+        module = import_module(module_name)
+        return getattr(module, class_name)
+
+    @classmethod
+    def _get_module_and_class_name_from_path(
+        cls, path: str
+    ) -> Tuple[str, str]:
+        parts = path.split(cls._PATH_DELIMITER)
+        return cls._PATH_DELIMITER.join(parts[:-1]), parts[-1]
+
+
+class StorageProvider:
+    @classmethod
+    def provide(cls, backend_path: str):
+        backend_class = DynamicStorageLoader.load_class(path=backend_path)
+        constructor_arguments = StorageConstructorArgumentsExtractor.extract(
+            storage_backend_class=backend_class
+        )
+        constructor_argument_values = EnvironmentVariablesCollector.collect(
+            names=constructor_arguments
+        )
+        return backend_class(**constructor_argument_values)  # type: ignore
+
+
+default_storage = StorageProvider.provide(
+    backend_path=environ["STORAGES_BACKEND"]
+)

tests/test_aws_s3_storage.py

@@ -1,11 +1,13 @@
-from datetime import datetime, timedelta, tzinfo, timezone
+from datetime import datetime, timedelta
+from os import environ
 from unittest import TestCase
 from uuid import uuid4
 
 import pytest
 
+from storages.backends.amazon_s3 import AmazonS3Storage
 from storages.backends.base import Storage
-from storages.backends.amazon_s3 import amazon_s3_storage, AmazonS3Storage
+from storages.exceptions import ImproperlyConfiguredError
 
 
 class aws_temp_file:
@@ -30,19 +32,33 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
 
 class TestAWSS3Storage(TestCase):
-    _DATE_TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
-
     @pytest.fixture(autouse=True)
     def init_storage(self, tmpdir):
-        self._storage = amazon_s3_storage
+        self._storage = AmazonS3Storage(
+            aws_access_key_id=environ.get("STORAGES_AWS_ACCESS_KEY_ID"),
+            aws_secret_access_key=environ.get(
+                "STORAGES_AWS_SECRET_ACCESS_KEY"
+            ),
+            bucket_name=environ.get("STORAGES_BUCKET_NAME"),
+        )
 
     def test_improper_initialization(self):
-        with pytest.raises(Exception):
-            AmazonS3Storage(aws_access_key_id=None, aws_secret_access_key=None, bucket_name=None)
-        with pytest.raises(Exception):
-            AmazonS3Storage(aws_access_key_id="somekey", aws_secret_access_key=None, bucket_name=None)
-        with pytest.raises(Exception):
-            AmazonS3Storage(aws_access_key_id="some_key", aws_secret_access_key="some_secret", bucket_name=None)
+        with pytest.raises(ImproperlyConfiguredError):
+            AmazonS3Storage(
+                aws_access_key_id="", aws_secret_access_key="", bucket_name=""
+            )
+        with pytest.raises(ImproperlyConfiguredError):
+            AmazonS3Storage(
+                aws_access_key_id="somekey",
+                aws_secret_access_key="",
+                bucket_name="",
+            )
+        with pytest.raises(ImproperlyConfiguredError):
+            AmazonS3Storage(
+                aws_access_key_id="some_key",
+                aws_secret_access_key="some_secret",
+                bucket_name="",
+            )
 
     def test_file_not_exists(self):
         assert not self._storage.exists("some_non_existent.file")
@@ -57,7 +73,10 @@ def test_file_contains_written_data(self):
 
     def test_file_contains_binary_written_data(self):
         with aws_temp_file(storage=self._storage, binary=True) as temp_file:
-            assert self._storage.read(temp_file, mode="rb") == aws_temp_file.CONTENT_BINARY
+            assert (
+                self._storage.read(temp_file, mode="rb")
+                == aws_temp_file.CONTENT_BINARY
+            )
 
     def test_file_does_not_exist_upon_writing_and_deletion(self):
         with aws_temp_file(storage=self._storage) as temp_file:
@@ -69,14 +88,16 @@ def test_file_size_matches_content_size(self):
             assert self._storage.size(temp_file) == len(aws_temp_file.CONTENT)
 
     def test_file_creation_time(self):
-        with pytest.raises(Exception):
+        with pytest.raises(NotImplementedError):
             self._storage.get_created_time("any_name.ext")
 
     def test_file_access_time(self):
-        with pytest.raises(Exception):
+        with pytest.raises(NotImplementedError):
             self._storage.get_access_time("any_name.ext")
 
     def test_file_modification_time(self):
         with aws_temp_file(storage=self._storage) as temp_file:
-            modification_time = self._storage.get_modified_time(temp_file).replace(tzinfo=None)
+            modification_time = self._storage.get_modified_time(
+                temp_file
+            ).replace(tzinfo=None)
             assert modification_time - datetime.now() < timedelta(seconds=10)

tests/test_file_system_storage.py

@@ -1,23 +1,23 @@
-import pytest
+from datetime import datetime, timedelta
 from unittest import TestCase
 
-from datetime import datetime
+import pytest
 
 from storages.backends.file_system import FileSystemStorage
+from storages.exceptions import ImproperlyConfiguredError
 
 
 class TestFileSystemStorage(TestCase):
     _TEST_FILE_NAME = "test_file.txt"
     _TEST_FILE_CONTENT = "Lorem ipsum dolor sit amet..."
     _TEST_FILE_CONTENT_BINARY = b"Binary lorem ipsum dolor sit amet..."
-    _DATE_TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
 
     @pytest.fixture(autouse=True)
     def init_storage(self, tmpdir):
         self._storage = FileSystemStorage(path=tmpdir)
 
     def test_improper_storage_initialization(self):
-        with pytest.raises(ValueError):
+        with pytest.raises(ImproperlyConfiguredError):
             FileSystemStorage(path="")
 
     def test_file_not_exists(self):
@@ -29,11 +29,16 @@ def test_file_exists_upon_writing(self):
 
     def test_file_contains_written_data(self):
         self._write_contents_to_file()
-        assert self._storage.read(self._TEST_FILE_NAME) == self._TEST_FILE_CONTENT
+        assert (
+            self._storage.read(self._TEST_FILE_NAME) == self._TEST_FILE_CONTENT
+        )
 
     def test_file_contains_binary_written_data(self):
         self._write_contents_to_file(binary=True)
-        assert self._storage.read(self._TEST_FILE_NAME, mode="rb") == self._TEST_FILE_CONTENT_BINARY
+        assert (
+            self._storage.read(self._TEST_FILE_NAME, mode="rb")
+            == self._TEST_FILE_CONTENT_BINARY
+        )
 
     def test_file_does_not_exist_upon_writing_and_deletion(self):
         self._write_contents_to_file()
@@ -42,29 +47,35 @@ def test_file_does_not_exist_upon_writing_and_deletion(self):
 
     def test_file_size_matches_content_size(self):
         self._write_contents_to_file()
-        assert self._storage.size(self._TEST_FILE_NAME) == len(self._TEST_FILE_CONTENT)
+        assert self._storage.size(self._TEST_FILE_NAME) == len(
+            self._TEST_FILE_CONTENT
+        )
 
     def test_file_creation_time(self):
         now = datetime.utcnow()
         self._write_contents_to_file()
         creation_time = self._storage.get_created_time(self._TEST_FILE_NAME)
-        assert creation_time.strftime(self._DATE_TIME_FORMAT) == now.strftime(self._DATE_TIME_FORMAT)
+        assert creation_time - now < timedelta(seconds=1)
 
     def test_file_modification_time(self):
         now = datetime.utcnow()
         self._write_contents_to_file()
-        modification_time = self._storage.get_modified_time(self._TEST_FILE_NAME)
-        assert modification_time.strftime(self._DATE_TIME_FORMAT) == now.strftime(self._DATE_TIME_FORMAT)
+        modification_time = self._storage.get_modified_time(
+            self._TEST_FILE_NAME
+        )
+        assert modification_time - now < timedelta(seconds=1)
 
     def test_file_access_time(self):
         now = datetime.utcnow()
         self._write_contents_to_file()
         access_time = self._storage.get_access_time(self._TEST_FILE_NAME)
-        assert access_time.strftime(self._DATE_TIME_FORMAT) == now.strftime(self._DATE_TIME_FORMAT)
+        assert access_time - now < timedelta(seconds=1)
 
     def _write_contents_to_file(self, binary: bool = False):
         self._storage.write(
             self._TEST_FILE_NAME,
-            content=self._TEST_FILE_CONTENT_BINARY if binary else self._TEST_FILE_CONTENT,
-            mode="xb" if binary else "x"
+            content=self._TEST_FILE_CONTENT_BINARY
+            if binary
+            else self._TEST_FILE_CONTENT,
+            mode="xb" if binary else "x",
         )