247 lines
8.0 KiB
Python
247 lines
8.0 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,
|
||
"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("/{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="权限不存在")
|
||
|
||
# 使用FastCRUD删除权限
|
||
deleted = await permission_crud.delete(db, id=permission_id)
|
||
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="角色不存在")
|
||
|
||
# 返回用户权限信息
|
||
return RolePermissionResponse(
|
||
role_id=role.id,
|
||
role_name=role.name,
|
||
permissions=[PermissionResponse.from_orm(p) for p in role.permissions]
|
||
) |