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

from typing import List

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

from app.database import get_db
from app.core.deps import require_roles
from app.models.user import User
from app.models.role import Role
from app.schemas.common import ApiResponse, PagedData
from app.schemas.role import RoleCreate, RoleUpdate, RoleOut
from app.schemas.user import UserOut
from app.services.role import RoleService


class AssignMenusBody(BaseModel):
    """与 bx-web / bx-mobile 一致：请求体使用 camelCase（menuIds）。"""

    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    menu_ids: List[int]


class AssignUsersBody(BaseModel):
    """与 bx-web / bx-mobile 一致：请求体使用 camelCase（userIds）。"""

    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    user_ids: List[int]

router = APIRouter(prefix="/api/roles", tags=["角色管理"])


def _raise_if_role_code_integrity_error(exc: IntegrityError) -> None:
    """将角色编码唯一约束冲突转为 400，避免未处理 IntegrityError 导致 500。"""
    orig = getattr(exc, "orig", None)
    msg = str(orig) if orig is not None else str(exc)
    if "uk_role_code" in msg or "role_code" in msg.lower():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="角色编码已存在，请更换其他编码",
        ) from exc
    raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail="保存失败：数据约束冲突",
    ) from exc


@router.get("/", response_model=ApiResponse[PagedData[RoleOut]])
def list_roles(
    page: int = 1,
    pageSize: int = 10,
    keyword: str = "",
    db: Session = Depends(get_db),
    _: User = Depends(require_roles("ADMIN")),
):
    filters = [Role.del_flag == 0]
    if keyword:
        filters.append(or_(Role.role_name.contains(keyword), Role.role_code.contains(keyword)))
    rows, total = RoleService.get_paged(db, page=page, page_size=pageSize, filters=filters)
    out_rows = [RoleOut.model_validate(r) for r in rows]
    paged_data = PagedData(rows=out_rows, total=total, page=page, page_size=pageSize)
    return ApiResponse(data=paged_data)


@router.get("/{role_id}", response_model=ApiResponse[RoleOut])
def get_role(role_id: int, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    role = RoleService.get_by_id(db, role_id)
    if not role:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="角色不存在")
    return ApiResponse(data=RoleOut.model_validate(role))


@router.post("/", response_model=ApiResponse[RoleOut])
def create_role(body: RoleCreate, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    data = body.model_dump(exclude_unset=True)
    code = (data.get("role_code") or "").strip()
    if code:
        data["role_code"] = code
        taken = db.query(Role).filter(Role.role_code == code, Role.del_flag == 0).first()
        if taken:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="角色编码已存在，请更换其他编码",
            )
    try:
        role = RoleService.create(db, data)
    except IntegrityError as e:
        _raise_if_role_code_integrity_error(e)
    return ApiResponse(data=RoleOut.model_validate(role))


@router.put("/{role_id}", response_model=ApiResponse[RoleOut])
def update_role(role_id: int, body: RoleUpdate, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    data = body.model_dump(exclude_unset=True)
    if "role_code" in data and data["role_code"] is not None:
        code = str(data["role_code"]).strip()
        data["role_code"] = code
        if code:
            taken = (
                db.query(Role)
                .filter(Role.role_code == code, Role.del_flag == 0, Role.id != role_id)
                .first()
            )
            if taken:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="角色编码已存在，请更换其他编码",
                )
    try:
        role = RoleService.update(db, role_id, data)
    except IntegrityError as e:
        _raise_if_role_code_integrity_error(e)
    if not role:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="角色不存在")
    return ApiResponse(data=RoleOut.model_validate(role))


@router.delete("/{role_id}", response_model=ApiResponse[None])
def delete_role(role_id: int, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    RoleService.soft_delete(db, role_id)
    return ApiResponse(data=None)


@router.put("/{role_id}/menus", response_model=ApiResponse[RoleOut])
def assign_menus(role_id: int, body: AssignMenusBody, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    role = RoleService.assign_menus(db, role_id, body.menu_ids)
    if not role:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="角色不存在")
    return ApiResponse(data=RoleOut.model_validate(role))


@router.get("/{role_id}/users", response_model=ApiResponse[List[UserOut]])
def get_role_users(role_id: int, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    users = RoleService.get_users_by_role(db, role_id)
    return ApiResponse(data=[UserOut.model_validate(u) for u in users])


@router.put("/{role_id}/users", response_model=ApiResponse[RoleOut])
def assign_users(role_id: int, body: AssignUsersBody, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    role = RoleService.assign_users(db, role_id, body.user_ids)
    if not role:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="角色不存在")
    return ApiResponse(data=RoleOut.model_validate(role))