from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy.orm import Session from app.auth.jwt_auth import CurrentUser, create_access_token, get_current_user from app.config import get_settings from app.database import get_db from app.services.user_auth import authenticate_user router = APIRouter(prefix="/auth", tags=["auth"]) class LoginRequest(BaseModel): username: str password: str class TokenResponse(BaseModel): access_token: str token_type: str = "bearer" username: str role: str class MeResponse(BaseModel): username: str role: str @router.post("/login", response_model=TokenResponse) def login(body: LoginRequest, db: Session = Depends(get_db)) -> TokenResponse: settings = get_settings() if not settings.sac_admin_password and not _has_any_user(db): raise HTTPException( status_code=503, detail="SAC UI users are not configured (run alembic upgrade and set SAC_ADMIN_PASSWORD for bootstrap)", ) user = authenticate_user(db, body.username, body.password) if user is None: raise HTTPException(status_code=401, detail="Invalid username or password") token = create_access_token(user.username, user.role) return TokenResponse( access_token=token, username=user.username, role=user.role, ) @router.get("/me", response_model=MeResponse) def me(current_user: CurrentUser = Depends(get_current_user)) -> MeResponse: return MeResponse(username=current_user.username, role=current_user.role) def _has_any_user(db: Session) -> bool: from sqlalchemy import func, select from app.models.user import User try: count = db.scalar(select(func.count()).select_from(User)) or 0 return count > 0 except Exception: db.rollback() return False