redirect handling in middleware

Author: boddumanohar

src/api/__init__.py

@@ -4,14 +4,17 @@
 import re
 import sys
 from importlib.resources import files
-from typing import Any, Literal
+from typing import Any, Literal, MutableMapping
+from urllib.parse import urlparse, urlunparse
 
 from alembic import command
 from alembic.config import Config
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.routing import APIRoute
 from pydantic import BaseModel
+from starlette.datastructures import MutableHeaders
+from starlette.types import ASGIApp, Receive, Scope, Send
 
 from ..deployment.monitors.health import vm_monitor
 from .backup import router as backup_router
@@ -95,6 +98,39 @@ def _logging_config() -> dict[str, Any]:
 logger = logging.getLogger(__name__)
 
 
+class _RootPathRedirectMiddleware:
+    """Fix Starlette's automatic redirect responses to include root_path.
+
+    Starlette's Router generates trailing-slash redirects (307) using
+    URL(scope=...) which only reads scope["path"] and ignores
+    scope["root_path"]. This middleware rewrites the Location header on
+    redirect responses so the root_path prefix is included.
+    """
+
+    def __init__(self, app: ASGIApp, root_path: str = "") -> None:
+        self.app = app
+        self.root_path = root_path
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http" or not self.root_path:
+            await self.app(scope, receive, send)
+            return
+
+        root_path = self.root_path
+
+        async def send_wrapper(message: MutableMapping[str, Any]) -> None:
+            if message["type"] == "http.response.start" and 300 <= message.get("status", 200) < 400:
+                headers = MutableHeaders(scope=message)
+                location = headers.get("location")
+                if location:
+                    parsed = urlparse(location)
+                    if parsed.path and not parsed.path.startswith(root_path):
+                        headers["location"] = urlunparse(parsed._replace(path=root_path + parsed.path))
+            await send(message)
+
+        await self.app(scope, receive, send_wrapper)
+
+
 class _FastAPI(FastAPI):
     def openapi(self) -> dict[str, Any]:
         if self.openapi_schema:
@@ -220,6 +256,7 @@ def _use_route_names_as_operation_ids(app: FastAPI) -> None:
 
 app = _FastAPI(openapi_tags=_tags, root_path=get_settings().root_path)
 
+app.add_middleware(_RootPathRedirectMiddleware, root_path=get_settings().root_path)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=get_settings().cors_origins,

src/api/_util/__init__.py

@@ -30,4 +30,4 @@ def wrapper(*args, **kwargs):
 
 
 def url_path_for(request: Request, name: str, **kwargs) -> str:
-    return request.scope.get("root_path") + request.app.url_path_for(name, **kwargs)
+    return request.url_for(name, **kwargs).path