add support for root mount

Author: igorbenav

crudadmin/admin_interface/admin_site.py

@@ -144,6 +144,10 @@ def __init__(
 
         self.secure_cookies: bool = secure_cookies
 
+    def get_url_prefix(self) -> str:
+        """Get the URL prefix for admin routes, handling root mount path correctly."""
+        return f"/{self.mount_path}" if self.mount_path else ""
+
     def setup_routes(self) -> None:
         """
         Configure all admin interface routes including auth, dashboard and model views.
@@ -270,16 +274,17 @@ async def login_page_inner(
 
                     logger.info(f"Session created successfully: {session_id}")
 
-                    response = RedirectResponse(
-                        url=f"/{self.mount_path}/", status_code=303
+                    dashboard_url = (
+                        f"{self.get_url_prefix()}/" if self.mount_path else "/"
                     )
+                    response = RedirectResponse(url=dashboard_url, status_code=303)
 
                     self.session_manager.set_session_cookies(
                         response=response,
                         session_id=session_id,
                         csrf_token=csrf_token,
                         secure=self.secure_cookies,
-                        path=f"/{self.mount_path}",
+                        path=f"{self.get_url_prefix()}/" if self.mount_path else "/",
                     )
 
                     await db.commit()
@@ -340,13 +345,12 @@ async def logout_endpoint_inner(
             if session_id:
                 await self.session_manager.terminate_session(session_id=session_id)
 
-            response = RedirectResponse(
-                url=f"/{self.mount_path}/login", status_code=303
-            )
+            login_url = f"{self.get_url_prefix()}/login"
+            response = RedirectResponse(url=login_url, status_code=303)
 
             self.session_manager.clear_session_cookies(
                 response=response,
-                path=f"/{self.mount_path}",
+                path=f"{self.get_url_prefix()}/" if self.mount_path else "/",
             )
 
             return response
@@ -380,9 +384,10 @@ async def admin_login_page_inner(
                     )
 
                     if is_valid_session:
-                        return RedirectResponse(
-                            url=f"/{self.mount_path}/", status_code=303
+                        dashboard_url = (
+                            f"{self.get_url_prefix()}/" if self.mount_path else "/"
                         )
+                        return RedirectResponse(url=dashboard_url, status_code=303)
 
             except Exception:
                 pass

crudadmin/admin_interface/crud_admin.py

@@ -318,7 +318,12 @@ def __init__(
         track_events: bool = False,
         track_sessions_in_db: bool = False,
     ) -> None:
-        self.mount_path = mount_path.strip("/") if mount_path else "admin"
+        if mount_path == "/":
+            self.mount_path = ""
+        elif mount_path:
+            self.mount_path = mount_path.strip("/")
+        else:
+            self.mount_path = "admin"
         self.theme = theme or "dark-theme"
         self.track_events = track_events
         self.track_sessions_in_db = track_sessions_in_db
@@ -376,7 +381,9 @@ def __init__(
         self.initial_admin = initial_admin
         self.models: Dict[str, ModelConfig] = {}
         self.router = APIRouter(tags=["admin"])
-        self.oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"/{self.mount_path}/login")
+        self.oauth2_scheme = OAuth2PasswordBearer(
+            tokenUrl=f"{self.get_url_prefix()}/login"
+        )
         self.secure_cookies = secure_cookies
 
         session_backend = getattr(self, "_session_backend", "memory")
@@ -434,6 +441,10 @@ def __init__(
 
         self.app.include_router(self.router)
 
+    def get_url_prefix(self) -> str:
+        """Get the URL prefix for admin routes, handling root mount path correctly."""
+        return f"/{self.mount_path}" if self.mount_path else ""
+
     async def initialize(self) -> None:
         """
         Initialize admin database tables and create initial admin user.

crudadmin/admin_interface/middleware/auth.py

@@ -38,7 +38,12 @@ def _should_add_cache_headers(self, response: Response) -> bool:
         return not (300 <= response.status_code < 400)
 
     async def dispatch(self, request: Request, call_next):
-        if not request.url.path.startswith(f"/{self.admin_instance.mount_path}/"):
+        expected_prefix = (
+            f"/{self.admin_instance.mount_path}/"
+            if self.admin_instance.mount_path
+            else "/"
+        )
+        if not request.url.path.startswith(expected_prefix):
             return await call_next(request)
 
         is_login_path = request.url.path.endswith("/login")
@@ -58,8 +63,9 @@ async def dispatch(self, request: Request, call_next):
 
                 if not session_id:
                     logger.debug("Missing session_id")
+                    login_url = f"{self.admin_instance.get_url_prefix()}/login?error=Please+log+in+to+access+this+page"
                     return RedirectResponse(
-                        url=f"/{self.admin_instance.mount_path}/login?error=Please+log+in+to+access+this+page",
+                        url=login_url,
                         status_code=303,
                     )
 
@@ -72,8 +78,9 @@ async def dispatch(self, request: Request, call_next):
 
                     if not session_data:
                         logger.debug("Invalid or expired session")
+                        login_url = f"{self.admin_instance.get_url_prefix()}/login?error=Session+expired"
                         return RedirectResponse(
-                            url=f"/{self.admin_instance.mount_path}/login?error=Session+expired",
+                            url=login_url,
                             status_code=303,
                         )
 
@@ -84,8 +91,9 @@ async def dispatch(self, request: Request, call_next):
 
                     if not user:
                         logger.debug("User not found for session")
+                        login_url = f"{self.admin_instance.get_url_prefix()}/login?error=User+not+found"
                         return RedirectResponse(
-                            url=f"/{self.admin_instance.mount_path}/login?error=User+not+found",
+                            url=login_url,
                             status_code=303,
                         )
 
@@ -107,8 +115,9 @@ async def dispatch(self, request: Request, call_next):
                         or "/crud/" in request.url.path
                     ):
                         raise
