from collections.abc import Callable from typing import Any from fastapi import Depends, HTTPException, Request, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from starlette.middleware.sessions import SessionMiddleware from app.config import settings from app.database import get_db from app.models import AdminUser from app.security import verify_password def install_session_middleware(app: Any) -> None: app.add_middleware( SessionMiddleware, secret_key=settings.secret_key, session_cookie=settings.session_cookie, max_age=settings.session_max_age, same_site="lax", https_only=settings.session_https_only, ) async def authenticate_admin(session: AsyncSession, username: str, password: str) -> AdminUser | None: result = await session.execute(select(AdminUser).where(AdminUser.username == username)) user = result.scalar_one_or_none() if not user or not user.is_active: return None if not verify_password(password, user.password_hash): return None return user async def get_current_admin(request: Request, db: AsyncSession = Depends(get_db)) -> AdminUser: admin_id = request.session.get("admin_id") if not admin_id: raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/admin/login"}) user = await db.get(AdminUser, admin_id) if not user or not user.is_active: request.session.clear() raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/admin/login"}) return user def login_required(view: Callable): return view