Cheng Zhou
cb36606607
问题：api_permissions.py 中的路由已包含 /studies/{study_id}，
但在 router.py 中又添加了相同前缀，导致路径重复：
/studies/{study_id}/api-permissions/studies/{study_id}
修复：移除 api_permissions.py 中的 /studies/{study_id} 路径前缀，
只保留相对路径 "" 和 ""，由 router.py 的 include_router 提供前缀。
Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
133 lines
3.8 KiB
Python
133 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端点的权限配置",
|
|
)
|
|
async def get_study_api_permissions(
|
|
study_id: uuid.UUID,
|
|
_=Depends(require_study_roles(["PM"])),
|
|
db: Annotated[AsyncSession, Depends(get_db_session)] = None,
|
|
) -> dict[str, dict[str, dict[str, bool]]]:
|
|
"""获取项目的接口级权限矩阵
|
|
|
|
返回格式:
|
|
{
|
|
"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,
|
|
)
|
|
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,
|
|
) -> dict[str, dict[str, dict[str, bool]]]:
|
|
"""更新项目的接口级权限矩阵
|
|
|
|
请求体格式:
|
|
{
|
|
"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
|