✨ Added local-file-system storage

classquiz/config.py

@@ -26,8 +26,11 @@ class Settings(BaseSettings):
     # storage_backend
     storage_backend: str | None = "deta"
     # if storage_backend == "deta":
-    deta_project_key: str
-    deta_project_id: str
+    deta_project_key: str | None
+    deta_project_id: str | None
+
+    # if storage_backend == "local":
+    storage_path: str | None
 
     class Config:
         env_file = ".env"

classquiz/kahoot_importer/import_quiz.py

@@ -66,7 +66,7 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | str:
     quiz_questions: list[dict] = []
     quiz_id = uuid.uuid4()
     storage = Storage(backend=settings.storage_backend, deta_key=settings.deta_project_key,
-                      deta_id=settings.deta_project_id)
+                      deta_id=settings.deta_project_id, storage_path=settings.storage_path)
 
     for q in quiz.kahoot.questions:
         answers: list[QuizAnswer] = []

classquiz/routers/storage.py

@@ -1,19 +1,23 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException
 from fastapi.responses import StreamingResponse
 from classquiz.config import settings
-import io
+import re
 from classquiz.storage import Storage
 
 settings = settings()
 
 router = APIRouter()
 
+file_regex = r"^[a-z0-9]{8}-[a-z0-9-]{27}--[a-z0-9-]{36}$"
+
 
 @router.get('/download/{file_name}')
 async def download_file(file_name: str):
     storage = Storage(backend=settings.storage_backend, deta_key=settings.deta_project_key,
-                      deta_id=settings.deta_project_id)
+                      deta_id=settings.deta_project_id, storage_path=settings.storage_path)
     download = await storage.download(file_name)
+    if not re.match(file_regex, file_name):
+        raise HTTPException(status_code=400, detail="Invalid file name")
 
     def iter_file():
         yield from download

classquiz/storage/__init__.py

@@ -1,17 +1,22 @@
 from .deta_storage import DetaStorage
+from .local_storage import LocalStorage
 from io import BytesIO
 
 
 class Storage:
-    def __init__(self, backend: str, deta_key: str | None, deta_id: str | None):
+    def __init__(self, backend: str, deta_key: str | None, deta_id: str | None, storage_path: 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)
+        self.local_instance = LocalStorage(base_path=storage_path)
         if backend == "deta":
             if deta_key is None or deta_id is None:
                 raise ValueError("deta_key and deta_id must be provided")
+        if backend == "local":
+            if storage_path is None:
+                raise ValueError("storage_path must be provided")
         else:
             raise NotImplementedError(f"Backend {backend} not implemented")
 
@@ -19,7 +24,11 @@ class Storage:
         if self.backend == "deta":
             return await self.deta_instance.download(
                 file_name)
+        elif self.backend == "local":
+            return await self.local_instance.get_file(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)
+        elif self.backend == "local":
+            return await self.local_instance.write_file(file_name=file_name, data=file_data)

classquiz/storage/local_storage.py

@@ -0,0 +1,17 @@
+import io
+import os
+import aiofiles
+
+
+class LocalStorage:
+    def __init__(self, base_path: str):
+        self.base_path = base_path
+
+    async def get_file(self, file_name: str) -> io.BytesIO:
+        async with aiofiles.open(file=os.path.join(self.base_path, file_name), mode='rb') as f:
+            return io.BytesIO(await f.read())
+
+    async def write_file(self, file_name: str, data: bytes) -> None:
+        async with aiofiles.open(file=os.path.join(self.base_path, file_name), mode='wb') as f:
+            await f.write(data)
+