+                    login_url = f"{self.admin_instance.get_url_prefix()}/login?error=Authentication+error"
                     return RedirectResponse(
-                        url=f"/{self.admin_instance.mount_path}/login?error=Authentication+error",
+                        url=login_url,
                         status_code=303,
                     )
 

crudadmin/admin_interface/model_view.py

@@ -423,6 +423,14 @@ def __init__(
 
         self.setup_routes()
 
+    def get_url_prefix(self) -> str:
+        """Get the URL prefix for admin routes, handling root mount path correctly."""
+        if self.admin_site:
+            return (
+                f"/{self.admin_site.mount_path}" if self.admin_site.mount_path else ""
+            )
+        return ""
+
     def _model_is_admin_model(self, model: Type[DeclarativeBase]) -> bool:
         """Check if the given model is one of the admin-specific models."""
         admin_model_names = [
@@ -638,15 +646,16 @@ async def form_create_endpoint_inner(
 
                         if result:
                             request.state.crud_result = result
+                            model_list_url = (
+                                f"{self.get_url_prefix()}/{self.model.__name__}/"
+                            )
                             if "HX-Request" in request.headers:
                                 return RedirectResponse(
-                                    url=f"/{self.admin_site.mount_path}/{self.model.__name__}/",
-                                    headers={
-                                        "HX-Redirect": f"/{self.admin_site.mount_path}/{self.model.__name__}/"
-                                    },
+                                    url=model_list_url,
+                                    headers={"HX-Redirect": model_list_url},
                                 )
                             return RedirectResponse(
-                                url=f"/{self.admin_site.mount_path}/{self.model.__name__}/",
+                                url=model_list_url,
                                 status_code=303,
                             )
 
@@ -668,7 +677,7 @@ async def form_create_endpoint_inner(
                 "error": error_message,
                 "field_errors": field_errors,
                 "field_values": field_values,
-                "mount_path": self.admin_site.mount_path,
+                "mount_path": self.admin_site.mount_path if self.admin_site else "",
             }
 
             return self.templates.TemplateResponse(
@@ -827,7 +836,7 @@ async def bulk_delete_endpoint_inner(
                     "current_page": adjusted_page,
                     "rows_per_page": rows_per_page,
                     "primary_key_info": primary_key_info,
-                    "mount_path": self.admin_site.mount_path,
+                    "mount_path": self.admin_site.mount_path if self.admin_site else "",
                 }
 
                 return self.templates.TemplateResponse(
@@ -1204,8 +1213,11 @@ async def form_update_endpoint_inner(
                             )
                             await db.commit()
 
+                        model_list_url = (
+                            f"{self.get_url_prefix()}/{self.model.__name__}/"
+                        )
                         return RedirectResponse(
-                            url=f"/{self.admin_site.mount_path}/{self.model.__name__}/",
+                            url=model_list_url,
                             status_code=303,
                         )
 
@@ -1232,7 +1244,7 @@ async def form_update_endpoint_inner(
                 "error": error_message,
                 "field_errors": field_errors,
                 "field_values": field_values,
-                "mount_path": self.admin_site.mount_path,
+                "mount_path": self.admin_site.mount_path if self.admin_site else "",
                 "id": id,
                 "include_sidebar_and_header": False,
             }

tests/crud/test_admin.py

@@ -98,6 +98,30 @@ async def test_crud_admin_with_custom_settings(async_session):
     assert admin.track_events is True
 
 
+@pytest.mark.asyncio
+async def test_crud_admin_root_mount_path(async_session):
+    """Test CRUDAdmin initialization with root mount path."""
+    secret_key = "test-secret-key-for-testing-only-32-chars"
+    db_config = create_test_db_config(async_session)
+
+    admin = CRUDAdmin(
+        session=async_session,
+        SECRET_KEY=secret_key,
+        mount_path="/",
+        db_config=db_config,
+        setup_on_initialization=False,
+    )
+
+    # Test that mount_path is properly set to empty string for root
+    assert admin.mount_path == ""
+
+    # Test that URL prefix is correctly generated
+    assert admin.get_url_prefix() == ""
+
+    # Test that OAuth2 token URL is correctly set for root path
+    assert admin.oauth2_scheme.model.flows.password.tokenUrl == "/login"
+
+
 @pytest.mark.asyncio
 async def test_crud_admin_with_allowed_ips(async_session):
     """Test CRUDAdmin initialization with IP restrictions."""