Cheng Zhou
248 lines
8.8 KiB
Python
248 lines
8.8 KiB
Python
from typing import Optional
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.permission_template import PermissionTemplate, PermissionTemplateVersion, TemplateType
|
|
from app.models.api_endpoint_permission import ApiEndpointPermission
|
|
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, PROJECT_PERMISSION_ROLES
|
|
from app.schemas.permission_template import PermissionTemplateCreate, PermissionTemplateUpdate
|
|
|
|
|
|
class PermissionTemplateService:
|
|
"""权限模板服务"""
|
|
|
|
@staticmethod
|
|
async def create_template(
|
|
db: AsyncSession,
|
|
payload: PermissionTemplateCreate,
|
|
created_by: UUID,
|
|
) -> PermissionTemplate:
|
|
"""创建权限模板"""
|
|
if payload.template_type != TemplateType.CUSTOM:
|
|
raise ValueError("只能新增自定义角色")
|
|
|
|
# 验证权限配置
|
|
await PermissionTemplateService._validate_permissions(payload.permissions)
|
|
|
|
template = PermissionTemplate(
|
|
name=payload.name,
|
|
description=payload.description,
|
|
template_type=payload.template_type,
|
|
is_system=False,
|
|
created_by=created_by,
|
|
permissions=payload.permissions,
|
|
tags=payload.tags,
|
|
category=payload.category,
|
|
recommended_roles=payload.recommended_roles,
|
|
)
|
|
|
|
db.add(template)
|
|
await db.flush() # 获取 template.id
|
|
|
|
# 创建版本 1
|
|
version = PermissionTemplateVersion(
|
|
template_id=template.id,
|
|
version=1,
|
|
permissions=payload.permissions,
|
|
change_log="初始版本",
|
|
)
|
|
|
|
db.add(version)
|
|
await db.commit()
|
|
await db.refresh(template)
|
|
|
|
return template
|
|
|
|
@staticmethod
|
|
async def get_template(db: AsyncSession, template_id: UUID) -> Optional[PermissionTemplate]:
|
|
"""获取权限模板"""
|
|
result = await db.execute(
|
|
select(PermissionTemplate)
|
|
.options(selectinload(PermissionTemplate.versions))
|
|
.where(PermissionTemplate.id == template_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
@staticmethod
|
|
async def list_templates(
|
|
db: AsyncSession,
|
|
template_type: Optional[TemplateType] = None,
|
|
category: Optional[str] = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> list[PermissionTemplate]:
|
|
"""列表权限模板"""
|
|
query = select(PermissionTemplate).options(selectinload(PermissionTemplate.versions))
|
|
|
|
if template_type:
|
|
query = query.where(PermissionTemplate.template_type == template_type)
|
|
if category:
|
|
query = query.where(PermissionTemplate.category == category)
|
|
elif template_type == TemplateType.ROLE:
|
|
query = query.where(
|
|
(PermissionTemplate.is_system.is_(False))
|
|
| (PermissionTemplate.category.in_(PROJECT_PERMISSION_ROLES))
|
|
)
|
|
|
|
query = query.offset(skip).limit(limit)
|
|
result = await db.execute(query)
|
|
return result.scalars().all()
|
|
|
|
@staticmethod
|
|
async def update_template(
|
|
db: AsyncSession,
|
|
template_id: UUID,
|
|
payload: PermissionTemplateUpdate,
|
|
) -> PermissionTemplate:
|
|
"""更新权限模板"""
|
|
template = await PermissionTemplateService.get_template(db, template_id)
|
|
if not template:
|
|
raise ValueError(f"模板 {template_id} 不存在")
|
|
|
|
# 系统预设模板只允许更新展示元数据，权限配置和分类等结构字段保持受保护。
|
|
if template.is_system:
|
|
protected_fields = {"permissions", "tags", "category", "recommended_roles"}
|
|
if protected_fields.intersection(payload.model_fields_set):
|
|
raise ValueError("不允许修改系统预设模板")
|
|
|
|
# 验证权限配置
|
|
if payload.permissions:
|
|
await PermissionTemplateService._validate_permissions(payload.permissions)
|
|
|
|
# 更新字段
|
|
if payload.name:
|
|
template.name = payload.name
|
|
if payload.description is not None:
|
|
template.description = payload.description
|
|
if payload.permissions:
|
|
# 加载 versions 以获取最新版本号
|
|
await db.refresh(template, ["versions"])
|
|
latest_version = max([v.version for v in template.versions], default=0)
|
|
new_version = PermissionTemplateVersion(
|
|
template_id=template.id,
|
|
version=latest_version + 1,
|
|
permissions=payload.permissions,
|
|
change_log="",
|
|
)
|
|
db.add(new_version)
|
|
template.permissions = payload.permissions
|
|
if payload.tags is not None:
|
|
template.tags = payload.tags
|
|
if payload.category is not None:
|
|
template.category = payload.category
|
|
if payload.recommended_roles is not None:
|
|
template.recommended_roles = payload.recommended_roles
|
|
|
|
await db.commit()
|
|
await db.refresh(template)
|
|
|
|
return template
|
|
|
|
@staticmethod
|
|
async def delete_template(db: AsyncSession, template_id: UUID) -> None:
|
|
"""删除权限模板"""
|
|
template = await PermissionTemplateService.get_template(db, template_id)
|
|
if not template:
|
|
raise ValueError(f"模板 {template_id} 不存在")
|
|
|
|
# 不允许删除系统预设模板
|
|
if template.is_system:
|
|
raise ValueError("不允许删除系统预设模板")
|
|
|
|
await db.delete(template)
|
|
await db.commit()
|
|
|
|
@staticmethod
|
|
async def apply_template(
|
|
db: AsyncSession,
|
|
study_id: UUID,
|
|
template_id: UUID,
|
|
roles: Optional[list[str]] = None,
|
|
override: bool = True,
|
|
) -> dict:
|
|
"""应用权限模板到项目"""
|
|
template = await PermissionTemplateService.get_template(db, template_id)
|
|
if not template:
|
|
raise ValueError(f"模板 {template_id} 不存在")
|
|
|
|
# 确定要应用的角色
|
|
if roles is None:
|
|
if template.recommended_roles:
|
|
roles = template.recommended_roles.split(",")
|
|
else:
|
|
roles = list(template.permissions.keys())
|
|
|
|
# 删除现有权限（如果覆盖）
|
|
if override:
|
|
await db.execute(
|
|
delete(ApiEndpointPermission).where(
|
|
ApiEndpointPermission.study_id == study_id,
|
|
ApiEndpointPermission.role.in_(roles),
|
|
)
|
|
)
|
|
|
|
# 应用模板权限
|
|
for role in roles:
|
|
if role not in template.permissions:
|
|
continue
|
|
|
|
role_permissions = template.permissions[role]
|
|
for endpoint_key, allowed in role_permissions.items():
|
|
# 检查权限是否存在
|
|
if endpoint_key not in API_ENDPOINT_PERMISSIONS:
|
|
continue
|
|
|
|
# 检查是否已存在
|
|
result = await db.execute(
|
|
select(ApiEndpointPermission).where(
|
|
ApiEndpointPermission.study_id == study_id,
|
|
ApiEndpointPermission.role == role,
|
|
ApiEndpointPermission.endpoint_key == endpoint_key,
|
|
)
|
|
)
|
|
perm = result.scalar_one_or_none()
|
|
|
|
if perm:
|
|
perm.allowed = allowed
|
|
else:
|
|
perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role=role,
|
|
endpoint_key=endpoint_key,
|
|
allowed=allowed,
|
|
)
|
|
db.add(perm)
|
|
|
|
await db.commit()
|
|
|
|
# 返回应用后的权限矩阵
|
|
from app.core.project_permissions import get_api_endpoint_permissions
|
|
|
|
permissions = await get_api_endpoint_permissions(db, study_id)
|
|
|
|
return {
|
|
"study_id": study_id,
|
|
"applied_roles": roles,
|
|
"permissions": permissions,
|
|
}
|
|
|
|
@staticmethod
|
|
async def _validate_permissions(permissions: dict) -> None:
|
|
"""验证权限配置"""
|
|
for role, role_permissions in permissions.items():
|
|
if not isinstance(role_permissions, dict):
|
|
raise ValueError(f"角色 {role} 的权限配置格式错误")
|
|
|
|
for endpoint_key, allowed in role_permissions.items():
|
|
if endpoint_key.startswith("_"): # 跳过元数据字段
|
|
continue
|
|
|
|
if endpoint_key not in API_ENDPOINT_PERMISSIONS:
|
|
raise ValueError(f"权限操作 {endpoint_key} 不存在")
|
|
|
|
if not isinstance(allowed, bool):
|
|
raise ValueError(f"权限 {endpoint_key} 的值必须是布尔类型")
|