feat(static): add StaticFiles and App.mount

oxyroute/__init__.py

@@ -10,6 +10,7 @@
 from oxyroute.router import APIRouter
 from oxyroute.security_headers import SecurityHeadersConfig
 from oxyroute.sse import SSEEvent, send_sse, sse_done
+from oxyroute.static import StaticFiles
 from oxyroute.streaming import stream_bytes, stream_jsonl, stream_text
 
 __all__ = [
@@ -24,6 +25,7 @@
     "Response",
     "SSEEvent",
     "SecurityHeadersConfig",
+    "StaticFiles",
     "WebSocket",
     "__version__",
     "apply_cors",

oxyroute/app.py

@@ -132,6 +132,15 @@ def set_security_headers(self, config: Any | None) -> None:
         """
         self._app.set_security_headers(config)
 
+    def mount(self, path: str, app: Any) -> None:
+        """Mount another application or handler at a specific path prefix."""
+        path = path.rstrip("/")
+        # Mount the exact prefix
+        self.get(path)(app)
+        self.get(path + "/")(app)
+        # Mount all subpaths
+        self.get(path + "/*path")(app)
+
     def get(
         self,
         path: str,

oxyroute/router.py

@@ -67,6 +67,15 @@ def include_router(
             full = join_path(prefix, rel)
             self._routes.append((method, full, handler, merged))
 
+    def mount(self, path: str, app: Any) -> None:
+        """Mount another application or handler at a specific path prefix."""
+        path = path.rstrip("/")
+        # Mount the exact prefix
+        self.get(path)(app)
+        self.get(path + "/")(app)
+        # Mount all subpaths
+        self.get(path + "/*path")(app)
+
     def get(
         self,
         path: str,

oxyroute/static.py

@@ -0,0 +1,66 @@
+import mimetypes
+import os
+from typing import Any
+
+from oxyroute.exceptions import HTTPException
+
+
+class StaticFiles:
+    """
+    Serve static files from a directory.
+
+    Can be mounted via:
+        app.mount("/static", StaticFiles("static", html=True))
+    """
+
+    __name__ = "StaticFiles"
+
+    def __init__(
+        self,
+        directory: str,
+        html: bool = False,
+        max_age: int | None = None,
+    ) -> None:
+        self.directory = os.path.abspath(directory)
+        if not os.path.isdir(self.directory):
+            raise RuntimeError(f"Directory {directory} does not exist")
+        self.html = html
+        self.max_age = max_age
+
+    def __call__(self, protocol: Any, path: str = "") -> Any:
+        if ".." in path.split("/"):
+            raise HTTPException(status_code=403, detail="Forbidden")
+
+        file_path = os.path.abspath(os.path.join(self.directory, path.lstrip("/")))
+
+        if not file_path.startswith(self.directory):
+            raise HTTPException(status_code=403, detail="Forbidden")
+
+        if not os.path.exists(file_path) or not os.path.isfile(file_path):
+            if self.html and os.path.isfile(os.path.join(file_path, "index.html")):
+                file_path = os.path.join(file_path, "index.html")
+            else:
+                raise HTTPException(status_code=404, detail="Not Found")
+
+        content_type, _ = mimetypes.guess_type(file_path)
+        if content_type is None:
+            content_type = "application/octet-stream"
+
+        headers = [("content-type", content_type)]
+        if self.max_age is not None:
+            headers.append(("cache-control", f"public, max-age={self.max_age}"))
+
+        if hasattr(protocol, "response_file"):
+            # Rust fast path via tokio-fs
+            protocol.response_file(200, headers, file_path)
+            from oxyroute.streaming import stream_done
+
+            return stream_done()
+        else:
+            # Fallback for Python testing transports without response_file
+            with open(file_path, "rb") as f:
+                body = f.read()
+            protocol.response_bytes(200, headers, body)
+            from oxyroute.streaming import stream_done
+
+            return stream_done()

oxyroute/testing.py

@@ -250,6 +250,16 @@ def response_bytes(self, status: int, headers: list, body: bytes) -> None:
             }
         )
 
+    def response_file(self, status: int, headers: list, file: str) -> None:
+        import anyio
+
+        async def _read_file() -> bytes:
+            async with await anyio.open_file(file, "rb") as f:
+                return await f.read()
+
+        body = asyncio.run_coroutine_threadsafe(_read_file(), self._loop).result()
+        self.response_bytes(status, headers, body)
+
     def response_empty(self, status: int, headers: list) -> None:
         self._status = int(status)
         self.status = int(status)

tests/test_static.py

@@ -0,0 +1,66 @@
+import os
+from tempfile import TemporaryDirectory
+
+from oxyroute import App, StaticFiles
+from oxyroute.testing import TestClient
+
+
+def test_static_files():
+    with TemporaryDirectory() as tmpdir:
+        with open(os.path.join(tmpdir, "test.txt"), "w") as f:
+            f.write("hello world")
+
+        app = App()
+        app.mount("/static", StaticFiles(tmpdir))
+
+        with TestClient(app) as client:
+            resp = client.get("/static/test.txt")
+            assert resp.status_code == 200
+            assert resp.content == b"hello world"
+            assert resp.headers["content-type"] == "text/plain"
+
+
+def test_static_files_missing():
+    with TemporaryDirectory() as tmpdir:
+        app = App()
+        app.mount("/static", StaticFiles(tmpdir))
+
+        with TestClient(app) as client:
+            resp = client.get("/static/missing.txt")
+            assert resp.status_code == 404
+
+
+def test_static_files_traversal():
+    with TemporaryDirectory() as tmpdir:
+        # Create a file outside the static directory
+        with open(os.path.join(tmpdir, "secret.txt"), "w") as f:
+            f.write("secret")
+
+        static_dir = os.path.join(tmpdir, "static")
+        os.makedirs(static_dir)
+
+        app = App()
+        app.mount("/static", StaticFiles(static_dir))
+
+        with TestClient(app) as client:
+            resp = client.get("/static/../secret.txt")
+            assert resp.status_code in (403, 404)
+
+
+def test_static_files_index_html():
+    with TemporaryDirectory() as tmpdir:
+        with open(os.path.join(tmpdir, "index.html"), "w") as f:
+            f.write("<h1>Hello</h1>")
+
+        app = App()
+        app.mount("/static", StaticFiles(tmpdir, html=True))
+
+        with TestClient(app) as client:
+            resp = client.get("/static/")
+            assert resp.status_code == 200
+            assert resp.content == b"<h1>Hello</h1>"
+            assert resp.headers["content-type"] == "text/html"
+
+            resp = client.get("/static")
+            assert resp.status_code == 200
+            assert resp.content == b"<h1>Hello</h1>"