improve files middlewares and example

examples/middleware_files/application.py

@@ -0,0 +1,10 @@
+from asgikit.responses import respond_text
+from selva.web import controller, get
+
+
+@controller
+class Controller:
+    @get
+    async def index(self, request):
+        request.response.content_type = "text/html"
+        await respond_text(request.response, "Lorem ipsum dolor sit amet")

examples/middleware_files/configuration/settings.yaml

@@ -0,0 +1,6 @@
+middleware:
+  - selva.web.middleware.files.StaticFilesMiddleware
+  - selva.web.middleware.files.UploadedFilesMiddleware
+staticfiles:
+  mappings:
+    favicon.ico: python.ico

examples/middleware_files/resources/uploads/lorem-ipsum.txt

@@ -0,0 +1 @@
+Lorem ipsum dolor sit amet, consectetur adipiscing elit. Etiam congue feugiat tortor sit amet bibendum. Phasellus euismod risus eget lorem ornare, ac porta quam lobortis. Sed eget est at nulla scelerisque tincidunt. Aliquam condimentum purus id nisl vestibulum, nec eleifend magna interdum. Curabitur lacinia libero sed nisl mollis, et accumsan justo efficitur. Integer sit amet libero eget lorem varius faucibus vel quis sem. Duis consequat tempor orci eu porta. Sed ut metus porta, elementum nunc a, tincidunt massa. Sed volutpat tincidunt odio, vel venenatis est ultrices ut. Morbi et purus sed sapien elementum commodo. Sed rutrum odio vel magna dignissim laoreet.

pyproject.toml

@@ -49,7 +49,8 @@ memcached = ["emcache"]
 optional = true
 
 [tool.poetry.group.dev.dependencies]
-uvicorn = { version = "^0.23", extras = ["standard"] }
+uvicorn = { version = "^0.29", extras = ["standard"] }
+granian = "^1.3"
 
 [tool.poetry.group.test]
 optional = true

src/selva/web/middleware/files.py

@@ -1,3 +1,4 @@
+from abc import ABCMeta, abstractmethod
 from pathlib import Path
 from typing import Annotated
 
@@ -10,58 +11,67 @@
 from selva.web.exception import HTTPNotFoundException
 
 
-class StaticFilesMiddleware(Middleware):
+class BaseFilesMiddleware(Middleware, metaclass=ABCMeta):
     settings: Annotated[Settings, Inject]
+    settings_property: str
+
+    path: str
+    root: Path
 
     def initialize(self):
-        self.path = self.settings.staticfiles.path
-        self.root = Path(self.settings.staticfiles.root).resolve()
-        self.mappings = {
-            name.lstrip("/"): value.lstrip("/") for name, value in
-            self.settings.staticfiles.get("mappings").items()
-        }
+        settings = self.settings.get(self.settings_property)
+        self.path = settings.path.lstrip("/")
+        self.root = Path(settings.root).resolve()
+
+    @abstractmethod
+    def get_file_to_serve(self, request: Request) -> str | None:
+        pass
+
+    async def __call__(self, call_next: CallNext, request: Request):
+        if file_to_serve := self.get_file_to_serve(request):
+            file_to_serve = (self.root / file_to_serve).resolve()
+            if not (
+                file_to_serve.is_file() and file_to_serve.is_relative_to(self.root)
+            ):
+                raise HTTPNotFoundException()
+
+            await respond_file(request.response, file_to_serve)
+        else:
+            await call_next(request)
 
-    async def __call__(
-        self,
-        call_next: CallNext,
-        request: Request,
-    ):
-        file_to_serve = self.mappings.get(request.path.lstrip("/"))
 
-        if not file_to_serve:
-            if not request.path.startswith(self.path):
-                await call_next(request)
-                return
+class UploadedFilesMiddleware(BaseFilesMiddleware):
+    settings_property = "uploadedfiles"
 
-            file_to_serve = request.path.removeprefix(self.path)
+    def get_file_to_serve(self, request: Request) -> str | None:
+        request_path = request.path.lstrip("/")
 
-        file_to_serve = file_to_serve.lstrip("/")
-        path_to_serve = (self.root / file_to_serve).resolve()
-        if not path_to_serve.is_relative_to(self.root) or not path_to_serve.exists() or not path_to_serve.is_file():
-            raise HTTPNotFoundException()
+        if request_path.startswith(self.path):
+            return request_path.removeprefix(self.path).lstrip("/")
 
-        await respond_file(request.response, path_to_serve)
+        return None
 
 
-class UploadedFilesMiddleware(Middleware):
-    settings: Annotated[Settings, Inject]
+class StaticFilesMiddleware(BaseFilesMiddleware):
+    settings_property = "staticfiles"
+    mappings: dict[str, str]
 
     def initialize(self):
-        self.path = self.settings.uploadedfiles.path
-        self.root = Path(self.settings.uploadedfiles.root).resolve()
-
-    async def __call__(
-        self,
-        call_next: CallNext,
-        request: Request,
-    ):
-        if not request.path.startswith(self.path):
-            await call_next(request)
-            return
+        super().initialize()
+
+        settings = self.settings.get(self.settings_property)
+        self.mappings = {
+            name.lstrip("/"): value.lstrip("/")
+            for name, value in settings.get("mappings", {}).items()
+        }
+
+    def get_file_to_serve(self, request: Request) -> str | None:
+        request_path = request.path.lstrip("/")
+
+        if file_to_serve := self.mappings.get(request_path):
+            return file_to_serve.lstrip("/")
 
-        file_to_serve = request.path.removeprefix(self.path).lstrip("/")
-        path_to_serve = (self.root / file_to_serve).resolve()
-        if not path_to_serve.is_relative_to(self.root) or not path_to_serve.exists() or not path_to_serve.is_file():
-            raise HTTPNotFoundException()
+        if request_path.startswith(self.path):
+            return request_path.removeprefix(self.path).lstrip("/")
 
-        await respond_file(request.response, path_to_serve)
+        return None