From d738d90cefb5f518cd5a5fd104ac871a10587676 Mon Sep 17 00:00:00 2001 From: dwindown Date: Wed, 1 Apr 2026 19:49:22 +0700 Subject: [PATCH] Replace incompatible admin CRUD landing with SQLAlchemy-safe pages --- app/admin.py | 203 +++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 166 insertions(+), 37 deletions(-) diff --git a/app/admin.py b/app/admin.py index 30b0839..6b6dd15 100644 --- a/app/admin.py +++ b/app/admin.py @@ -24,9 +24,8 @@ from fastapi_admin.resources import ( from fastapi_admin.template import templates from fastapi_admin.widgets import displays, inputs from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint -from starlette.responses import RedirectResponse +from starlette.responses import HTMLResponse, RedirectResponse from starlette.status import HTTP_303_SEE_OTHER, HTTP_401_UNAUTHORIZED from app.core.config import get_settings @@ -121,30 +120,7 @@ class EnvCredentialProvider(Provider): @staticmethod def _admin_home(request: Request) -> str: - """ - Resolve a concrete admin page path. - - This project uses SQLAlchemy models, while fastapi-admin's built-in - Model CRUD pages are Tortoise-oriented. Prefer custom Link resources - and known safe admin pages. - """ - admin_path = request.app.admin_path.rstrip("/") - for resource in getattr(request.app, "resources", []): - try: - if issubclass(resource, Model): - model_name = resource.model.__name__.lower() - return f"{admin_path}/{model_name}/list" - except TypeError: - continue - for resource in getattr(request.app, "resources", []): - try: - if issubclass(resource, Link): - url = getattr(resource, "url", "") - if isinstance(url, str) and url.startswith("/"): - return url - except TypeError: - continue - return f"{admin_path}/password" + return request.app.admin_path.rstrip("/") + "/dashboard" @staticmethod def _login_url(request: Request) -> str: @@ -768,6 +744,153 @@ class SessionOverviewLink(Link): # Initialize FastAPI Admin # ============================================================================= +def _render_admin_page(title: str, page_title: str, body: str) -> HTMLResponse: + html = f""" + + + + + {title} + + + +
+ +
+
+

{page_title}

+ {body} +
+
+
+ +""" + return HTMLResponse(html) + + +def _table(headers: list[str], rows: list[list[Any]]) -> str: + head = "".join(f"{header}" for header in headers) + body_rows = [] + for row in rows: + cols = "".join(f"{value}" for value in row) + body_rows.append(f"{cols}") + body = "".join(body_rows) or f"No data" + return f"{head}{body}
" + + +def _prune_incompatible_admin_routes() -> None: + admin_app.router.routes[:] = [ + route + for route in admin_app.router.routes + if not getattr(route, "path", "").startswith("/{resource}/") + ] + + +async def dashboard_view(request: Request, admin: AdminPrincipal = Depends(get_current_admin)): + _ = admin + body = ( + '

This admin runs in SQLAlchemy-safe mode. ' + "The original fastapi-admin CRUD pages depend on Tortoise ORM and were removed.

" + "

Use the navigation links to inspect operational data.

" + ) + return _render_admin_page("IRT Bank Soal Admin", "Dashboard", body) + + +async def calibration_status_view( + request: Request, + admin: AdminPrincipal = Depends(get_current_admin), +): + _ = admin + data = await CalibrationDashboardLink().get(request) + rows = [ + [ + item["tryout_id"], + item["name"], + item["total_items"], + item["calibrated_items"], + f'{item["calibration_percentage"]:.2f}%', + "Yes" if item["ready_for_irt"] else "No", + ] + for item in data["data"] + ] + body = _table( + ["Tryout ID", "Name", "Total Items", "Calibrated", "Calibration %", "Ready for IRT"], + rows, + ) + return _render_admin_page("Calibration Status", "Calibration Status", body) + + +async def item_statistics_view( + request: Request, + admin: AdminPrincipal = Depends(get_current_admin), +): + _ = admin + data = await ItemStatisticsLink().get(request) + rows = [ + [ + item["level"], + item["total_items"], + item["calibrated_items"], + f'{item["calibration_percentage"]:.2f}%', + item["total_responses"], + f'{item["avg_correctness"]:.4f}', + ] + for item in data["data"] + ] + body = _table( + ["Level", "Total Items", "Calibrated", "Calibration %", "Responses", "Avg Correctness"], + rows, + ) + return _render_admin_page("Item Statistics", "Item Statistics", body) + + +async def session_overview_view( + request: Request, + admin: AdminPrincipal = Depends(get_current_admin), +): + _ = admin + data = await SessionOverviewLink().get(request) + rows = [ + [ + item["session_id"], + item["wp_user_id"], + item["tryout_id"], + "Yes" if item["is_completed"] else "No", + item["scoring_mode_used"], + item["total_benar"], + item["NM"], + item["NN"], + item["theta"], + ] + for item in data["data"] + ] + body = _table( + ["Session ID", "WP User", "Tryout", "Completed", "Mode", "Benar", "NM", "NN", "Theta"], + rows, + ) + return _render_admin_page("Session Overview", "Session Overview", body) + def create_admin_app() -> Any: """ Create and configure FastAPI Admin application. @@ -784,23 +907,29 @@ def create_admin_app() -> Any: # NOTE: fastapi-admin 1.0.4 requires provider registration via app.configure(...). # Keep provider implementation here for future integration during startup configure. - # NOTE: - # fastapi-admin Model resources rely on Tortoise ORM query APIs. - # This codebase uses SQLAlchemy, so register only Link resources here. - # Keep Model resource classes in source for future migration work. + # Reset singleton registries so stale state cannot survive restarts. + admin_app.resources = [] + admin_app.model_resources = {} + _prune_incompatible_admin_routes() - # Register dashboard links (safe for SQLAlchemy-backed custom views) admin_app.register(CalibrationDashboardLink) admin_app.register(ItemStatisticsLink) admin_app.register(SessionOverviewLink) - calibration_link = CalibrationDashboardLink() - item_stats_link = ItemStatisticsLink() - session_overview_link = SessionOverviewLink() + admin_app.get("/dashboard", dependencies=[Depends(get_current_admin)])(dashboard_view) + admin_app.get("/calibration_status", dependencies=[Depends(get_current_admin)])(calibration_status_view) + admin_app.get("/item_statistics", dependencies=[Depends(get_current_admin)])(item_statistics_view) + admin_app.get("/session_overview", dependencies=[Depends(get_current_admin)])(session_overview_view) - admin_app.get("/calibration_status", dependencies=[Depends(get_current_admin)])(calibration_link.get) - admin_app.get("/item_statistics", dependencies=[Depends(get_current_admin)])(item_stats_link.get) - admin_app.get("/session_overview", dependencies=[Depends(get_current_admin)])(session_overview_link.get) + # Preserve previously exposed broken list URLs and redirect them to the safe dashboard. + for legacy_path in ( + "/tryout/list", + "/item/list", + "/user/list", + "/session/list", + "/tryoutstats/list", + ): + admin_app.get(legacy_path, dependencies=[Depends(get_current_admin)])(dashboard_view) return admin_app