feat(role_management): add delete role api

This commit is contained in:
icecheng 2025-07-21 10:47:59 +08:00
parent 4f39f888c4
commit 419e58da0c
8 changed files with 80 additions and 19 deletions

View File

@ -6,6 +6,7 @@ from backend.models.permission.models import PermissionDoc, RoleDoc
from beanie import PydanticObjectId
from datetime import datetime
class PermissionHandler:
def __init__(self):
pass
@ -39,21 +40,25 @@ class PermissionHandler:
doc = await PermissionDoc.get(permission_id)
if not doc:
raise RequestValidationError("Permission not found.")
if doc.is_default:
raise RequestValidationError("Default permission cannot be updated.")
# Check for uniqueness (exclude self)
if permission_key and permission_name:
conflict = await PermissionDoc.find_one({
"$and": [
{"_id": {"$ne": permission_id}},
{"$or": [
{str(PermissionDoc.permission_key): permission_key},
{str(PermissionDoc.permission_name): permission_name}
]}
]
})
if conflict:
raise RequestValidationError("Permission name or permission key already exists.")
doc.permission_key = permission_key
conflict = await PermissionDoc.find_one({
"$and": [
{"_id": {"$ne": permission_id}},
{"$or": [
{str(PermissionDoc.permission_key): permission_key},
{str(PermissionDoc.permission_name): permission_name}
]}
]
})
if conflict:
raise RequestValidationError("Permission name or permission key already exists.")
doc.permission_key = permission_key
doc.permission_name = permission_name
doc.description = description
doc.updated_at = datetime.now()
await doc.save()
return doc
@ -76,7 +81,7 @@ class PermissionHandler:
return docs, total
async def delete_permission(self, permission_id: PydanticObjectId) -> None:
"""Delete a permission document after checking if it is referenced by any role"""
"""Delete a permission document after checking if it is referenced by any role and is not default"""
if not permission_id:
raise RequestValidationError("permission_id is required.")
# Check if any role references this permission
@ -86,4 +91,7 @@ class PermissionHandler:
doc = await PermissionDoc.get(permission_id)
if not doc:
raise RequestValidationError("Permission not found.")
# Check if the permission is default
if doc.is_default:
raise RequestValidationError("Default permission cannot be deleted.")
await doc.delete()

View File

@ -2,7 +2,7 @@ from typing import Optional, List, Tuple
from fastapi.exceptions import RequestValidationError
from backend.models.permission.models import RoleDoc, PermissionDoc
from backend.models.permission.models import RoleDoc, PermissionDoc, UserRoleDoc
from beanie import PydanticObjectId
from datetime import datetime
@ -39,6 +39,8 @@ class RoleHandler:
doc = await RoleDoc.get(role_id)
if not doc:
raise RequestValidationError("role not found.")
if doc.is_default:
raise RequestValidationError("Default role cannot be updated.")
# Check for uniqueness (exclude self)
conflict = await RoleDoc.find_one({
"$and": [
@ -93,3 +95,19 @@ class RoleHandler:
doc.updated_at = datetime.now()
await doc.save()
return doc
async def delete_role(self, role_id: PydanticObjectId) -> None:
"""Delete a role document after checking if it is referenced by any user and is not default"""
if not role_id:
raise RequestValidationError("role_id is required.")
# Check if any user references this role
user_role = await UserRoleDoc.find_one({"role_ids": str(role_id)})
if user_role:
raise RequestValidationError("Role is referenced by a user and cannot be deleted.")
doc = await RoleDoc.get(role_id)
if not doc:
raise RequestValidationError("Role not found.")
# Check if the role is default
if doc.is_default:
raise RequestValidationError("Default role cannot be deleted.")
await doc.delete()

View File

@ -9,6 +9,7 @@ class PermissionDoc(Document):
description: Optional[str] = None # Description of the permission, optional
created_at: datetime = datetime.now() # Creation timestamp, auto-generated
updated_at: datetime = datetime.now() # Last update timestamp, auto-updated
is_default: bool = False
class Settings:
# Default collections created by Freeleaps for tenant databases use '_' prefix
@ -27,6 +28,7 @@ class RoleDoc(Document):
role_level: int
created_at: datetime = datetime.now() # Creation timestamp, auto-generated
updated_at: datetime = datetime.now() # Last update timestamp, auto-updated
is_default: bool = False
class Settings:
# Default collections created by Freeleaps for tenant databases use '_' prefix

View File

@ -37,4 +37,8 @@ class RoleService:
async def assign_permissions_to_role(self, role_id: str, permission_ids: List[str]) -> RoleDoc:
"""Assign permissions to a role by updating the permission_ids field"""
return await self.role_handler.assign_permissions_to_role(PydanticObjectId(role_id), permission_ids)
return await self.role_handler.assign_permissions_to_role(PydanticObjectId(role_id), permission_ids)
async def delete_role(self, role_id: str) -> None:
"""Delete a role document after checking if it is referenced by any user"""
return await self.role_handler.delete_role(PydanticObjectId(role_id))

View File

@ -19,6 +19,7 @@ def register(app):
permission_key=default_permission.value.permission_key,
permission_name=default_permission.value.permission_name,
description=default_permission.value.permission_description,
is_default=True,
).insert()
default_permission_ids.append(str(doc.id))
logging.info(f"default permissions initialized {default_permission_ids}")
@ -32,6 +33,7 @@ def register(app):
role_description=default_role.value.role_description,
permission_ids=default_permission_ids,
role_level=default_role.value.role_level,
is_default=True,
).insert()
default_role_ids.append(str(doc.id))
logging.info(f"default roles initialized {default_role_ids}")

View File

@ -1,3 +1,5 @@
from datetime import datetime
from fastapi import APIRouter
from pydantic import BaseModel
from typing import Optional
@ -18,8 +20,8 @@ class PermissionResponse(BaseModel):
permission_key: str
permission_name: str
description: Optional[str] = None
created_at: str
updated_at: str
created_at: datetime
updated_at: datetime
@router.post(
"/create",

View File

@ -3,10 +3,12 @@ from .create_role import router as create_role_router
from .update_role import router as update_role_router
from .query_role import router as query_role_router
from .assign_permissions import router as assign_permissions_router
from .delete_role import router as delete_role_router
router = APIRouter()
router.include_router(create_role_router, prefix="/role", tags=["role"])
router.include_router(update_role_router, prefix="/role", tags=["role"])
router.include_router(query_role_router, prefix="/role", tags=["role"])
router.include_router(assign_permissions_router, prefix="/role", tags=["role"])
router.include_router(assign_permissions_router, prefix="/role", tags=["role"])
router.include_router(delete_role_router, prefix="/role", tags=["role"])

View File

@ -0,0 +1,23 @@
from fastapi import APIRouter
from pydantic import BaseModel
from backend.services.permission.role_service import RoleService
router = APIRouter()
role_service = RoleService()
class DeleteRoleRequest(BaseModel):
role_id: str
class DeleteRoleResponse(BaseModel):
success: bool
@router.post(
"/delete",
response_model=DeleteRoleResponse,
operation_id="delete-role",
summary="Delete Role",
description="Delete a role after checking if it is referenced by any user."
)
async def delete_role(req: DeleteRoleRequest) -> DeleteRoleResponse:
await role_service.delete_role(req.role_id)
return DeleteRoleResponse(success=True)