统一项目权限读取模型

1、合并列表与详情读取权限,统一使用 subjects:read、project_members:read、monitoring_issues:read 等业务读取 key。

2、增加历史读取权限 key 的兼容映射,确保存量项目权限覆盖值不会丢失。

3、同步前端权限工具、路由权限与权限管理页面,避免继续引用已合并的旧权限 key。

4、补充项目角色启停保护与抽屉未保存变更守卫,防止停用 PM 或仍有关联成员的角色。
This commit is contained in:
Cheng Zhou
2026-06-04 11:06:03 +08:00
parent 44d69c2d7b
commit ab4f0d93ed
22 changed files with 852 additions and 349 deletions
+14 -2
View File
@@ -7,7 +7,12 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.models.api_endpoint_permission import ApiEndpointPermission
from app.models.study import Study
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, OPERATION_PREREQUISITES, PROJECT_PERMISSION_ROLES
from app.core.api_permissions import (
API_ENDPOINT_PERMISSIONS,
OPERATION_PREREQUISITES,
PROJECT_PERMISSION_ROLES,
canonical_permission_key,
)
from app.core.permission_cache import get_permission_cache
@@ -34,7 +39,10 @@ async def _get_project_permission_overrides(
permissions: dict[str, dict[str, bool]] = {}
for row in rows:
permissions.setdefault(row.role, {})[row.endpoint_key] = row.allowed
endpoint_key = canonical_permission_key(row.endpoint_key)
role_permissions = permissions.setdefault(row.role, {})
if row.endpoint_key == endpoint_key or endpoint_key not in role_permissions:
role_permissions[endpoint_key] = row.allowed
cache.set_project_permissions(study_id, permissions)
return permissions
@@ -48,6 +56,7 @@ async def role_has_api_permission(
check_prerequisites: bool = True,
) -> bool:
"""检查角色是否有权访问特定接口"""
endpoint_key = canonical_permission_key(endpoint_key)
if endpoint_key not in API_ENDPOINT_PERMISSIONS:
return False
@@ -93,6 +102,7 @@ async def get_missing_prerequisites(
return []
missing = []
endpoint_key = canonical_permission_key(endpoint_key)
prerequisites = OPERATION_PREREQUISITES.get(endpoint_key, [])
for prereq_endpoint in prerequisites:
has_prereq = await role_has_api_permission(
@@ -132,6 +142,7 @@ async def get_api_endpoint_permissions(
if role not in matrix:
matrix[role] = {}
for endpoint_key, allowed in endpoints.items():
endpoint_key = canonical_permission_key(endpoint_key)
if endpoint_key not in API_ENDPOINT_PERMISSIONS:
continue
matrix[role][endpoint_key] = {"allowed": allowed}
@@ -153,6 +164,7 @@ async def replace_api_endpoint_permissions(
if role in ("ADMIN", "PM"):
continue
for endpoint_key, allowed in endpoints.items():
endpoint_key = canonical_permission_key(endpoint_key)
if endpoint_key not in API_ENDPOINT_PERMISSIONS:
continue
result = await db.execute(