"""
FastAPI Admin configuration for IRT Bank Soal system.
Provides admin panel for managing tryouts, items, sessions, users, and tryout stats.
Includes custom actions for calibration, AI generation toggle, and normalization reset.
"""
import secrets
import uuid
from dataclasses import dataclass
from typing import Any, Dict, Optional
import aioredis
from fastapi import Depends, Form, HTTPException, Request
from fastapi_admin import constants
from fastapi_admin.app import app as admin_app
from fastapi_admin.depends import get_current_admin
from fastapi_admin.providers import Provider
from fastapi_admin.resources import (
Field,
Link,
Model,
)
from fastapi_admin.widgets import displays, inputs
from sqlalchemy import select
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import HTMLResponse, RedirectResponse
from starlette.status import HTTP_303_SEE_OTHER, HTTP_401_UNAUTHORIZED
from app.core.config import get_settings
from app.database import get_db
from app.models import Item, Session, Tryout, TryoutStats, User
settings = get_settings()
# =============================================================================
# Authentication Provider
# =============================================================================
@dataclass
class AdminPrincipal:
"""Minimal admin user object expected by fastapi-admin templates."""
pk: str
username: str
avatar: str = ""
class EnvCredentialProvider(Provider):
"""
FastAPI-Admin provider backed by env credentials and Redis session tokens.
Compatible with fastapi-admin 1.0.x provider API without requiring
Tortoise admin models.
"""
# fastapi-admin login templates expect `request.app.login_provider`
# to exist for resolving login/logout URLs.
name = "login_provider"
access_token = "access_token"
def __init__(
self,
username: str,
password: str,
login_path: str = "/login",
logout_path: str = "/logout",
login_title: str = "Admin Login",
login_logo_url: str | None = None,
expire_seconds: int = 3600,
) -> None:
self._username = username
self._password = password
self.login_path = login_path
self.logout_path = logout_path
self.login_title = login_title
self.login_logo_url = login_logo_url
self.expire_seconds = expire_seconds
async def register(self, app: "FastAPIAdmin") -> None:
await super().register(app)
# Keep explicit assignment for compatibility across fastapi-admin versions.
app.login_provider = self
app.get("/")(self.index_view)
app.get(self.login_path)(self.login_view)
app.post(self.login_path)(self.login)
app.get(self.logout_path)(self.logout)
app.get("/password")(self.password_view)
app.post("/password")(self.password)
app.add_middleware(BaseHTTPMiddleware, dispatch=self.authenticate)
def _render_auth_page(
self,
request: Request,
title: str,
subtitle: str,
body: str,
status_code: int = 200,
) -> HTMLResponse:
remember_me_checked = "checked" if request.cookies.get("remember_me") == "on" else ""
html = f"""
{title}
{title}
{subtitle}
{body.replace("__REMEMBER_ME_CHECKED__", remember_me_checked)}
"""
return HTMLResponse(html, status_code=status_code)
@staticmethod
def _admin_home(request: Request) -> str:
return request.app.admin_path.rstrip("/") + "/dashboard"
@staticmethod
def _login_url(request: Request) -> str:
return request.app.admin_path.rstrip("/") + "/login"
async def index_view(self, request: Request):
# fastapi-admin has no default "/" page in this setup.
if getattr(request.state, "admin", None):
return RedirectResponse(url=self._admin_home(request), status_code=HTTP_303_SEE_OTHER)
return RedirectResponse(url=self._login_url(request), status_code=HTTP_303_SEE_OTHER)
async def login_view(self, request: Request):
body = f"""
Direct environment-backed admin access.
"""
return self._render_auth_page(
request=request,
title=self.login_title,
subtitle="Use the configured admin credentials to access the dashboard.",
body=body,
)
async def login(
self,
request: Request,
username: str = Form(...),
password: str = Form(...),
remember_me: Optional[str] = Form(None),
):
if not (
secrets.compare_digest(username, self._username)
and secrets.compare_digest(password, self._password)
):
body = f"""
Invalid username or password.
"""
return self._render_auth_page(
request=request,
title=self.login_title,
subtitle="Use the configured admin credentials to access the dashboard.",
body=body,
status_code=HTTP_401_UNAUTHORIZED,
)
response = RedirectResponse(url=self._admin_home(request), status_code=HTTP_303_SEE_OTHER)
expire = self.expire_seconds
if remember_me == "on":
expire = max(self.expire_seconds, 3600 * 24 * 30)
response.set_cookie("remember_me", "on")
else:
response.delete_cookie("remember_me")
token = uuid.uuid4().hex
response.set_cookie(
self.access_token,
token,
expires=expire,
path=request.app.admin_path,
httponly=True,
)
await request.app.redis.set(constants.LOGIN_USER.format(token=token), self._username, ex=expire)
return response
async def authenticate(self, request: Request, call_next: RequestResponseEndpoint):
token = request.cookies.get(self.access_token)
path = request.scope["path"]
admin = None
if token:
key = constants.LOGIN_USER.format(token=token)
username = await request.app.redis.get(key)
if username:
admin = AdminPrincipal(pk=str(username), username=str(username))
request.state.admin = admin
if path.endswith(self.login_path) and admin:
return RedirectResponse(url=self._admin_home(request), status_code=HTTP_303_SEE_OTHER)
return await call_next(request)
async def logout(self, request: Request):
response = RedirectResponse(
url=request.app.admin_path + self.login_path,
status_code=HTTP_303_SEE_OTHER,
)
token = request.cookies.get(self.access_token)
if token:
await request.app.redis.delete(constants.LOGIN_USER.format(token=token))
response.delete_cookie(self.access_token, path=request.app.admin_path)
return response
async def password_view(
self,
request: Request,
admin: AdminPrincipal = Depends(get_current_admin),
):
body = f"""
Signed in as {admin.username}.
Password changes are disabled in the UI for this deployment.
Update ADMIN_PASSWORD in the server environment, then restart the app.
Session expiry is currently set to {self.expire_seconds} seconds.
Back to dashboard
"""
return self._render_auth_page(
request=request,
title="Password Management",
subtitle="Runtime password rotation is intentionally disabled.",
body=body,
)
async def password(
self,
request: Request,
old_password: str = Form(...),
new_password: str = Form(...),
re_new_password: str = Form(...),
admin: AdminPrincipal = Depends(get_current_admin),
):
_ = (old_password, new_password, re_new_password, admin)
body = f"""
Password rotation via UI is disabled.
Update ADMIN_PASSWORD in the server environment, then restart the app.
Back to dashboard
"""
return self._render_auth_page(
request=request,
title="Password Management",
subtitle="Runtime password rotation is intentionally disabled.",
body=body,
status_code=400,
)
# =============================================================================
# Admin Model Resources
# =============================================================================
class TryoutResource(Model):
"""
Admin resource for Tryout model.
Displays tryout configuration and provides calibration and AI generation actions.
"""
label = "Tryouts"
model = Tryout
page_size = 20
# Fields to display
fields = [
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
Field(name="name", label="Name", input_=inputs.Input(), display=displays.Display()),
Field(
name="description",
label="Description",
input_=inputs.TextArea(),
display=displays.Display(),
),
Field(
name="scoring_mode",
label="Scoring Mode",
input_=inputs.Select(default="ctt"),
display=displays.Display(),
),
Field(
name="selection_mode",
label="Selection Mode",
input_=inputs.Select(default="fixed"),
display=displays.Display(),
),
Field(
name="normalization_mode",
label="Normalization Mode",
input_=inputs.Select(default="static"),
display=displays.Display(),
),
Field(
name="min_sample_for_dynamic",
label="Min Sample for Dynamic",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="static_rataan",
label="Static Mean (Rataan)",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="static_sb",
label="Static Std Dev (SB)",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="ai_generation_enabled",
label="Enable AI Generation",
input_=inputs.Switch(),
display=displays.Boolean(true_text="Enabled", false_text="Disabled"),
),
Field(
name="hybrid_transition_slot",
label="Hybrid Transition Slot",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="min_calibration_sample",
label="Min Calibration Sample",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="theta_estimation_method",
label="Theta Estimation Method",
input_=inputs.Select(default="mle"),
display=displays.Display(),
),
Field(
name="fallback_to_ctt_on_error",
label="Fallback to CTT on Error",
input_=inputs.Switch(),
display=displays.Boolean(true_text="Yes", false_text="No"),
),
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
]
class ItemResource(Model):
"""
Admin resource for Item model.
Displays items with CTT and IRT parameters, and calibration status.
"""
label = "Items"
model = Item
page_size = 50
# Fields to display
fields = [
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
Field(name="slot", label="Slot", input_=inputs.Input(type="number"), display=displays.Display()),
Field(
name="level",
label="Difficulty Level",
input_=inputs.Select(default="sedang"),
display=displays.Display(),
),
Field(
name="stem",
label="Question Stem",
input_=inputs.TextArea(),
display=displays.Display(),
),
Field(name="options", label="Options", input_=inputs.Json(), display=displays.Json()),
Field(name="correct_answer", label="Correct Answer", input_=inputs.Input(), display=displays.Display()),
Field(
name="explanation",
label="Explanation",
input_=inputs.TextArea(),
display=displays.Display(),
),
Field(
name="ctt_p",
label="CTT p-value",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="ctt_bobot",
label="CTT Bobot",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="ctt_category",
label="CTT Category",
input_=inputs.Select(),
display=displays.Display(),
),
Field(
name="irt_b",
label="IRT b-parameter",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="irt_se",
label="IRT SE",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="calibrated",
label="Calibrated",
input_=inputs.Switch(),
display=displays.Boolean(true_text="Yes", false_text="No"),
),
Field(
name="calibration_sample_size",
label="Calibration Sample Size",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="generated_by",
label="Generated By",
input_=inputs.Select(default="manual"),
display=displays.Display(),
),
Field(name="ai_model", label="AI Model", input_=inputs.Input(), display=displays.Display()),
Field(
name="basis_item_id",
label="Basis Item ID",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
]
class UserResource(Model):
"""
Admin resource for User model.
Displays WordPress users and their tryout sessions.
"""
label = "Users"
model = User
page_size = 50
# Fields
fields = [
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
Field(name="wp_user_id", label="WordPress User ID", input_=inputs.Input(), display=displays.Display()),
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
]
class SessionResource(Model):
"""
Admin resource for Session model.
Displays tryout sessions with scoring results (NM, NN, theta).
"""
label = "Sessions"
model = Session
page_size = 50
# Fields
fields = [
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
Field(name="session_id", label="Session ID", input_=inputs.Input(), display=displays.Display()),
Field(name="wp_user_id", label="WordPress User ID", input_=inputs.Input(), display=displays.Display()),
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
Field(name="start_time", label="Start Time", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
Field(name="end_time", label="End Time", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
Field(
name="is_completed",
label="Completed",
input_=inputs.Switch(),
display=displays.Boolean(true_text="Yes", false_text="No"),
),
Field(
name="scoring_mode_used",
label="Scoring Mode Used",
input_=inputs.Select(),
display=displays.Display(),
),
Field(name="total_benar", label="Total Benar", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="total_bobot_earned", label="Total Bobot Earned", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="NM", label="NM Score", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="NN", label="NN Score", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="theta", label="Theta", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="theta_se", label="Theta SE", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="rataan_used", label="Rataan Used", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="sb_used", label="SB Used", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
]
class TryoutStatsResource(Model):
"""
Admin resource for TryoutStats model.
Displays tryout-level statistics and provides normalization reset action.
"""
label = "Tryout Stats"
model = TryoutStats
page_size = 20
# Fields
fields = [
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
Field(
name="participant_count",
label="Participant Count",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="total_nm_sum",
label="Total NM Sum",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(
name="total_nm_sq_sum",
label="Total NM Squared Sum",
input_=inputs.Input(type="number"),
display=displays.Display(),
),
Field(name="rataan", label="Rataan", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="sb", label="SB", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="min_nm", label="Min NM", input_=inputs.Input(type="number"), display=displays.Display()),
Field(name="max_nm", label="Max NM", input_=inputs.Input(type="number"), display=displays.Display()),
Field(
name="last_calculated",
label="Last Calculated",
input_=inputs.DateTime(),
display=displays.DatetimeDisplay(),
),
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
]
# =============================================================================
# Custom Dashboard Views
# =============================================================================
class CalibrationDashboardLink(Link):
"""
Link to calibration status dashboard.
Displays calibration percentage and items awaiting calibration.
"""
label = "Calibration Status"
icon = "fas fa-chart-line"
url = "/admin/calibration_status"
async def get(self, request: Request) -> Dict[str, Any]:
"""Get calibration status for all tryouts."""
# Get all tryouts
db_gen = get_db()
db = await db_gen.__anext__()
try:
result = await db.execute(
select(
Tryout.id,
Tryout.tryout_id,
Tryout.name,
)
)
tryouts = result.all()
calibration_data = []
for tryout_id, tryout_str, name in tryouts:
# Get calibration status
from app.services.irt_calibration import get_calibration_status
status = await get_calibration_status(tryout_str, 1, db)
calibration_data.append({
"tryout_id": tryout_str,
"name": name,
"total_items": status["total_items"],
"calibrated_items": status["calibrated_items"],
"calibration_percentage": status["calibration_percentage"],
"ready_for_irt": status["ready_for_irt"],
})
return {
"status": "success",
"data": calibration_data,
}
finally:
await db_gen.aclose()
class ItemStatisticsLink(Link):
"""
Link to item statistics view.
Displays items grouped by difficulty level with calibration status.
"""
label = "Item Statistics"
icon = "fas fa-chart-bar"
url = "/admin/item_statistics"
async def get(self, request: Request) -> Dict[str, Any]:
"""Get item statistics grouped by difficulty level."""
db_gen = get_db()
db = await db_gen.__anext__()
try:
# Get items grouped by level
result = await db.execute(
select(
Item.level,
)
.distinct()
)
levels = result.scalars().all()
stats = []
for level in levels:
# Get items for this level
item_result = await db.execute(
select(Item)
.where(Item.level == level)
.order_by(Item.slot)
.limit(10)
)
items = item_result.scalars().all()
# Calculate average correctness rate
total_responses = sum(item.calibration_sample_size for item in items)
calibrated_count = sum(1 for item in items if item.calibrated)
level_stats = {
"level": level,
"total_items": len(items),
"calibrated_items": calibrated_count,
"calibration_percentage": (calibrated_count / len(items) * 100) if len(items) > 0 else 0,
"total_responses": total_responses,
"avg_correctness": sum(item.ctt_p or 0 for item in items) / len(items) if len(items) > 0 else 0,
"items": [
{
"id": item.id,
"slot": item.slot,
"calibrated": item.calibrated,
"ctt_p": item.ctt_p,
"irt_b": item.irt_b,
"calibration_sample_size": item.calibration_sample_size,
}
for item in items
],
}
stats.append(level_stats)
return {
"status": "success",
"data": stats,
}
finally:
await db_gen.aclose()
class SessionOverviewLink(Link):
"""
Link to session overview view.
Displays sessions with scores (NM, NN, theta) and completion status.
"""
label = "Session Overview"
icon = "fas fa-users"
url = "/admin/session_overview"
async def get(self, request: Request) -> Dict[str, Any]:
"""Get session overview with filters."""
db_gen = get_db()
db = await db_gen.__anext__()
try:
# Get recent sessions
result = await db.execute(
select(Session)
.order_by(Session.created_at.desc())
.limit(50)
)
sessions = result.scalars().all()
session_data = [
{
"session_id": session.session_id,
"wp_user_id": session.wp_user_id,
"tryout_id": session.tryout_id,
"is_completed": session.is_completed,
"scoring_mode_used": session.scoring_mode_used,
"total_benar": session.total_benar,
"NM": session.NM,
"NN": session.NN,
"theta": session.theta,
"theta_se": session.theta_se,
"start_time": session.start_time.isoformat() if session.start_time else None,
"end_time": session.end_time.isoformat() if session.end_time else None,
}
for session in sessions
]
return {
"status": "success",
"data": session_data,
}
finally:
await db_gen.aclose()
# =============================================================================
# Initialize FastAPI Admin
# =============================================================================
def _render_admin_page(title: str, page_title: str, body: str) -> HTMLResponse:
html = f"""
{title}
"""
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""
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.
Returns:
FastAPI app with admin panel
"""
# Configure admin app
# admin_app.settings.logo_url = "/static/logo.png"
# admin_app.settings.site_title = "IRT Bank Soal Admin"
# admin_app.settings.site_description = "Admin Panel for Adaptive Question Bank System"
# Register authentication provider
# NOTE: fastapi-admin 1.0.4 requires provider registration via app.configure(...).
# Keep provider implementation here for future integration during startup configure.
# Reset singleton registries so stale state cannot survive restarts.
admin_app.resources = []
admin_app.model_resources = {}
_prune_incompatible_admin_routes()
admin_app.register(CalibrationDashboardLink)
admin_app.register(ItemStatisticsLink)
admin_app.register(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)
# 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
_admin_configured = False
_admin_redis = None
async def configure_admin_app() -> None:
"""Configure fastapi-admin runtime (redis + auth provider)."""
global _admin_configured, _admin_redis
if _admin_configured:
return
if not settings.ADMIN_USERNAME or not settings.ADMIN_PASSWORD:
raise RuntimeError(
"ENABLE_ADMIN=true requires ADMIN_USERNAME and ADMIN_PASSWORD to be set."
)
_admin_redis = aioredis.from_url(
settings.REDIS_URL,
encoding="utf-8",
decode_responses=True,
)
provider = EnvCredentialProvider(
username=settings.ADMIN_USERNAME,
password=settings.ADMIN_PASSWORD,
login_title="IRT Bank Soal Admin",
expire_seconds=settings.ADMIN_SESSION_EXPIRE_SECONDS,
)
await admin_app.configure(
redis=_admin_redis,
admin_path="/admin",
providers=[provider],
)
_admin_configured = True
async def shutdown_admin_app() -> None:
"""Close admin redis client cleanly."""
global _admin_redis
if _admin_redis is None:
return
try:
await _admin_redis.close()
finally:
_admin_redis = None
# Export admin app for mounting in main.py
admin = create_admin_app()