shanxia/bx/bx-backend/app/routers/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, ConfigDict
from pydantic.alias_generators import to_camel
from sqlalchemy.orm import Session

from app.database import get_db
from app.core.security import verify_password, hash_password, create_access_token
from app.core.deps import get_current_user
from app.models.user import User
from app.schemas.common import ApiResponse
from app.schemas.user import UserOut

router = APIRouter(prefix="/api/auth", tags=["认证"])


class LoginRequest(BaseModel):
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    username: str
    password: str


class RegisterRequest(BaseModel):
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    username: str
    password: str
    real_name: str


class TokenResponse(BaseModel):
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    access_token: str
    token_type: str = "bearer"


@router.post("/login", response_model=ApiResponse[TokenResponse])
def login(body: LoginRequest, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.username == body.username, User.del_flag == 0).first()
    if not user or not verify_password(body.password, user.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )
    if user.status != 1:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="用户已被禁用",
        )
    token = create_access_token(data={"sub": str(user.id)})
    return ApiResponse(data=TokenResponse(access_token=token))


@router.post("/register", response_model=ApiResponse[UserOut])
def register(body: RegisterRequest, db: Session = Depends(get_db)):
    existing = db.query(User).filter(User.username == body.username, User.del_flag == 0).first()
    if existing:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户名已存在",
        )
    from app.models.role import Role
    employee_role = db.query(Role).filter(Role.role_code == "EMPLOYEE", Role.del_flag == 0).first()
    user = User(
        username=body.username,
        password=hash_password(body.password),
        real_name=body.real_name,
        status=1,
    )
    if employee_role:
        user.roles.append(employee_role)
    db.add(user)
    db.commit()
    db.refresh(user)
    return ApiResponse(data=UserOut.model_validate(user))


@router.get("/me", response_model=ApiResponse[UserOut])
def get_current_user_info(current_user: User = Depends(get_current_user)):
    return ApiResponse(data=UserOut.model_validate(current_user))