acb86e4fe1
1. 为权限相关响应schema新增createtime和updatetime字段 2. 在权限接口返回中补充返回创建和更新时间 3. 在设置管理页面新增更新时间列并调整创建时间展示顺序 4. 添加日期时间格式化工具函数 5. 注释删除设置的按钮暂时禁用删除功能
381 lines
12 KiB
Python
381 lines
12 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, and_
|
||
from sqlalchemy.orm import selectinload
|
||
from fastcrud import FastCRUD
|
||
from typing import Dict, Any
|
||
|
||
from database import get_db
|
||
from models import Permission, Role
|
||
from schemas import (
|
||
PermissionCreate,
|
||
PermissionResponse,
|
||
PermissionTreeResponse,
|
||
RolePermissionAssign,
|
||
RolePermissionResponse,
|
||
)
|
||
from middleware import get_current_user
|
||
from models import User
|
||
from utils.like_filter import ilike_contains
|
||
|
||
router = APIRouter(prefix="/permissions", tags=["permissions"])
|
||
|
||
# 创建FastCRUD实例
|
||
permission_crud = FastCRUD(Permission, PermissionResponse)
|
||
|
||
|
||
@router.post("/", response_model=PermissionResponse)
|
||
async def create_permission(
|
||
permission_data: PermissionCreate,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""创建权限"""
|
||
# 检查权限编码是否已存在
|
||
result = await db.execute(select(Permission).filter(Permission.code == permission_data.code))
|
||
existing_permission = result.scalar_one_or_none()
|
||
if existing_permission:
|
||
raise HTTPException(status_code=400, detail="权限编码已存在")
|
||
|
||
# 创建权限实例
|
||
db_permission = Permission(
|
||
name=permission_data.name,
|
||
code=permission_data.code,
|
||
parentid=permission_data.parentid,
|
||
description=permission_data.description,
|
||
)
|
||
|
||
db.add(db_permission)
|
||
await db.commit()
|
||
await db.refresh(db_permission)
|
||
|
||
return db_permission
|
||
|
||
|
||
@router.post("/search", response_model=list[PermissionResponse])
|
||
async def search_permissions(
|
||
query: dict = {},
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取权限列表（POST方式）；支持按名称、编码的模糊查询与组合查询（AND）"""
|
||
stmt = select(Permission)
|
||
filters = []
|
||
for col, key in ((Permission.name, "name"), (Permission.code, "code")):
|
||
raw = query.get(key)
|
||
cond = ilike_contains(col, raw)
|
||
if cond is not None:
|
||
filters.append(cond)
|
||
if filters:
|
||
stmt = stmt.where(and_(*filters))
|
||
result = await db.execute(stmt)
|
||
permissions = result.scalars().all()
|
||
return [PermissionResponse.model_validate(p) for p in permissions]
|
||
|
||
|
||
@router.get("/tree", response_model=list[PermissionTreeResponse])
|
||
async def get_permission_tree(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取权限树形结构（基于 parentid）"""
|
||
# 获取所有权限
|
||
result = await db.execute(select(Permission))
|
||
permissions = result.scalars().all()
|
||
|
||
# 转换为字典，便于查找
|
||
perm_dict = {perm.id: {
|
||
"id": perm.id,
|
||
"name": perm.name,
|
||
"code": perm.code,
|
||
"parentid": perm.parentid,
|
||
"description": perm.description,
|
||
"createtime": perm.createtime,
|
||
"updatetime": perm.updatetime,
|
||
"children": []
|
||
} for perm in permissions}
|
||
|
||
# 构建树形结构
|
||
tree = []
|
||
for perm_id, perm_data in perm_dict.items():
|
||
parentid = perm_data["parentid"]
|
||
if parentid == 0:
|
||
# 顶级节点
|
||
tree.append(perm_data)
|
||
elif parentid in perm_dict:
|
||
# 子节点，添加到父节点的 children
|
||
perm_dict[parentid]["children"].append(perm_data)
|
||
|
||
return tree
|
||
|
||
|
||
@router.get("/menu", response_model=list[PermissionTreeResponse])
|
||
async def get_menu(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取当前用户的菜单树形结构（仅包含有路由的权限）"""
|
||
# 获取用户角色的权限
|
||
result = await db.execute(
|
||
select(Role)
|
||
.options(selectinload(Role.permissions))
|
||
.filter(Role.id == current_user.role_id)
|
||
)
|
||
role = result.scalar_one_or_none()
|
||
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
|
||
# 获取用户有权限的权限编码集合
|
||
user_permission_codes = {perm.code for perm in role.permissions}
|
||
|
||
# 获取所有顶级模块和二级模块权限（有route字段的）
|
||
result = await db.execute(
|
||
select(Permission)
|
||
.filter(Permission.route.isnot(None))
|
||
)
|
||
all_menu_permissions = result.scalars().all()
|
||
|
||
# 过滤用户有权限的菜单
|
||
user_menu_permissions = []
|
||
for perm in all_menu_permissions:
|
||
# 检查用户是否有该权限或其子权限
|
||
has_access = False
|
||
for user_perm_code in user_permission_codes:
|
||
if user_perm_code.startswith(perm.code):
|
||
has_access = True
|
||
break
|
||
if has_access:
|
||
user_menu_permissions.append(perm)
|
||
|
||
# 转换为字典，便于查找
|
||
perm_dict = {perm.id: {
|
||
"id": perm.id,
|
||
"name": perm.name,
|
||
"code": perm.code,
|
||
"parentid": perm.parentid,
|
||
"route": perm.route,
|
||
"description": perm.description,
|
||
"createtime": perm.createtime,
|
||
"updatetime": perm.updatetime,
|
||
"children": []
|
||
} for perm in user_menu_permissions}
|
||
|
||
# 构建树形结构
|
||
tree = []
|
||
for perm_id, perm_data in perm_dict.items():
|
||
parentid = perm_data["parentid"]
|
||
if parentid == 0:
|
||
# 顶级节点
|
||
tree.append(perm_data)
|
||
elif parentid in perm_dict:
|
||
# 子节点，添加到父节点的 children
|
||
perm_dict[parentid]["children"].append(perm_data)
|
||
|
||
return tree
|
||
|
||
|
||
@router.get("/{permission_id}", response_model=PermissionResponse)
|
||
async def get_permission(
|
||
permission_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取单个权限"""
|
||
permission = await permission_crud.get(db, id=permission_id)
|
||
if not permission:
|
||
raise HTTPException(status_code=404, detail="权限不存在")
|
||
return permission
|
||
|
||
|
||
@router.put("/{permission_id}", response_model=PermissionResponse)
|
||
async def update_permission(
|
||
permission_id: int,
|
||
permission_data: PermissionCreate,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""更新权限"""
|
||
# 检查权限编码是否被其他权限使用
|
||
result = await db.execute(
|
||
select(Permission).filter(Permission.code == permission_data.code, Permission.id != permission_id)
|
||
)
|
||
existing_permission = result.scalar_one_or_none()
|
||
if existing_permission:
|
||
raise HTTPException(status_code=400, detail="权限编码已存在")
|
||
|
||
# 准备更新数据
|
||
update_data = permission_data.model_dump()
|
||
|
||
# 使用FastCRUD更新权限
|
||
updated_permission = await permission_crud.update(
|
||
db, update_data, id=permission_id, schema_to_select=PermissionResponse, return_as_model=True
|
||
)
|
||
if not updated_permission:
|
||
raise HTTPException(status_code=404, detail="权限不存在")
|
||
return updated_permission
|
||
|
||
|
||
@router.delete("/{permission_id}")
|
||
async def delete_permission(
|
||
permission_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""删除权限（自动清理角色关联和子权限关联）"""
|
||
# 检查权限是否存在
|
||
result = await db.execute(select(Permission).filter(Permission.id == permission_id))
|
||
existing_permission = result.scalar_one_or_none()
|
||
if not existing_permission:
|
||
raise HTTPException(status_code=404, detail="权限不存在")
|
||
|
||
# 1. 删除角色-权限关联表中的记录
|
||
from sqlalchemy import delete as sql_delete
|
||
from models import role_permission
|
||
|
||
stmt = sql_delete(role_permission).where(role_permission.c.permission_id == permission_id)
|
||
await db.execute(stmt)
|
||
|
||
# 2. 处理子权限：如果有子权限，将子权限的父ID置为0（提升为顶级节点）
|
||
from sqlalchemy import update as sql_update
|
||
|
||
stmt = (
|
||
sql_update(Permission)
|
||
.where(Permission.parentid == permission_id)
|
||
.values(parentid=0)
|
||
)
|
||
await db.execute(stmt)
|
||
|
||
# 3. 使用FastCRUD删除权限本身（不提交，与前述操作在同一事务中）
|
||
await permission_crud.delete(db, id=permission_id, commit=False)
|
||
await db.commit()
|
||
return {"message": "权限删除成功"}
|
||
|
||
|
||
# 角色权限关联接口
|
||
@router.post("/assign", response_model=RolePermissionResponse)
|
||
async def assign_permissions_to_role(
|
||
data: RolePermissionAssign,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""为角色分配权限"""
|
||
# 查找角色，同时加载权限关系
|
||
result = await db.execute(select(Role).options(selectinload(Role.permissions)).filter(Role.id == data.role_id))
|
||
role = result.scalar_one_or_none()
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
|
||
# 查找要分配的权限
|
||
permissions = []
|
||
for permission_id in data.permission_ids:
|
||
result = await db.execute(select(Permission).filter(Permission.id == permission_id))
|
||
permission = result.scalar_one_or_none()
|
||
if permission:
|
||
permissions.append(permission)
|
||
|
||
# 更新角色的权限
|
||
role.permissions = permissions
|
||
await db.commit()
|
||
await db.refresh(role)
|
||
|
||
# 返回角色权限信息
|
||
return RolePermissionResponse(
|
||
role_id=role.id,
|
||
role_name=role.name,
|
||
permissions=[PermissionResponse.from_orm(p) for p in permissions]
|
||
)
|
||
|
||
|
||
@router.get("/role/{role_id}", response_model=RolePermissionResponse)
|
||
async def get_role_permissions(
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取角色的权限列表"""
|
||
# 查找角色，同时加载权限关系
|
||
result = await db.execute(select(Role).options(selectinload(Role.permissions)).filter(Role.id == role_id))
|
||
role = result.scalar_one_or_none()
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
|
||
# 返回角色权限信息
|
||
return RolePermissionResponse(
|
||
role_id=role.id,
|
||
role_name=role.name,
|
||
permissions=[PermissionResponse.from_orm(p) for p in role.permissions]
|
||
)
|
||
|
||
|
||
@router.get("/user/me", response_model=RolePermissionResponse)
|
||
async def get_current_user_permissions(
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取当前登录用户的权限列表和菜单"""
|
||
# 获取用户角色的权限
|
||
result = await db.execute(
|
||
select(Role)
|
||
.options(selectinload(Role.permissions))
|
||
.filter(Role.id == current_user.role_id)
|
||
)
|
||
role = result.scalar_one_or_none()
|
||
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
|
||
# 获取用户有权限的权限编码集合
|
||
user_permission_codes = {perm.code for perm in role.permissions}
|
||
|
||
# 获取所有顶级模块和二级模块权限（有route字段的）
|
||
result = await db.execute(
|
||
select(Permission)
|
||
.filter(Permission.route.isnot(None))
|
||
)
|
||
all_menu_permissions = result.scalars().all()
|
||
|
||
# 过滤用户有权限的菜单
|
||
user_menu_permissions = []
|
||
for perm in all_menu_permissions:
|
||
# 检查用户是否有该权限或其子权限
|
||
has_access = False
|
||
for user_perm_code in user_permission_codes:
|
||
if user_perm_code.startswith(perm.code):
|
||
has_access = True
|
||
break
|
||
if has_access:
|
||
user_menu_permissions.append(perm)
|
||
|
||
# 转换为字典，便于查找
|
||
perm_dict = {perm.id: {
|
||
"id": perm.id,
|
||
"name": perm.name,
|
||
"code": perm.code,
|
||
"parentid": perm.parentid,
|
||
"route": perm.route,
|
||
"description": perm.description,
|
||
"createtime": perm.createtime,
|
||
"updatetime": perm.updatetime,
|
||
"children": []
|
||
} for perm in user_menu_permissions}
|
||
|
||
# 构建树形结构
|
||
menu_tree = []
|
||
for perm_id, perm_data in perm_dict.items():
|
||
parentid = perm_data["parentid"]
|
||
if parentid == 0:
|
||
# 顶级节点
|
||
menu_tree.append(perm_data)
|
||
elif parentid in perm_dict:
|
||
# 子节点，添加到父节点的 children
|
||
perm_dict[parentid]["children"].append(perm_data)
|
||
|
||
# 返回用户权限信息和菜单
|
||
return RolePermissionResponse(
|
||
role_id=role.id,
|
||
role_name=role.name,
|
||
permissions=[PermissionResponse.from_orm(p) for p in role.permissions],
|
||
menu=menu_tree
|
||
)
|