Files
ctms/backend/app/api/v1/project_permissions.py
T
Cheng Zhou 939fc70532 权限管理:接入项目级权限矩阵
新增项目级角色权限矩阵,覆盖 PM、CRA、PV、IMP、QA 等项目角色,并通过迁移表持久化各业务模块的读写权限。

后端将参与者、合同费用、物资、启动、里程碑、风险问题、监查稽查、FAQ、共享库、文件版本等接口接入矩阵校验,修复审计日志导出权限和登录 502 的依赖导入问题。

前端新增权限管理页面和项目路由权限映射,按矩阵控制导航显示、直接访问拦截、操作按钮启停,并限制管理后台仅 admin 和授权 PM 可进入。

补充后端权限归一化与接口静态测试、前端路由/权限工具/权限管理页面/工作台等回归测试。
2026-05-13 08:58:03 +08:00

67 lines
2.5 KiB
Python

from __future__ import annotations
import json
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.deps import get_current_user, get_db_session, require_roles, require_study_member
from app.core.project_permissions import PROJECT_PERMISSION_MODULES, get_project_role_permissions, replace_project_role_permissions
from app.crud import audit as audit_crud
from app.crud import study as study_crud
from app.schemas.project_permission import ProjectPermissionModule, ProjectRolePermissionsRead, ProjectRolePermissionsUpdate
router = APIRouter()
async def _ensure_study_exists(db: AsyncSession, study_id: uuid.UUID):
study = await study_crud.get(db, study_id)
if not study:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="项目不存在")
return study
@router.get("/", response_model=ProjectRolePermissionsRead, dependencies=[Depends(require_study_member())])
async def get_permissions(
study_id: uuid.UUID,
db: AsyncSession = Depends(get_db_session),
) -> ProjectRolePermissionsRead:
await _ensure_study_exists(db, study_id)
roles = await get_project_role_permissions(db, study_id)
modules = [ProjectPermissionModule(**item) for item in PROJECT_PERMISSION_MODULES]
return ProjectRolePermissionsRead(modules=modules, roles=roles)
@router.put("/", response_model=ProjectRolePermissionsRead, dependencies=[Depends(require_roles(["ADMIN"]))])
async def update_permissions(
study_id: uuid.UUID,
payload: ProjectRolePermissionsUpdate,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> ProjectRolePermissionsRead:
await _ensure_study_exists(db, study_id)
roles = await replace_project_role_permissions(
db,
study_id,
{
role: {
module: actions.model_dump()
for module, actions in modules.items()
}
for role, modules in payload.roles.items()
},
)
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="project_permissions",
entity_id=study_id,
action="PROJECT_PERMISSIONS_UPDATED",
detail=json.dumps({"targetName": "项目权限", "after": roles}, ensure_ascii=False),
operator_id=current_user.id,
operator_role=current_user.role,
)
modules = [ProjectPermissionModule(**item) for item in PROJECT_PERMISSION_MODULES]
return ProjectRolePermissionsRead(modules=modules, roles=roles)