"""Admin authentication: login, me, change password.""" from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.db import get_db from app.models import AdminUser from app.schemas import ChangePasswordIn, LoginIn, MeOut, TokenOut from app.security import ( create_token, get_current_admin, hash_password, verify_password, ) router = APIRouter(prefix="/auth") @router.post("/login", response_model=TokenOut) def login(body: LoginIn, db: Session = Depends(get_db)) -> TokenOut: user = db.query(AdminUser).filter(AdminUser.username == body.username).first() if not user or not verify_password(body.password, user.password_hash): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid username or password", ) return TokenOut(access_token=create_token(user.username), username=user.username) @router.get("/me", response_model=MeOut) def me(admin: AdminUser = Depends(get_current_admin)) -> MeOut: return MeOut(username=admin.username) @router.post("/change-password", response_model=MeOut) def change_password( body: ChangePasswordIn, admin: AdminUser = Depends(get_current_admin), db: Session = Depends(get_db), ) -> MeOut: if not verify_password(body.current_password, admin.password_hash): raise HTTPException(status_code=400, detail="Current password is incorrect") if len(body.new_password) < 6: raise HTTPException(status_code=400, detail="New password must be at least 6 characters") admin.password_hash = hash_password(body.new_password) db.commit() return MeOut(username=admin.username)