85 lines
3.6 KiB
Python
85 lines
3.6 KiB
Python
|
"""角色与权限的只读接口（B1：先让管理员看得见）。
|
|||
|
|
|
|||
|
|
前缀与 `admin.py` 相同（`/api/v1/admin`），但**独立成文件**：`admin.py` 是表驱动的
|
|||
|
|
配置面 CRUD 工厂（`RESOURCES` 单表增删改），而角色/权限是**多对多关系查询**，
|
|||
|
|
塞进那个工厂既不自然、也会让"配置管理"和"身份管理"两件事混在一个文件里。
|
|||
|
|
|
|||
|
|
四个接口全部只读，且**不写审计**——它们返回的就是审计材料本身（"谁能访问什么"，
|
|||
|
|
合规检查要看的正是这份清单）。`docs/05` §19 登记时"审计"列标"否"。
|
|||
|
|
|
|||
|
|
所有响应都走 `docs/05` §3.3 的统一信封（复用 `app/api/views/envelope.py`）：
|
|||
|
|
列表的 `data` 是纯数组、分页元数据在 `meta`；单对象的 `meta` 只有 `trace_id`。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from fastapi import APIRouter, Depends, Path
|
|||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
|
|
|||
|
|
from app.api.dependencies.auth import build_request_context
|
|||
|
|
from app.api.dependencies.database import get_session
|
|||
|
|
from app.api.dependencies.rate_limit import enforce_rate_limit
|
|||
|
|
from app.api.views.envelope import envelope, list_envelope
|
|||
|
|
from app.core.contracts import RequestContext
|
|||
|
|
from app.service.rbac_query_service import RbacQueryService
|
|||
|
|
|
|||
|
|
router = APIRouter(
|
|||
|
|
prefix="/api/v1/admin",
|
|||
|
|
tags=["platform-admin"],
|
|||
|
|
dependencies=[Depends(enforce_rate_limit)],
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
#: 角色码在 `sys_role` 里形如 `customer` / `risk_operator` / `admin`。
|
|||
|
|
ROLE_CODE = Path(min_length=1, max_length=64, pattern=r"^[a-z0-9_]+$")
|
|||
|
|
#: `sys_user.id` 是 BIGINT UNSIGNED。
|
|||
|
|
USER_ID = Path(pattern=r"^[0-9]{1,20}$")
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/roles")
|
|||
|
|
async def list_roles(
|
|||
|
|
context: RequestContext = Depends(build_request_context), # noqa: B008
|
|||
|
|
session: AsyncSession = Depends(get_session), # noqa: B008
|
|||
|
|
) -> dict[str, object]:
|
|||
|
|
"""所有角色 + 每个角色的权限数与在用人数。"""
|
|||
|
|
page = await RbacQueryService(session).list_roles(context)
|
|||
|
|
return list_envelope(page, context)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/roles/{role_code}")
|
|||
|
|
async def get_role(
|
|||
|
|
role_code: str = ROLE_CODE,
|
|||
|
|
context: RequestContext = Depends(build_request_context), # noqa: B008
|
|||
|
|
session: AsyncSession = Depends(get_session), # noqa: B008
|
|||
|
|
) -> dict[str, object]:
|
|||
|
|
"""单个角色的详情。
|
|||
|
|
|
|||
|
|
与 `/permissions` 分开是必要的：权限为空的角色（例如刚建好还没授权的）在一个合并
|
|||
|
|
接口里很容易被"查不到权限"误判成"角色不存在"。
|
|||
|
|
"""
|
|||
|
|
data = await RbacQueryService(session).get_role_detail(context, role_code)
|
|||
|
|
return envelope(data, context)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/roles/{role_code}/permissions")
|
|||
|
|
async def list_role_permissions(
|
|||
|
|
role_code: str = ROLE_CODE,
|
|||
|
|
context: RequestContext = Depends(build_request_context), # noqa: B008
|
|||
|
|
session: AsyncSession = Depends(get_session), # noqa: B008
|
|||
|
|
) -> dict[str, object]:
|
|||
|
|
"""该角色的权限清单，按权限码排序，便于与代码里各 Service 的 `require(...)` 对照。"""
|
|||
|
|
page = await RbacQueryService(session).list_role_permissions(context, role_code)
|
|||
|
|
return list_envelope(page, context)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/users/{user_id}/roles")
|
|||
|
|
async def get_user_roles(
|
|||
|
|
user_id: str = USER_ID,
|
|||
|
|
context: RequestContext = Depends(build_request_context), # noqa: B008
|
|||
|
|
session: AsyncSession = Depends(get_session), # noqa: B008
|
|||
|
|
) -> dict[str, object]:
|
|||
|
|
"""某个用户**实际解析出来**的角色 / 权限 / 数据范围 / 可见客户。
|
|||
|
|
|
|||
|
|
这是四个接口里最有用的一个：它直接回答"这个人为什么 403"——
|
|||
|
|
走的是 `IdentityService.resolve`，与请求期完全同一条链路。
|
|||
|
|
"""
|
|||
|
|
data = await RbacQueryService(session).get_user_identity(context, user_id)
|
|||
|
|
return envelope(data, context)
|