36 lines
1.9 KiB
Python
36 lines
1.9 KiB
Python
from typing import Optional, Dict, Any
|
|
|
|
from fastapi.exceptions import RequestValidationError
|
|
|
|
from backend.infra.permission.permission_handler import PermissionHandler
|
|
from backend.models.permission.models import PermissionDoc
|
|
from beanie import PydanticObjectId
|
|
|
|
class PermissionService:
|
|
def __init__(self):
|
|
self.permission_handler = PermissionHandler()
|
|
|
|
async def create_permission(self, permission_key: str, permission_name: str, description: Optional[str] = None) -> PermissionDoc:
|
|
"""Create a new permission document"""
|
|
return await self.permission_handler.create_permission(permission_key, permission_name, description)
|
|
|
|
async def update_permission(self, permission_id: str, permission_key: Optional[str] = None, permission_name: Optional[str] = None, description: Optional[str] = None) -> PermissionDoc:
|
|
"""Update an existing permission document by id"""
|
|
return await self.permission_handler.update_permission(PydanticObjectId(permission_id), permission_key, permission_name, description)
|
|
|
|
async def query_permissions(self, permission_key: Optional[str] = None, permission_name: Optional[str] = None, page: int = 1, page_size: int = 10) -> Dict[str, Any]:
|
|
"""Query permissions with pagination and fuzzy search"""
|
|
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.permission_handler.query_permissions(permission_key, permission_name, skip, page_size)
|
|
return {
|
|
"items": [doc.dict() for doc in docs],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size
|
|
}
|
|
|
|
async def delete_permission(self, permission_id: str) -> None:
|
|
"""Delete a permission document after checking if it is referenced by any role"""
|
|
return await self.permission_handler.delete_permission(PydanticObjectId(permission_id)) |