"""角色与权限的只读接口(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)