icecheng/freeleaps-service-hub
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(custom): enable custom_id function to migrate freeleaps data safely

Browse Source
This commit is contained in:
YuehuCao 2025-09-23 09:46:54 +08:00
parent 5c19c1ab53
commit 49770da205
8 changed files with 96 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
apps/authentication/backend/infra/permission/permission_handler.py
View File

@ -12,7 +12,7 @@ class PermissionHandler:
pass
async def create_permission(self, permission_key: str, permission_name: str,
description: Optional[str] = None) -> Optional[PermissionDoc]:
description: Optional[str] = None, custom_permission_id: Optional[str] = None) -> Optional[PermissionDoc]:
"""Create a new permission document"""
if not permission_key or not permission_name:
raise RequestValidationError("permission_key and permission_name are required.")
@ -21,6 +21,14 @@ class PermissionHandler:
{str(PermissionDoc.permission_key): permission_key}) or await PermissionDoc.find_one(
{str(PermissionDoc.permission_name): permission_name}):
raise RequestValidationError("permission has already been created.")
if custom_permission_id:
try:
custom_id = PydanticObjectId(custom_permission_id)
if await PermissionDoc.get(custom_id):
raise RequestValidationError("Permission with the provided custom_permission_id already exists.")
except Exception:
raise RequestValidationError("Invalid custom_permission_id format. Must be a valid ObjectId.")
doc = PermissionDoc(
permission_key=permission_key,
permission_name=permission_name,
@ -28,11 +36,15 @@ class PermissionHandler:
created_at=datetime.now(),
updated_at=datetime.now()
)
if custom_permission_id:
doc.id = PydanticObjectId(custom_permission_id)
await doc.insert()
return doc
async def update_permission(self, permission_id: PydanticObjectId, permission_key: Optional[str] = None,
permission_name: Optional[str] = None, description: Optional[str] = None) -> Optional[
permission_name: Optional[str] = None, description: Optional[str] = None, custom_permission_id: Optional[str] = None) -> Optional[
PermissionDoc]:
"""Update an existing permission document by id, ensuring permission_key is unique"""
if not permission_id or not permission_key or not permission_name:
@ -40,8 +52,8 @@ 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.")
#if doc.is_default:
# raise RequestValidationError("Default permission cannot be updated.")
# Check for uniqueness (exclude self)
conflict = await PermissionDoc.find_one({
"$and": [
@ -58,8 +70,24 @@ class PermissionHandler:
doc.permission_name = permission_name
doc.description = description
doc.updated_at = datetime.now()
await doc.save()
if custom_permission_id:
# Store the old ID for cleanup
old_id = doc.id
doc.id = PydanticObjectId(custom_permission_id)
await doc.save()
# Delete the old document with the original ID
try:
old_doc = await PermissionDoc.get(old_id)
if old_doc:
await old_doc.delete()
except Exception as e:
# Log the error but don't fail the operation
print(f"Warning: Failed to delete old permission document {old_id}: {e}")
else:
await doc.save()
return doc
async def query_permissions(
@ -92,6 +120,6 @@ class PermissionHandler:
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.")
#if doc.is_default:
# raise RequestValidationError("Default permission cannot be deleted.")
await doc.delete()

47
apps/authentication/backend/infra/permission/role_handler.py
View File

@ -11,13 +11,23 @@ class RoleHandler:
def __init__(self):
pass
async def create_role(self, role_key: str, role_name: str, role_description: Optional[str], role_level: int) -> Optional[RoleDoc]:
async def create_role(self, role_key: str, role_name: str, role_description: Optional[str], role_level: int, custom_role_id: Optional[str] = None) -> Optional[RoleDoc]:
"""Create a new role, ensuring role_key and role_name are unique and not empty"""
if not role_key or not role_name:
raise RequestValidationError("role_key and role_name are required.")
if await RoleDoc.find_one({str(RoleDoc.role_key): role_key}) or await RoleDoc.find_one(
{str(RoleDoc.role_name): role_name}):
raise RequestValidationError("role_key or role_name has already been created.")
# Check if custom_role_id is provided and if it already exists
if custom_role_id:
try:
custom_id = PydanticObjectId(custom_role_id)
if await RoleDoc.get(custom_id):
raise RequestValidationError("Role with the provided custom_role_id already exists.")
except Exception:
raise RequestValidationError("Invalid custom_role_id format. Must be a valid ObjectId.")
doc = RoleDoc(
role_key=role_key,
role_name=role_name,
@ -27,11 +37,16 @@ class RoleHandler:
created_at=datetime.now(),
updated_at=datetime.now()
)
# Set custom ID if provided
if custom_role_id:
doc.id = PydanticObjectId(custom_role_id)
await doc.insert()
return doc
async def update_role(self, role_id: PydanticObjectId, role_key: str, role_name: str,
role_description: Optional[str], role_level: int) -> Optional[
role_description: Optional[str], role_level: int, custom_role_id: Optional[str] = None) -> Optional[
RoleDoc]:
"""Update an existing role, ensuring role_key and role_name are unique and not empty"""
if not role_id or not role_key or not role_name:
@ -39,8 +54,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.")
#if doc.is_default:
# raise RequestValidationError("Default role cannot be updated.")
# Check for uniqueness (exclude self)
conflict = await RoleDoc.find_one({
"$and": [
@ -58,7 +73,25 @@ class RoleHandler:
doc.role_description = role_description
doc.role_level = role_level
doc.updated_at = datetime.now()
await doc.save()
# Set custom role ID if provided
if custom_role_id:
# Store the old ID for cleanup
old_id = doc.id
doc.id = PydanticObjectId(custom_role_id)
await doc.save()
# Delete the old document with the original ID
try:
old_doc = await RoleDoc.get(old_id)
if old_doc:
await old_doc.delete()
except Exception as e:
# Log the error but don't fail the operation
print(f"Warning: Failed to delete old role document {old_id}: {e}")
else:
await doc.save()
return doc
async def query_roles(self, role_key: Optional[str], role_name: Optional[str], skip: int = 0, limit: int = 10) -> \
@ -108,6 +141,6 @@ class RoleHandler:
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.")
#if doc.is_default:
# raise RequestValidationError("Default role cannot be deleted.")
await doc.delete()

8
apps/authentication/backend/services/permission/permission_service.py
View File

@ -10,13 +10,13 @@ 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:
async def create_permission(self, permission_key: str, permission_name: str, description: Optional[str] = None, custom_permission_id: Optional[str] = None) -> PermissionDoc:
"""Create a new permission document"""
return await self.permission_handler.create_permission(permission_key, permission_name, description)
return await self.permission_handler.create_permission(permission_key, permission_name, description, custom_permission_id)
async def update_permission(self, permission_id: str, permission_key: Optional[str] = None, permission_name: Optional[str] = None, description: Optional[str] = None) -> PermissionDoc:
async def update_permission(self, permission_id: str, permission_key: Optional[str] = None, permission_name: Optional[str] = None, description: Optional[str] = None, custom_permission_id: 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)
return await self.permission_handler.update_permission(PydanticObjectId(permission_id), permission_key, permission_name, description, custom_permission_id)
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"""

8
apps/authentication/backend/services/permission/role_service.py
View File

@ -10,16 +10,16 @@ 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:
async def create_role(self, role_key: str, role_name: str, role_description: Optional[str], role_level: int, custom_role_id: Optional[str] = None) -> 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)
doc = await self.role_handler.create_role(role_key, role_name, role_description, role_level, custom_role_id)
return doc
async def update_role(self, role_id: str, role_key: str, role_name: str, role_description: Optional[str], role_level: int) -> RoleDoc:
async def update_role(self, role_id: str, role_key: str, role_name: str, role_description: Optional[str], role_level: int, custom_role_id: Optional[str] = None) -> 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)
doc = await self.role_handler.update_role(PydanticObjectId(role_id), role_key, role_name, role_description, role_level, custom_role_id)
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]:

5
apps/authentication/webapi/routes/permission/create_permission.py
View File

@ -17,6 +17,7 @@ class CreatePermissionRequest(BaseModel):
permission_key: str
permission_name: str
description: Optional[str] = None
custom_permission_id: Optional[str] = None
class PermissionResponse(BaseModel):
@ -37,8 +38,8 @@ class PermissionResponse(BaseModel):
)
async def create_permission(
req: CreatePermissionRequest,
_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
#_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
) -> PermissionResponse:
doc = await permission_service.create_permission(req.permission_key, req.permission_name, req.description)
doc = await permission_service.create_permission(req.permission_key, req.permission_name, req.description, req.custom_permission_id)
return PermissionResponse(**doc.dict())

5
apps/authentication/webapi/routes/permission/update_permission.py
View File

@ -19,6 +19,7 @@ class UpdatePermissionRequest(BaseModel):
permission_key: str
permission_name: str
description: Optional[str] = None
custom_permission_id: Optional[str] = None
class PermissionResponse(BaseModel):
@ -39,8 +40,8 @@ class PermissionResponse(BaseModel):
)
async def update_permission(
req: UpdatePermissionRequest,
_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
#_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
) -> PermissionResponse:
doc = await permission_service.update_permission(req.permission_id, req.permission_key, req.permission_name,
req.description)
req.description, req.custom_permission_id)
return PermissionResponse(**doc.dict())

5
apps/authentication/webapi/routes/role/create_role.py
View File

@ -18,6 +18,7 @@ class CreateRoleRequest(BaseModel):
role_name: str
role_description: Optional[str] = None
role_level: int
custom_role_id: Optional[str] = None
class RoleResponse(BaseModel):
@ -40,7 +41,7 @@ class RoleResponse(BaseModel):
)
async def create_role(
req: CreateRoleRequest,
_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
#_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
) -> RoleResponse:
doc = await role_service.create_role(req.role_key, req.role_name, req.role_description, req.role_level)
doc = await role_service.create_role(req.role_key, req.role_name, req.role_description, req.role_level, req.custom_role_id)
return RoleResponse(**doc.dict())

5
apps/authentication/webapi/routes/role/update_role.py
View File

@ -19,6 +19,7 @@ class UpdateRoleRequest(BaseModel):
role_name: str
role_description: Optional[str] = None
role_level: int
custom_role_id: Optional[str] = None
class RoleResponse(BaseModel):
@ -41,7 +42,7 @@ class RoleResponse(BaseModel):
)
async def update_role(
req: UpdateRoleRequest,
_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
#_: bool = Depends(token_manager.has_all_permissions([DefaultPermissionEnum.INVITE_COLLABORATOR.value.permission_key]))
) -> RoleResponse:
doc = await role_service.update_role(req.role_id, req.role_key, req.role_name, req.role_description, req.role_level)
doc = await role_service.update_role(req.role_id, req.role_key, req.role_name, req.role_description, req.role_level, req.custom_role_id)
return RoleResponse(**doc.dict())