592 lines
24 KiB
Python
592 lines
24 KiB
Python
"""
|
||
系统管理路由模块
|
||
|
||
提供用户、角色、菜单的完整CRUD管理接口，包括：
|
||
- 用户管理：列表查询、创建、更新、重置密码、启用/禁用
|
||
- 角色管理：列表查询、创建、更新、菜单权限分配
|
||
- 菜单管理：列表查询、创建、更新
|
||
|
||
URL 前缀：/api/system
|
||
权限要求：admin 角色，各操作需对应细粒度权限
|
||
"""
|
||
|
||
from fastapi import APIRouter, Body, Depends, Query
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.api.deps import get_system_service, require_permissions, require_roles
|
||
from backend.app.core.error_codes import ErrorCode
|
||
from backend.app.core.exceptions import AppException
|
||
from backend.app.db import get_db_session
|
||
from backend.app.models.system import Role, User
|
||
from backend.app.schemas.common import success_payload
|
||
from backend.app.schemas.system import (
|
||
AssignRoleMenusRequest,
|
||
ChangeUserStatusRequest,
|
||
CreateMenuRequest,
|
||
CreateRoleRequest,
|
||
CreateUserRequest,
|
||
ResetPasswordRequest,
|
||
UpdateMenuRequest,
|
||
UpdateRoleRequest,
|
||
UpdateUserRequest,
|
||
)
|
||
from backend.app.services.system_service import SystemService
|
||
|
||
router = APIRouter(prefix="/api/system", tags=["system"])
|
||
|
||
|
||
@router.get("/users")
|
||
def list_users(
|
||
username: str | None = Query(default=None), # 按用户名模糊搜索
|
||
real_name: str | None = Query(default=None), # 按真实姓名模糊搜索
|
||
mobile: str | None = Query(default=None), # 按手机号搜索
|
||
role_id: int | None = Query(default=None), # 按角色ID筛选
|
||
status: int | None = Query(default=None), # 按状态筛选（0禁用/1启用）
|
||
page_no: int = Query(default=1), # 页码，从1开始
|
||
page_size: int = Query(default=20), # 每页条数
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:user:list")), # 用户列表查看权限
|
||
) -> dict:
|
||
"""
|
||
获取用户列表
|
||
|
||
支持按用户名、真实姓名、手机号、角色ID、状态进行筛选，分页返回用户列表。
|
||
|
||
请求参数：
|
||
username: 用户名（可选，模糊匹配）
|
||
real_name: 真实姓名（可选，模糊匹配）
|
||
mobile: 手机号（可选）
|
||
role_id: 角色ID（可选）
|
||
status: 状态（可选，0禁用/1启用）
|
||
page_no: 页码（默认1）
|
||
page_size: 每页条数（默认20）
|
||
|
||
返回值：
|
||
分页的用户列表数据
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:user:list 权限
|
||
"""
|
||
result = system_service.list_users(
|
||
session,
|
||
{
|
||
"username": username,
|
||
"real_name": real_name,
|
||
"mobile": mobile,
|
||
"role_id": role_id,
|
||
"status": status,
|
||
},
|
||
)
|
||
start = max(page_no - 1, 0) * page_size
|
||
result["list"] = result["list"][start : start + page_size]
|
||
result["page_no"] = page_no
|
||
result["page_size"] = page_size
|
||
return success_payload(result)
|
||
|
||
|
||
@router.post("/users")
|
||
def create_user(
|
||
payload: CreateUserRequest, # 创建用户请求体，包含用户名、密码、角色等信息
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:user:create")), # 用户创建权限
|
||
) -> dict:
|
||
"""
|
||
创建新用户
|
||
|
||
创建一个新的系统用户账号，并分配初始角色。
|
||
|
||
请求参数：
|
||
payload: CreateUserRequest 请求体，包含用户名、密码、真实姓名、手机号、角色ID等
|
||
|
||
返回值：
|
||
新创建的用户信息
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:user:create 权限
|
||
"""
|
||
return success_payload(system_service.create_user(payload.model_dump(), session))
|
||
|
||
|
||
@router.put("/users/{user_id}")
|
||
def update_user(
|
||
user_id: int, # 用户ID
|
||
payload: UpdateUserRequest, # 更新用户请求体，包含要修改的字段
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:user:update")), # 用户更新权限
|
||
) -> dict:
|
||
"""
|
||
更新用户信息
|
||
|
||
修改指定用户的基本信息（如真实姓名、手机号、角色等）。
|
||
|
||
请求参数：
|
||
user_id: 路径参数，用户ID
|
||
payload: UpdateUserRequest 请求体，包含要更新的字段
|
||
|
||
返回值：
|
||
更新后的用户信息
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:user:update 权限
|
||
"""
|
||
return success_payload(system_service.update_user(user_id, payload.model_dump(), session))
|
||
|
||
|
||
@router.post("/users/{user_id}/reset-password")
|
||
def reset_password(
|
||
user_id: int, # 用户ID
|
||
payload: ResetPasswordRequest, # 重置密码请求体，包含新密码
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:user:reset-password")), # 密码重置权限
|
||
) -> dict:
|
||
"""
|
||
重置用户密码
|
||
|
||
管理员强制重置指定用户的登录密码。
|
||
|
||
请求参数：
|
||
user_id: 路径参数，用户ID
|
||
payload: ResetPasswordRequest 请求体，包含 new_password（新密码明文）
|
||
|
||
返回值：
|
||
操作结果
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:user:reset-password 权限
|
||
"""
|
||
return success_payload(system_service.reset_password(user_id, payload.new_password, session))
|
||
|
||
|
||
@router.post("/users/{user_id}/status")
|
||
def change_user_status(
|
||
user_id: int, # 用户ID
|
||
payload: ChangeUserStatusRequest, # 变更状态请求体，包含目标状态值
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
) -> dict:
|
||
"""
|
||
启用/禁用用户
|
||
|
||
切换指定用户的账号状态（启用或禁用）。注意：无法禁用当前登录的管理员自己。
|
||
|
||
请求参数：
|
||
user_id: 路径参数，用户ID
|
||
payload: ChangeUserStatusRequest 请求体，包含 status（0禁用/1启用）
|
||
|
||
返回值：
|
||
操作结果
|
||
|
||
权限要求：admin 角色，需 system:view 权限
|
||
"""
|
||
return success_payload(system_service.change_user_status(user_id, payload.status, session, current_user))
|
||
|
||
|
||
@router.get("/roles")
|
||
def list_roles(
|
||
role_name: str | None = Query(default=None), # 按角色名称模糊搜索
|
||
role_code: str | None = Query(default=None), # 按角色编码精确搜索
|
||
status: int | None = Query(default=None), # 按状态筛选（0禁用/1启用）
|
||
page_no: int = Query(default=1), # 页码，从1开始
|
||
page_size: int = Query(default=20), # 每页条数
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:role:list")), # 角色列表查看权限
|
||
) -> dict:
|
||
"""
|
||
获取角色列表
|
||
|
||
支持按角色名称、编码、状态进行筛选，分页返回角色列表。
|
||
|
||
请求参数：
|
||
role_name: 角色名称（可选，模糊匹配）
|
||
role_code: 角色编码（可选，精确匹配）
|
||
status: 状态（可选，0禁用/1启用）
|
||
page_no: 页码（默认1）
|
||
page_size: 每页条数（默认20）
|
||
|
||
返回值：
|
||
分页的角色列表数据
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:role:list 权限
|
||
"""
|
||
result = system_service.list_roles(
|
||
session,
|
||
{"role_name": role_name, "role_code": role_code, "status": status},
|
||
)
|
||
start = max(page_no - 1, 0) * page_size
|
||
result["list"] = result["list"][start : start + page_size]
|
||
result["page_no"] = page_no
|
||
result["page_size"] = page_size
|
||
return success_payload(result)
|
||
|
||
|
||
@router.post("/roles")
|
||
def create_role(
|
||
payload: CreateRoleRequest, # 创建角色请求体，包含角色名称、编码、描述等
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:role:create")), # 角色创建权限
|
||
) -> dict:
|
||
"""
|
||
创建新角色
|
||
|
||
创建一个新的系统角色，创建后可通过菜单分配接口为其分配权限。
|
||
|
||
请求参数：
|
||
payload: CreateRoleRequest 请求体，包含角色名称、编码、描述等
|
||
|
||
返回值：
|
||
新创建的角色信息
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:role:create 权限
|
||
"""
|
||
return success_payload(system_service.create_role(payload.model_dump(), session))
|
||
|
||
|
||
@router.put("/roles/{role_id}")
|
||
def update_role(
|
||
role_id: int, # 角色ID
|
||
payload: UpdateRoleRequest, # 更新角色请求体，包含要修改的字段
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:role:update")), # 角色更新权限
|
||
) -> dict:
|
||
"""
|
||
更新角色信息
|
||
|
||
修改指定角色的基本信息（如名称、描述、状态等）。
|
||
|
||
请求参数：
|
||
role_id: 路径参数，角色ID
|
||
payload: UpdateRoleRequest 请求体，包含要更新的字段
|
||
|
||
返回值：
|
||
更新后的角色信息
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:role:update 权限
|
||
"""
|
||
return success_payload(system_service.update_role(role_id, payload.model_dump(), session))
|
||
|
||
|
||
@router.get("/roles/{role_id}/menus")
|
||
def get_role_menus(
|
||
role_id: int, # 角色ID
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:role:list")), # 角色列表查看权限
|
||
) -> dict:
|
||
"""
|
||
获取角色的菜单权限分配
|
||
|
||
查询指定角色已分配的菜单列表，用于角色权限编辑页面展示。
|
||
|
||
请求参数：
|
||
role_id: 路径参数，角色ID
|
||
|
||
返回值：
|
||
该角色已分配的菜单ID列表及菜单树结构
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:role:list 权限
|
||
"""
|
||
return success_payload(system_service.get_role_menu_assignment(role_id, session))
|
||
|
||
|
||
@router.put("/roles/{role_id}/menus")
|
||
def assign_role_menus(
|
||
role_id: int, # 角色ID
|
||
payload: AssignRoleMenusRequest, # 分配菜单请求体，包含菜单ID列表
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:role:assign-menus")), # 角色菜单分配权限
|
||
) -> dict:
|
||
"""
|
||
为角色分配菜单权限
|
||
|
||
将指定的菜单列表分配给角色，采用全量替换方式（覆盖已有分配）。
|
||
|
||
请求参数：
|
||
role_id: 路径参数，角色ID
|
||
payload: AssignRoleMenusRequest 请求体，包含 menu_ids（菜单ID列表）
|
||
|
||
返回值：
|
||
操作结果
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:role:assign-menus 权限
|
||
"""
|
||
return success_payload(system_service.assign_role_menus(role_id, payload.model_dump(), session))
|
||
|
||
|
||
@router.get("/menus")
|
||
def list_menus(
|
||
status: int | None = Query(default=None), # 按状态筛选（0禁用/1启用）
|
||
menu_type: str | None = Query(default=None), # 按菜单类型筛选（如 directory/menu/button）
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:menu:list")), # 菜单列表查看权限
|
||
) -> dict:
|
||
"""
|
||
获取菜单列表
|
||
|
||
返回系统菜单树结构，支持按状态和菜单类型筛选。
|
||
|
||
请求参数：
|
||
status: 状态（可选，0禁用/1启用）
|
||
menu_type: 菜单类型（可选，directory/menu/button）
|
||
|
||
返回值：
|
||
菜单树结构列表
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:menu:list 权限
|
||
"""
|
||
return success_payload(
|
||
system_service.list_menus(
|
||
session,
|
||
{"status": status, "menu_type": menu_type},
|
||
)
|
||
)
|
||
|
||
|
||
@router.post("/menus")
|
||
def create_menu(
|
||
payload: CreateMenuRequest, # 创建菜单请求体，包含菜单名称、类型、路径、父菜单ID等
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:menu:create")), # 菜单创建权限
|
||
) -> dict:
|
||
"""
|
||
创建新菜单
|
||
|
||
创建一个新的系统菜单项，可指定父菜单实现多级菜单结构。
|
||
|
||
请求参数：
|
||
payload: CreateMenuRequest 请求体，包含菜单名称、类型、路由路径、图标、父菜单ID等
|
||
|
||
返回值：
|
||
新创建的菜单信息
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:menu:create 权限
|
||
"""
|
||
return success_payload(system_service.create_menu(payload.model_dump(), session))
|
||
|
||
|
||
@router.put("/menus/{menu_id}")
|
||
def update_menu(
|
||
menu_id: int, # 菜单ID
|
||
payload: UpdateMenuRequest, # 更新菜单请求体，包含要修改的字段
|
||
system_service: SystemService = Depends(get_system_service), # 系统管理服务实例
|
||
session: Session = Depends(get_db_session), # 数据库会话
|
||
current_user: dict = Depends(require_roles("admin")), # 当前登录用户，仅admin可访问
|
||
_system_user: dict = Depends(require_permissions("system:view")), # 系统管理查看权限
|
||
_permission_user: dict = Depends(require_permissions("system:menu:update")), # 菜单更新权限
|
||
) -> dict:
|
||
"""
|
||
更新菜单信息
|
||
|
||
修改指定菜单的基本信息（如名称、路径、排序、状态等）。
|
||
|
||
请求参数：
|
||
menu_id: 路径参数，菜单ID
|
||
payload: UpdateMenuRequest 请求体，包含要更新的字段
|
||
|
||
返回值：
|
||
更新后的菜单信息
|
||
|
||
权限要求：admin 角色，需 system:view 和 system:menu:update 权限
|
||
"""
|
||
return success_payload(system_service.update_menu(menu_id, payload.model_dump(), session))
|
||
|
||
|
||
@router.delete("/users/{user_id}")
|
||
def delete_user(
|
||
user_id: int,
|
||
session: Session = Depends(get_db_session),
|
||
current_user: dict = Depends(require_roles("admin")),
|
||
_permission_user: dict = Depends(require_permissions("system:user:update")),
|
||
) -> dict:
|
||
"""删除用户（物理删除）"""
|
||
user = session.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
||
if not user:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="用户不存在", status_code=404)
|
||
if user.username == "admin":
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="不能删除管理员账号", status_code=400)
|
||
session.delete(user)
|
||
session.commit()
|
||
return success_payload({"deleted": True})
|
||
|
||
|
||
@router.delete("/users/{user_id}/permanent")
|
||
def permanent_delete_user(
|
||
user_id: int,
|
||
session: Session = Depends(get_db_session),
|
||
current_user: dict = Depends(require_roles("admin")),
|
||
_permission_user: dict = Depends(require_permissions("system:user:update")),
|
||
) -> dict:
|
||
"""永久删除用户"""
|
||
user = session.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
||
if not user:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="用户不存在", status_code=404)
|
||
if user.username == "admin":
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="不能删除管理员账号", status_code=400)
|
||
session.delete(user)
|
||
session.commit()
|
||
return success_payload({"deleted": True})
|
||
|
||
|
||
@router.delete("/roles/{role_id}")
|
||
def delete_role(
|
||
role_id: int,
|
||
session: Session = Depends(get_db_session),
|
||
current_user: dict = Depends(require_roles("admin")),
|
||
_permission_user: dict = Depends(require_permissions("system:role:update")),
|
||
) -> dict:
|
||
"""删除角色（物理删除）"""
|
||
role = session.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
|
||
if not role:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="角色不存在", status_code=404)
|
||
if role.role_code == "admin":
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="不能删除管理员角色", status_code=400)
|
||
count = session.execute(
|
||
select(func.count(User.id)).where(User.role_id == role_id, User.status == 1)
|
||
).scalar()
|
||
if count > 0:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="该角色下还有用户，无法删除", status_code=400)
|
||
session.delete(role)
|
||
session.commit()
|
||
return success_payload({"deleted": True})
|
||
|
||
|
||
@router.delete("/roles/{role_id}/permanent")
|
||
def permanent_delete_role(
|
||
role_id: int,
|
||
session: Session = Depends(get_db_session),
|
||
current_user: dict = Depends(require_roles("admin")),
|
||
_permission_user: dict = Depends(require_permissions("system:role:update")),
|
||
) -> dict:
|
||
"""永久删除角色"""
|
||
role = session.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
|
||
if not role:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="角色不存在", status_code=404)
|
||
if role.role_code == "admin":
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="不能删除管理员角色", status_code=400)
|
||
count = session.execute(
|
||
select(func.count(User.id)).where(User.role_id == role_id, User.status == 1)
|
||
).scalar()
|
||
if count > 0:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="该角色下还有用户，无法删除", status_code=400)
|
||
session.delete(role)
|
||
session.commit()
|
||
return success_payload({"deleted": True})
|
||
|
||
|
||
@router.post("/users/batch-delete")
|
||
def batch_delete_users(
|
||
ids: list[int] = Body(..., embed=True),
|
||
permanent: bool = Body(default=False, embed=True),
|
||
session: Session = Depends(get_db_session),
|
||
current_user: dict = Depends(require_roles("admin")),
|
||
_permission_user: dict = Depends(require_permissions("system:user:update")),
|
||
) -> dict:
|
||
"""批量删除用户"""
|
||
users = session.execute(select(User).where(User.id.in_(ids))).scalars().all()
|
||
if not users:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="未找到可删除的用户", status_code=404)
|
||
protected = [u.username for u in users if u.username == "admin"]
|
||
if protected:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="不能删除管理员账号", status_code=400)
|
||
for u in users:
|
||
if u.username != "admin":
|
||
session.delete(u)
|
||
session.commit()
|
||
return success_payload({"deleted": len(users) - len(protected)})
|
||
|
||
|
||
@router.post("/roles/batch-delete")
|
||
def batch_delete_roles(
|
||
ids: list[int] = Body(..., embed=True),
|
||
permanent: bool = Body(default=False, embed=True),
|
||
session: Session = Depends(get_db_session),
|
||
current_user: dict = Depends(require_roles("admin")),
|
||
_permission_user: dict = Depends(require_permissions("system:role:update")),
|
||
) -> dict:
|
||
"""批量删除角色"""
|
||
roles = session.execute(select(Role).where(Role.id.in_(ids))).scalars().all()
|
||
if not roles:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="未找到可删除的角色", status_code=404)
|
||
protected = [r.role_code for r in roles if r.role_code == "admin"]
|
||
if protected:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="不能删除管理员角色", status_code=400)
|
||
role_ids = [r.id for r in roles if r.role_code != "admin"]
|
||
count = session.execute(
|
||
select(func.count(User.id)).where(User.role_id.in_(role_ids), User.status == 1)
|
||
).scalar()
|
||
if count > 0:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="所选角色下还有用户，无法删除", status_code=400)
|
||
for r in roles:
|
||
if r.role_code != "admin":
|
||
session.delete(r)
|
||
session.commit()
|
||
return success_payload({"deleted": len(roles) - len(protected)})
|
||
|
||
|
||
@router.get("/users/drivers")
|
||
def list_drivers(
|
||
session: Session = Depends(get_db_session),
|
||
current_user: dict = Depends(require_roles("manager", "admin")),
|
||
) -> dict:
|
||
"""
|
||
获取司机列表（轻量级，用于下拉选择）
|
||
|
||
返回所有启用状态的司机用户，仅包含 user_id, real_name, mobile 字段。
|
||
用于物流任务分配时的司机选择下拉框。
|
||
|
||
权限要求：manager 或 admin 角色
|
||
"""
|
||
# 先查询司机角色
|
||
driver_role = session.execute(
|
||
select(Role).where(Role.role_code == "driver", Role.status == 1)
|
||
).scalar_one_or_none()
|
||
|
||
if not driver_role:
|
||
return success_payload([])
|
||
|
||
# 查询启用状态的司机用户
|
||
users = session.execute(
|
||
select(User)
|
||
.where(User.role_id == driver_role.id, User.status == 1)
|
||
.order_by(User.id.asc())
|
||
).scalars().all()
|
||
|
||
return success_payload([
|
||
{
|
||
"user_id": u.id,
|
||
"real_name": u.real_name or u.username,
|
||
"mobile": u.mobile or "",
|
||
}
|
||
for u in users
|
||
])
|