from typing import Optional, Dict, Any from fastapi.exceptions import RequestValidationError from backend.infra.permission.role_handler import RoleHandler from backend.models.permission.models import RoleDoc from beanie import PydanticObjectId class RoleService: def __init__(self): self.role_handler = RoleHandler() async def create_role(self, role_key: str, role_name: str, role_description: Optional[str], role_level: int) -> RoleDoc: """Create a new role, ensuring role_key and role_name are unique and not empty""" doc = await self.role_handler.create_role(role_key, role_name, role_description, role_level) return doc async def update_role(self, role_id: str, role_key: str, role_name: str, role_description: Optional[str], role_level: int) -> RoleDoc: """Update an existing role, ensuring role_key and role_name are unique and not empty""" doc = await self.role_handler.update_role(PydanticObjectId(role_id), role_key, role_name, role_description, role_level) return doc async def query_roles(self, role_key: Optional[str], role_name: Optional[str], page: int = 1, page_size: int = 10) -> Dict[str, Any]: """Query roles with pagination and fuzzy search by role_key and role_name""" if page < 1 or page_size < 1: raise RequestValidationError("page and page_size must be positive integers.") skip = (page - 1) * page_size docs, total = await self.role_handler.query_roles(role_key, role_name, skip, page_size) return { "items": [doc.dict() for doc in docs], "total": total, "page": page, "page_size": page_size }