覆盖所有模块: - api 层:17 个路由文件,每个接口标注用途、参数、返回值、权限 - services 层:18 个服务文件,每个方法标注作用、参数、返回值、调用方 - repositories 层:13 个仓储文件,每个方法标注查询逻辑和被调用方 - schemas 层:11 个请求/响应体文件,每个字段标注业务含义 - core 层:config、security、exceptions、responses、error_codes - models 层:19 个 ORM 模型类,每个表标注业务含义和关联关系 - scripts:bootstrap_data、smoke_check - migrations:env.py 和版本迁移文件 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
418 lines
17 KiB
Python
418 lines
17 KiB
Python
"""
|
||
系统管理路由模块
|
||
|
||
提供用户、角色、菜单的完整CRUD管理接口,包括:
|
||
- 用户管理:列表查询、创建、更新、重置密码、启用/禁用
|
||
- 角色管理:列表查询、创建、更新、菜单权限分配
|
||
- 菜单管理:列表查询、创建、更新
|
||
|
||
URL 前缀:/api/system
|
||
权限要求:admin 角色,各操作需对应细粒度权限
|
||
"""
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.api.deps import get_system_service, require_permissions, require_roles
|
||
from backend.app.db import get_db_session
|
||
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))
|