Cheng Zhou
6eea6b35fd
1. 修复 ApiPermissions.vue 中的数据绑定错误 - 从 axios 响应中正确提取 .data 属性 - 修复 loadPermissionData、loadMonitoringData、save 函数 2. 禁用 FastAPI 响应验证 - 在 api_permissions.py 中添加 response_model=None - 避免 Pydantic 验证错误 3. 注册权限监控 API 路由 - 在 router.py 中导入并注册 permission_monitoring 模块 - 修改 permission_monitoring.py 的路由前缀 Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
135 lines
3.8 KiB
Python
135 lines
3.8 KiB
Python
"""接口级权限管理API"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.deps import get_db_session, require_study_roles
|
|
from app.core.project_permissions import (
|
|
get_api_endpoint_permissions,
|
|
replace_api_endpoint_permissions,
|
|
)
|
|
from app.models.api_endpoint_registry import ApiEndpointRegistry
|
|
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, PROJECT_PERMISSION_ROLES
|
|
|
|
router = APIRouter(prefix="/api-permissions", tags=["api-permissions"])
|
|
|
|
|
|
@router.get(
|
|
"/endpoints",
|
|
summary="获取系统中所有已注册的API端点",
|
|
description="返回系统中所有已注册的API端点及其元数据",
|
|
)
|
|
async def list_api_endpoints(
|
|
db: Annotated[AsyncSession, Depends(get_db_session)],
|
|
) -> dict[str, list[dict]]:
|
|
"""获取所有已注册的API端点"""
|
|
result = await db.execute(select(ApiEndpointRegistry))
|
|
endpoints = result.scalars().all()
|
|
|
|
endpoints_list = [
|
|
{
|
|
"endpoint_key": ep.endpoint_key,
|
|
"method": ep.method,
|
|
"path": ep.path,
|
|
"module": ep.module,
|
|
"action": ep.action,
|
|
"description": ep.description,
|
|
"default_roles": ep.default_roles.split(",") if ep.default_roles else [],
|
|
}
|
|
for ep in endpoints
|
|
]
|
|
|
|
return {"endpoints": endpoints_list}
|
|
|
|
|
|
@router.get(
|
|
"",
|
|
summary="获取项目的接口级权限矩阵",
|
|
description="返回项目中各角色对API端点的权限配置",
|
|
response_model=None,
|
|
)
|
|
async def get_study_api_permissions(
|
|
study_id: uuid.UUID,
|
|
_=Depends(require_study_roles(["PM"])),
|
|
db: Annotated[AsyncSession, Depends(get_db_session)] = None,
|
|
):
|
|
"""获取项目的接口级权限矩阵
|
|
|
|
返回格式:
|
|
{
|
|
"role": {
|
|
"endpoint_key": {
|
|
"allowed": true/false
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
permissions = await get_api_endpoint_permissions(db, study_id)
|
|
|
|
# 构建返回格式
|
|
result: dict[str, dict[str, dict[str, bool]]] = {}
|
|
for role in PROJECT_PERMISSION_ROLES:
|
|
if role == "ADMIN":
|
|
continue
|
|
result[role] = {}
|
|
for endpoint_key in API_ENDPOINT_PERMISSIONS.keys():
|
|
result[role][endpoint_key] = {
|
|
"allowed": permissions.get(role, {}).get(endpoint_key, False)
|
|
}
|
|
|
|
return result
|
|
|
|
|
|
@router.put(
|
|
"",
|
|
summary="更新项目的接口级权限矩阵",
|
|
description="批量更新项目中各角色对API端点的权限配置",
|
|
status_code=status.HTTP_200_OK,
|
|
response_model=None,
|
|
)
|
|
async def update_study_api_permissions(
|
|
study_id: uuid.UUID,
|
|
payload: dict[str, dict[str, bool]],
|
|
_=Depends(require_study_roles(["PM"])),
|
|
db: Annotated[AsyncSession, Depends(get_db_session)] = None,
|
|
):
|
|
"""更新项目的接口级权限矩阵
|
|
|
|
请求体格式:
|
|
{
|
|
"role": {
|
|
"endpoint_key": true/false
|
|
}
|
|
}
|
|
"""
|
|
# 验证输入
|
|
for role in payload.keys():
|
|
if role == "ADMIN":
|
|
continue
|
|
if role not in PROJECT_PERMISSION_ROLES:
|
|
raise ValueError(f"无效的角色: {role}")
|
|
|
|
# 替换权限配置
|
|
await replace_api_endpoint_permissions(db, study_id, payload)
|
|
|
|
# 返回更新后的权限矩阵
|
|
permissions = await get_api_endpoint_permissions(db, study_id)
|
|
|
|
result: dict[str, dict[str, dict[str, bool]]] = {}
|
|
for role in PROJECT_PERMISSION_ROLES:
|
|
if role == "ADMIN":
|
|
continue
|
|
result[role] = {}
|
|
for endpoint_key in API_ENDPOINT_PERMISSIONS.keys():
|
|
result[role][endpoint_key] = {
|
|
"allowed": permissions.get(role, {}).get(endpoint_key, False)
|
|
}
|
|
|
|
return result
|