ajyy/backend/app/api/auth.py

"""鉴权路由：login / logout / current-user / menus"""
from typing import Annotated

from fastapi import APIRouter, Depends, Request
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.deps import get_current_user
from app.core.response import error_response, success_response
from app.models.system import SysUser
from app.schemas.auth import LoginBody
from app.services import auth_service

router = APIRouter(prefix="/web/auth", tags=["auth"])


@router.post("/login")
def login(body: LoginBody, request: Request, db: Annotated[Session, Depends(get_db)]):
    ip = request.client.host if request.client else None
    try:
        data = auth_service.login(db, body.username, body.password, ip=ip)
    except auth_service.AuthFailure as exc:
        return JSONResponse(
            status_code=401,
            content=error_response(401, exc.message),
        )
    return success_response(data)


@router.post("/logout")
def logout(user: Annotated[SysUser, Depends(get_current_user)]):
    return success_response({"username": user.username})


@router.get("/current-user")
def current_user(
    user: Annotated[SysUser, Depends(get_current_user)],
    db: Annotated[Session, Depends(get_db)],
):
    try:
        data = auth_service.current_user_payload(db, user.id)
    except auth_service.AuthFailure as exc:
        return JSONResponse(
            status_code=401,
            content=error_response(401, exc.message),
        )
    return success_response(data)


@router.get("/menus")
def menus(
    user: Annotated[SysUser, Depends(get_current_user)],
    db: Annotated[Session, Depends(get_db)],
):
    return success_response(auth_service.menus_payload(db, user.role_id))