document-management/backend/dms/api/v1/roles.py

"""角色管理：角色 CRUD、权限矩阵与用户关联。"""

from __future__ import annotations

from flask import request

from dms.api.v1.blueprint import api_v1
from dms.common.errors import InvalidArgumentError
from dms.common.pagination import PageRequest
from dms.common.response import success_response
from dms.security.decorators import admin_required, bearer_auth_required
from dms.services.role_service import (
    create_role,
    delete_role,
    list_role_permissions,
    list_role_users,
    list_roles,
    update_role,
    update_role_permissions,
    update_role_users,
)


def _integer_query(name: str, default: int) -> int:
    raw_value = request.args.get(name)
    if raw_value is None or raw_value == "":
        return default
    try:
        return int(raw_value)
    except ValueError as exc:
        raise InvalidArgumentError(f"{name}必须是整数") from exc


def _json_body():
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        raise InvalidArgumentError("请求体必须是JSON对象")
    return body


def _parse_role_id(value: str) -> int:
    if not value.isdecimal() or int(value) <= 0:
        raise InvalidArgumentError("id必须是正整数形式的字符串ID")
    return int(value)


def _parse_delete_row_version(value: str | None) -> int:
    if value is None or not value.isdecimal():
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return int(value)


@api_v1.get("/roles")
@bearer_auth_required
@admin_required
def roles():
    data = list_roles(
        keyword=request.args.get("keyword"),
        status=request.args.get("status"),
        page_request=PageRequest(
            page=_integer_query("page", 1),
            page_size=_integer_query("pageSize", 20),
        ),
    )
    return success_response(data)


@api_v1.post("/roles")
@bearer_auth_required
@admin_required
def create_role_route():
    return success_response(create_role(_json_body()), status=201)


@api_v1.put("/roles/<id>")
@bearer_auth_required
@admin_required
def update_role_route(id: str):
    return success_response(update_role(_parse_role_id(id), _json_body()))


@api_v1.delete("/roles/<id>")
@bearer_auth_required
@admin_required
def delete_role_route(id: str):
    return success_response(
        delete_role(
            _parse_role_id(id),
            _parse_delete_row_version(request.args.get("rowVersion")),
        )
    )


@api_v1.get("/roles/<id>/permissions")
@bearer_auth_required
@admin_required
def role_permissions(id: str):
    return success_response(list_role_permissions(_parse_role_id(id)))


@api_v1.put("/roles/<id>/permissions")
@bearer_auth_required
@admin_required
def update_role_permissions_route(id: str):
    return success_response(
        update_role_permissions(_parse_role_id(id), _json_body())
    )


@api_v1.get("/roles/<id>/users")
@bearer_auth_required
@admin_required
def role_users(id: str):
    return success_response(list_role_users(_parse_role_id(id)))


@api_v1.put("/roles/<id>/users")
@bearer_auth_required
@admin_required
def update_role_users_route(id: str):
    return success_response(update_role_users(_parse_role_id(id), _json_body()))