ctms/backend/app/core/decorators.py

"""API端点权限装饰器和初始化函数"""

from __future__ import annotations

from functools import wraps
from typing import Callable, Any
import uuid

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.models.api_endpoint_registry import ApiEndpointRegistry
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS


def register_api_endpoint(
    endpoint_key: str,
    module: str,
    action: str,
    description: str = "",
    default_roles: list[str] | None = None,
):
    """装饰器：注册API端点权限

    在函数上附加元数据，用于系统初始化时自动注册端点。

    参数:
        endpoint_key: 接口标识，格式为 "METHOD:/path"
        module: 关联的模块，用于向后兼容
        action: 操作类型，"read" 或 "write"
        description: 接口描述
        default_roles: 默认允许的角色列表
    """
    def decorator(func: Callable) -> Callable:
        func._endpoint_key = endpoint_key
        func._module = module
        func._action = action
        func._description = description
        func._default_roles = default_roles or []
        return func
    return decorator


async def initialize_api_endpoint_registry(db: AsyncSession) -> None:
    """初始化API端点注册表

    扫描所有已定义的API端点配置，将其写入数据库。
    如果端点已存在，则跳过；否则创建新记录。
    """
    for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
        # 检查端点是否已存在
        result = await db.execute(
            select(ApiEndpointRegistry).where(
                ApiEndpointRegistry.endpoint_key == endpoint_key,
            )
        )
        existing = result.scalar_one_or_none()
        if existing:
            continue

        # 创建新的端点注册记录
        registry = ApiEndpointRegistry(
            id=uuid.uuid4(),
            endpoint_key=endpoint_key,
            method=endpoint_key.split(":")[0],
            path=endpoint_key.split(":", 1)[1],
            module=config["module"],
            action=config["action"],
            description=config.get("description", ""),
            default_roles=",".join(config.get("default_roles", [])),
        )
        db.add(registry)

    await db.commit()


async def initialize_project_api_permissions(
    db: AsyncSession,
    study_id: uuid.UUID,
) -> None:
    """为新项目初始化默认的接口级权限

    根据API端点的默认角色配置，为项目创建初始权限记录。
    """
    from app.models.api_endpoint_permission import ApiEndpointPermission

    # 获取所有已注册的端点
    result = await db.execute(select(ApiEndpointRegistry))
    endpoints = result.scalars().all()

    # 为每个端点和默认角色创建权限记录
    for endpoint in endpoints:
        default_roles = endpoint.default_roles.split(",") if endpoint.default_roles else []
        for role in default_roles:
            if not role or role == "ADMIN":
                continue

            # 检查权限是否已存在
            perm_result = await db.execute(
                select(ApiEndpointPermission).where(
                    ApiEndpointPermission.study_id == study_id,
                    ApiEndpointPermission.role == role,
                    ApiEndpointPermission.endpoint_key == endpoint.endpoint_key,
                )
            )
            if perm_result.scalar_one_or_none():
                continue

            # 创建新的权限记录
            permission = ApiEndpointPermission(
                id=uuid.uuid4(),
                study_id=study_id,
                role=role,
                endpoint_key=endpoint.endpoint_key,
                allowed=True,
            )
            db.add(permission)

    await db.commit()