✨ Migrated to deta from imgur for image-storing

classquiz/storage/__init__.py

@@ -0,0 +1,25 @@
+from .deta_storage import DetaStorage
+from io import BytesIO
+
+
+class Storage:
+    def __init__(self, backend: str, deta_key: str | None, deta_id: str | None):
+        self.backend = backend
+        self.deta_key = deta_key
+        self.deta_id = deta_id
+        self.deta_base_url = f"https://drive.deta.sh/v1/{deta_id}/classquiz"
+        self.deta_instance = DetaStorage(deta_base_url=self.deta_base_url, deta_key=self.deta_key, deta_id=self.deta_id)
+        if backend == "deta":
+            if deta_key is None or deta_id is None:
+                raise ValueError("deta_key and deta_id must be provided")
+        else:
+            raise NotImplementedError(f"Backend {backend} not implemented")
+
+    async def download(self, file_name: str) -> BytesIO | None:
+        if self.backend == "deta":
+            return await self.deta_instance.download(
+                file_name)
+
+    async def upload(self, file_name: str, file_data: bytes) -> None:
+        if self.backend == "deta":
+            return await self.deta_instance.upload(file=file_data, file_name=file_name)

classquiz/storage/deta_storage.py

@@ -0,0 +1,43 @@
+from classquiz.config import settings
+from aiohttp import ClientSession
+from io import BytesIO
+
+settings = settings()
+
+
+class DetaStorage:
+    def __init__(self, deta_base_url: str, deta_id: str, deta_key: str):
+        self.deta_url = deta_base_url
+        self.deta_id = deta_id
+        self.deta_key = deta_key
+        self.headers = {
+            "X-Api-Key": self.deta_key,
+        }
+
+    async def download(self, file_name: str) -> BytesIO | None:
+        """
+
+        :param file_name: The name of the file to be downloaded
+        :return: Either bytes f successfull download or None if failed
+        """
+        async with ClientSession(headers=self.headers) as session:
+            async with session.get(f"{self.deta_url}/files/download?name={file_name}") as response:
+                if response.status == 200:
+                    return BytesIO(await response.read())
+                elif response.status == 404:
+                    return None
+                else:
+                    raise Exception("Download failed")
+
+    async def upload(self, file: bytes, file_name: str) -> None:
+        """
+        :param file: The file in bytes
+        :param file_name: The name of the file
+        :return:
+        """
+        async with ClientSession(headers=self.headers) as session:
+            async with session.post(f"{self.deta_url}/files?name={file_name}", data=file) as response:
+                if response.status == 201:
+                    return None
+                else:
+                    raise Exception("Upload failed")