"""角色管理业务编排。 需求 2.5.1 只定义 admin / user 两个角色，本模块属用户确认的**自主完善** （轻量版：不含权限点分配 —— `permissions` / `role_permissions` 表继续闲置， 待业务方给出权限清单后再启用）。 保护规则（都是系统自举与授权安全的最低防线，不只是界面体验）： - `admin` 角色**禁删禁停** —— `cli` 建管理员与 `require_admin` 判定都依赖 该编码存在，删除会使系统无法自举 - **有用户绑定的角色禁删** —— `user_roles` 外键是 CASCADE，直接删除会静默 剥离绑定用户的权限，属安全与审计事故；删除前的校验走 `FOR UPDATE` 锁行，堵住「校验与删除之间并发绑定」的窗口 - `role_code` 创建后不可改 —— 它是授权判定（`ROLE_ADMIN` 常量按编码比对）的锚点 - 停用角色**不解除绑定、不影响既有授权**（授权判定只看编码、不看状态）， 但**不可再分配给用户**（`user_service._resolve_roles` 拦截） """ import logging from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.constants import ( ACTION_CREATE, ACTION_DELETE, ACTION_DISABLE, ACTION_ENABLE, ACTION_UPDATE, MODULE_ROLE, ROLE_ADMIN, ) from app.core.exceptions import BadRequestError, ForbiddenError, NotFoundError from app.core.response import PageResult from app.models import Role, User from app.repositories import role_repository from app.schemas.role import RoleCreateIn, RoleManageOut, RoleQuery, RoleUpdateIn from app.services import audit_service from app.services.audit_service import AuditRecord, RequestMeta logger = logging.getLogger(__name__) def _to_out(role: Role, user_count: int) -> RoleManageOut: """ORM 对象转响应模型。禁止把 ORM 对象直接当 response_model。""" return RoleManageOut( id=role.id, role_code=role.role_code, role_name=role.role_name, description=role.description, sort_order=role.sort_order, status=role.status, user_count=user_count, ) def _user_count(session: Session, role_id: int) -> int: """单个角色的绑定用户数（复用批量统计，保持口径一致）。""" return role_repository.count_users_by_role(session, [role_id]).get(role_id, 0) def list_roles(session: Session, query: RoleQuery) -> PageResult[RoleManageOut]: """分页查询角色。 与用户表单下拉（`/users/meta/roles`）口径不同：**管理列表含停用角色**， 下拉只给启用项。 """ roles = role_repository.list_roles( session, offset=query.offset, limit=query.page_size, keyword=query.keyword, status=query.status, ) total = role_repository.count_roles(session, keyword=query.keyword, status=query.status) counts = role_repository.count_users_by_role(session, [role.id for role in roles]) return PageResult[RoleManageOut]( records=[_to_out(role, counts.get(role.id, 0)) for role in roles], total=total, page_num=query.page_num, page_size=query.page_size, ) def create_role( session: Session, payload: RoleCreateIn, *, operator: User, meta: RequestMeta | None = None, ) -> RoleManageOut: """新增角色。编码唯一、创建后不可修改，新角色固定启用。 :raises BadRequestError: 角色编码已存在（含并发提交的兜底） """ with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_ROLE, action=ACTION_CREATE, summary=f"新增角色 {payload.role_name}({payload.role_code})", request_data=payload.model_dump(exclude_none=True), ), meta=meta, ): if role_repository.get_by_code(session, payload.role_code) is not None: raise BadRequestError(f"角色编码 {payload.role_code!r} 已存在") role = Role( role_code=payload.role_code, role_name=payload.role_name, description=payload.description, sort_order=payload.sort_order, status=1, ) try: role_repository.create(session, role) out = _to_out(role, 0) session.commit() except IntegrityError as exc: # 先查后写不是原子操作：并发提交同一编码会撞 uk_roles_role_code。 # 回滚并把唯一键冲突翻译为可读的 400，而非 500 session.rollback() raise BadRequestError(f"角色编码 {payload.role_code!r} 已存在") from exc logger.info("新增角色 | role_code=%s", payload.role_code) return out def update_role( session: Session, role_id: int, payload: RoleUpdateIn, *, operator: User, meta: RequestMeta | None = None, ) -> RoleManageOut: """编辑角色名称/说明/排序（全量替换，不含 role_code）。 :raises NotFoundError: 角色不存在 """ role = role_repository.get_by_id(session, role_id) with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_ROLE, action=ACTION_UPDATE, summary=f"修改角色 {role.role_name if role else f'#{role_id}'}", request_data=payload.model_dump(exclude_none=True), ), meta=meta, ): if role is None: raise NotFoundError("角色不存在") role.role_name = payload.role_name role.description = payload.description role.sort_order = payload.sort_order # 响应与后置动作全部在 commit 之前完成（见 user_service 顶部说明） out = _to_out(role, _user_count(session, role_id)) session.commit() logger.info("修改角色 | role_id=%s", role_id) return out def set_status( session: Session, role_id: int, status: int, *, operator: User, meta: RequestMeta | None = None, ) -> RoleManageOut: """启用或禁用角色。 停用不解除用户绑定、不影响既有授权（授权判定只看编码不看状态）， 但该角色不可再分配给用户（`user_service._resolve_roles` 拦截）； 内置管理员角色不可停用。 :raises NotFoundError: 角色不存在 :raises ForbiddenError: 试图停用内置管理员角色 """ role = role_repository.get_by_id(session, role_id) verb = "启用" if status == 1 else "禁用" with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_ROLE, action=ACTION_ENABLE if status == 1 else ACTION_DISABLE, summary=f"{verb}角色 {role.role_name if role else f'#{role_id}'}", request_data={"status": status}, ), meta=meta, ): if role is None: raise NotFoundError("角色不存在") if role.role_code == ROLE_ADMIN and status == 0: raise ForbiddenError("内置管理员角色不可停用") role.status = status out = _to_out(role, _user_count(session, role_id)) session.commit() logger.info("变更角色状态 | role_id=%s | status=%s", role_id, status) return out def delete_role( session: Session, role_id: int, *, operator: User, meta: RequestMeta | None = None, ) -> None: """删除角色。 内置管理员角色不可删除；仍有关联用户的角色不可删除（提示先解除绑定）。 查询走 `FOR UPDATE` 锁行：校验与删除之间若发生并发绑定，CASCADE 会把 新绑定一并静默剥离 —— 锁行使并发绑定失败（可见）而非数据被吞（不可见）。 :raises NotFoundError: 角色不存在 :raises ForbiddenError: 试图删除内置管理员角色 :raises BadRequestError: 角色下仍有绑定的用户 """ role = role_repository.get_by_id(session, role_id, for_update=True) with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_ROLE, action=ACTION_DELETE, summary=f"删除角色 {role.role_name if role else f'#{role_id}'}", ), meta=meta, ): if role is None: raise NotFoundError("角色不存在") if role.role_code == ROLE_ADMIN: raise ForbiddenError("内置管理员角色不可删除") # user_roles 外键是 CASCADE：不拦截的话，删除会静默剥离绑定用户的权限 user_count = _user_count(session, role_id) if user_count > 0: raise BadRequestError(f"该角色下仍有 {user_count} 个用户，请先解除绑定") role_code = role.role_code # 固化：对象随删除失效 role_repository.delete(session, role) session.commit() logger.info("删除角色 | role_code=%s", role_code)