41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
from typing import Optional, List
|
|
from fastapi.exceptions import RequestValidationError
|
|
from backend.models.permission.models import RoleDoc, UserRoleDoc
|
|
from beanie import PydanticObjectId
|
|
|
|
|
|
class UserRoleHandler:
|
|
def __init__(self):
|
|
pass
|
|
|
|
async def assign_roles_to_user(self, user_id: str, role_ids: List[str]) -> Optional[UserRoleDoc]:
|
|
"""Assign roles to a user by updating or creating the UserRoleDoc"""
|
|
if not user_id or not role_ids:
|
|
raise RequestValidationError("user_id and role_ids are required.")
|
|
|
|
# Validate that all role_ids exist in the role collection
|
|
for role_id in role_ids:
|
|
role_doc = await RoleDoc.get(PydanticObjectId(role_id))
|
|
if not role_doc:
|
|
raise RequestValidationError(f"Role with id {role_id} not found.")
|
|
|
|
# Remove duplicates from role_ids
|
|
unique_role_ids = list(dict.fromkeys(role_ids))
|
|
|
|
# Check if UserRoleDoc already exists for this user
|
|
existing_user_role = await UserRoleDoc.find_one(UserRoleDoc.user_id == user_id)
|
|
|
|
if existing_user_role:
|
|
# Update existing UserRoleDoc
|
|
existing_user_role.role_ids = unique_role_ids
|
|
await existing_user_role.save()
|
|
return existing_user_role
|
|
else:
|
|
# Create new UserRoleDoc
|
|
user_role_doc = UserRoleDoc(
|
|
user_id=user_id,
|
|
role_ids=unique_role_ids
|
|
)
|
|
await user_role_doc.insert()
|
|
return user_role_doc
|