feat: storage facede

serpens/storages.py

@@ -0,0 +1,37 @@
+import importlib
+import logging
+import os
+from enum import Enum
+from typing import Any, Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class StorageProvider(Enum):
+    S3 = "s3"
+    CLOUD_STORAGE = "cloud_storage"
+
+
+class StorageClient:
+    _instance = None
+
+    def __init__(self, provider: Optional[StorageProvider] = None):
+        self._provider = provider or StorageProvider(os.getenv("STORAGE_PROVIDER"))
+        logger.debug(f"Provider: {self._provider.value}")
+        module = importlib.import_module(f"serpens.{self._provider.value}")
+        self._get_object = module.get_object
+
+    def get(
+        self,
+        bucket: str,
+        key: str,
+    ) -> Dict[str, Any]:
+        object = self._get_object(bucket, key)
+
+        return object
+
+    @classmethod
+    def instance(cls):
+        if cls._instance is None:
+            cls._instance = cls()
+        return cls._instance

tests/test_storage.py

@@ -0,0 +1,24 @@
+import os
+import unittest
+from unittest.mock import patch
+
+from serpens.storages import StorageClient
+
+
+class TestStorages(unittest.TestCase):
+    def setUp(self):
+        self.patch_boto3 = patch("serpens.s3.boto3")
+        self.mock_boto3 = self.patch_boto3.start()
+        self.s3_client = self.mock_boto3.client.return_value
+
+        self.bucket = ""
+        self.key = ""
+
+    def tearDown(self):
+        self.patch_boto3.stop()
+
+    @patch.dict(os.environ, {"STORAGE_PROVIDER": "s3"})
+    def test_get_object_s3(self):
+        StorageClient.instance().get(self.bucket, self.key)
+
+        self.s3_client.get_object.assert_called_once_with(Bucket=self.bucket, Key=self.key)