Implement user-role management system with predefined roles (#218)

* feat: implement user-role management system with predefined roles and user-role associations

* refactor: remove RoleAlreadyExistsException handling from global exception handler

* refactor: enhance user role handling in repository and service

* refactor: streamline role management and user role retrieval

* fix(migrations): update role existence check to be case-insensitive using regex

* feat: add role removal by ID functionality and update related endpoints

* refactor: streamline user role assignment in team creation process

* fix(user_role): improve role validation to handle attribute access for role values

Author: vinit717

todo/constants/role.py

@@ -6,21 +6,24 @@ class RoleScope(Enum):
     TEAM = "TEAM"
 
 
+class RoleName(Enum):
+    MODERATOR = "moderator"
+    OWNER = "owner"
+    ADMIN = "admin"
+    MEMBER = "member"
+
+
+GLOBAL_ROLES = [RoleName.MODERATOR.value]
+TEAM_ROLES = [RoleName.OWNER.value, RoleName.ADMIN.value, RoleName.MEMBER.value]
+
+DEFAULT_TEAM_ROLE = RoleName.MEMBER.value
+
 ROLE_SCOPE_CHOICES = [
     (RoleScope.GLOBAL.value, "Global"),
     (RoleScope.TEAM.value, "Team"),
 ]
 
-GLOBAL_ROLE_NAMES = [
-    "moderator",
-]
-
-TEAM_ROLE_NAMES = [
-    "owner",
-    "admin",
-]
-
 VALID_ROLE_NAMES_BY_SCOPE = {
-    RoleScope.GLOBAL.value: GLOBAL_ROLE_NAMES,
-    RoleScope.TEAM.value: TEAM_ROLE_NAMES,
+    RoleScope.GLOBAL.value: GLOBAL_ROLES,
+    RoleScope.TEAM.value: TEAM_ROLES,
 }

todo/exceptions/global_exception_handler.py

@@ -7,7 +7,6 @@
 
 from todo.exceptions.role_exceptions import (
     RoleNotFoundException,
-    RoleAlreadyExistsException,
     RoleOperationException,
 )
 
@@ -26,9 +25,6 @@ def wrapper(*args, **kwargs):
         except RoleNotFoundException as e:
             logger.error(f"RoleNotFoundException: {e}")
             return Response({"error": str(e)}, status=status.HTTP_404_NOT_FOUND)
-        except RoleAlreadyExistsException as e:
-            logger.error(f"RoleAlreadyExistsException: {e}")
-            return Response({"error": str(e)}, status=status.HTTP_409_CONFLICT)
         except RoleOperationException as e:
             logger.error(f"RoleOperationException: {e}")
             return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@@ -52,12 +48,6 @@ def handle_role_not_found(exc: RoleNotFoundException) -> Dict[str, Any]:
         logger.error(f"Role not found: {exc}")
         return {"error": str(exc), "status_code": status.HTTP_404_NOT_FOUND}
 
-    @staticmethod
-    def handle_role_already_exists(exc: RoleAlreadyExistsException) -> Dict[str, Any]:
-        """Handle RoleAlreadyExistsException"""
-        logger.error(f"Role already exists: {exc}")
-        return {"error": str(exc), "status_code": status.HTTP_409_CONFLICT}
-
     @staticmethod
     def handle_role_operation_error(exc: RoleOperationException) -> Dict[str, Any]:
         """Handle RoleOperationException"""

todo/exceptions/role_exceptions.py

@@ -14,19 +14,6 @@ def __init__(self, role_id: str | None = None, role_name: str | None = None):
         self.role_name = role_name
 
 
-class RoleAlreadyExistsException(Exception):
-    """Exception raised when attempting to create a role that already exists."""
-
-    def __init__(self, role_name: str, existing_role_id: str | None = None):
-        message = f"Role with name '{role_name}' already exists"
-        if existing_role_id:
-            message += f" (ID: {existing_role_id})"
-
-        super().__init__(message)
-        self.role_name = role_name
-        self.existing_role_id = existing_role_id
-
-
 class RoleOperationException(Exception):
     """Exception raised when a role operation fails."""
 
@@ -42,20 +29,3 @@ def __init__(self, message: str, operation: str | None = None, role_id: str | No
         self.operation = operation
         self.role_id = role_id
         self.original_message = message
-
-
-class RoleValidationException(Exception):
-    """Exception raised when role data validation fails."""
-
-    def __init__(self, message: str, field: str | None = None, value: str | None = None):
-        if field and value:
-            full_message = f"Validation failed for field '{field}' with value '{value}': {message}"
-        elif field:
-            full_message = f"Validation failed for field '{field}': {message}"
-        else:
-            full_message = f"Role validation failed: {message}"
-
-        super().__init__(full_message)
-        self.field = field
-        self.value = value
-        self.original_message = message

todo/management/commands/migrate_roles.py

@@ -0,0 +1,16 @@
+from django.core.management.base import BaseCommand
+from todo_project.db.migrations import run_all_migrations
+
+
+class Command(BaseCommand):
+    help = "Run database migrations including predefined roles"
+
+    def handle(self, *args, **options):
+        self.stdout.write("Starting database migrations...")
+
+        success = run_all_migrations()
+
+        if success:
+            self.stdout.write("All database migrations completed successfully!")
+        else:
+            self.stdout.write("Some database migrations failed!")

todo/models/role.py

@@ -2,7 +2,7 @@
 from typing import ClassVar
 from datetime import datetime
 
-from todo.constants.role import RoleScope
+from todo.constants.role import RoleScope, RoleName
 from todo.models.common.document import Document
 from todo.models.common.pyobjectid import PyObjectId
 
@@ -11,7 +11,7 @@ class RoleModel(Document):
     collection_name: ClassVar[str] = "roles"
 
     id: PyObjectId | None = Field(None, alias="_id")
-    name: str
+    name: RoleName
     description: str | None = None
     scope: RoleScope = RoleScope.GLOBAL
     is_active: bool = True

todo/models/team.py

@@ -53,7 +53,6 @@ class UserTeamDetailsModel(Document, ObjectIdValidatorMixin):
     user_id: PyObjectId
     team_id: PyObjectId
     is_active: bool = True
-    role_id: str
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
     updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
     created_by: PyObjectId

todo/models/user_role.py

@@ -0,0 +1,45 @@
+from pydantic import Field, validator, ConfigDict
+from typing import ClassVar
+from datetime import datetime, timezone
+
+from todo.models.common.document import Document
+from todo.models.common.pyobjectid import PyObjectId
+from todo.constants.role import RoleScope, RoleName, VALID_ROLE_NAMES_BY_SCOPE
+
+
+class UserRoleModel(Document):
+    """User-role relationship model"""
+
+    collection_name: ClassVar[str] = "user_roles"
+
+    id: PyObjectId | None = Field(None, alias="_id")
+    user_id: str
+    role_name: RoleName
+    scope: RoleScope
+    team_id: str | None = None
+    is_active: bool = True
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_by: str = "system"
+
+    model_config = ConfigDict(ser_enum="value", from_attributes=True, populate_by_name=True, use_enum_values=True)
+
+    @validator("role_name")
+    def validate_role_name(cls, v, values):
+        """Validate role_name is valid for the given scope."""
+        scope = values.get("scope")
+        if scope and scope.value in VALID_ROLE_NAMES_BY_SCOPE:
+            valid_roles = VALID_ROLE_NAMES_BY_SCOPE[scope.value]
+            role_value = v.value if hasattr(v, "value") else v
+            if role_value not in valid_roles:
+                raise ValueError(f"Invalid role '{role_value}' for scope '{scope.value}'. Valid roles: {valid_roles}")
+        return v
+
+    @validator("team_id")
+    def validate_team_id(cls, v, values):
+        """Validate team_id requirements based on scope."""
+        scope = values.get("scope")
+        if scope == RoleScope.TEAM and not v:
+            raise ValueError("team_id is required for TEAM scope roles")
+        if scope == RoleScope.GLOBAL and v:
+            raise ValueError("team_id should not be provided for GLOBAL scope roles")
+        return v

todo/repositories/role_repository.py

@@ -1,14 +1,10 @@
-from bson.errors import InvalidId
-from datetime import datetime, timezone
 from typing import List, Dict, Any, Optional
 from bson import ObjectId
-from pymongo import ReturnDocument
 import logging
 
 from todo.models.role import RoleModel
 from todo.repositories.common.mongo_repository import MongoRepository
 from todo.constants.role import RoleScope
-from todo.exceptions.role_exceptions import RoleAlreadyExistsException
 
 logger = logging.getLogger(__name__)
 
@@ -50,24 +46,6 @@ def _document_to_model(cls, role_doc: dict) -> RoleModel:
 
         return RoleModel(**role_doc)
 
-    @classmethod
-    def create(cls, role: RoleModel) -> RoleModel:
-        roles_collection = cls.get_collection()
-
-        scope_value = role.scope.value if isinstance(role.scope, RoleScope) else role.scope
-        existing_role = roles_collection.find_one({"name": role.name, "scope": scope_value})
-        if existing_role:
-            raise RoleAlreadyExistsException(role.name)
-
-        role.created_at = datetime.now(timezone.utc)
-        role.updated_at = None
-
-        role_dict = role.model_dump(mode="json", by_alias=True, exclude_none=True)
-        insert_result = roles_collection.insert_one(role_dict)
-
-        role.id = insert_result.inserted_id
-        return role
-
     @classmethod
     def get_by_id(cls, role_id: str) -> Optional[RoleModel]:
         roles_collection = cls.get_collection()
@@ -76,50 +54,6 @@ def get_by_id(cls, role_id: str) -> Optional[RoleModel]:
             return cls._document_to_model(role_data)
         return None
 
-    @classmethod
-    def update(cls, role_id: str, update_data: dict) -> Optional[RoleModel]:
-        try:
-            obj_id = ObjectId(role_id)
-        except InvalidId:
-            return None
-
-        if "name" in update_data:
-            scope_value = update_data.get("scope", "GLOBAL")
-            if isinstance(scope_value, RoleScope):
-                scope_value = scope_value.value
-
-            existing_role = cls.get_by_name_and_scope(update_data["name"], scope_value)
-            if existing_role and str(existing_role.id) != role_id:
-                raise RoleAlreadyExistsException(update_data["name"])
-
-        if "scope" in update_data and isinstance(update_data["scope"], RoleScope):
-            update_data["scope"] = update_data["scope"].value
-
-        update_data["updated_at"] = datetime.now(timezone.utc)
-
-        update_data.pop("_id", None)
-        update_data.pop("id", None)
-
-        roles_collection = cls.get_collection()
-        updated_role_doc = roles_collection.find_one_and_update(
-            {"_id": obj_id}, {"$set": update_data}, return_document=ReturnDocument.AFTER
-        )
-
-        if updated_role_doc:
-            return cls._document_to_model(updated_role_doc)
-        return None
-
-    @classmethod
-    def delete_by_id(cls, role_id: str) -> bool:
-        try:
-            obj_id = ObjectId(role_id)
-        except Exception:
-            return False
-
-        roles_collection = cls.get_collection()
-        result = roles_collection.delete_one({"_id": obj_id})
-        return result.deleted_count > 0
-
     @classmethod
     def get_by_name(cls, name: str) -> Optional[RoleModel]:
         roles_collection = cls.get_collection()

todo/repositories/user_role_repository.py

@@ -0,0 +1,95 @@
+from datetime import datetime, timezone
+from typing import List, Optional
+import logging
+from bson import ObjectId
+
+from todo.models.user_role import UserRoleModel
+from todo.repositories.common.mongo_repository import MongoRepository
+from todo.constants.role import RoleScope, RoleName
+
+logger = logging.getLogger(__name__)
+
+
+class UserRoleRepository(MongoRepository):
+    collection_name = UserRoleModel.collection_name
+
+    @classmethod
+    def create(cls, user_role: UserRoleModel) -> UserRoleModel:
+        collection = cls.get_collection()
+
+        role_name_value = user_role.role_name.value if hasattr(user_role.role_name, "value") else user_role.role_name
+        scope_value = user_role.scope.value if hasattr(user_role.scope, "value") else user_role.scope
+
+        # Check if already exists and is active
+        existing = collection.find_one(
+            {
+                "user_id": user_role.user_id,
+                "role_name": role_name_value,
+                "scope": scope_value,
+                "team_id": user_role.team_id,
+                "is_active": True,
+            }
+        )
+
+        if existing:
+            return UserRoleModel(**existing)
+
+        user_role.created_at = datetime.now(timezone.utc)
+        user_role_dict = user_role.model_dump(mode="json", by_alias=True, exclude_none=True)
+        result = collection.insert_one(user_role_dict)
+        user_role.id = result.inserted_id
+        return user_role
+
+    @classmethod
+    def get_user_roles(
+        cls, user_id: Optional[str] = None, scope: Optional["RoleScope"] = None, team_id: Optional[str] = None
+    ) -> List[UserRoleModel]:
+        collection = cls.get_collection()
+
+        query = {"is_active": True}
+
+        if user_id:
+            query["user_id"] = user_id
+
+        if scope:
+            scope_value = scope.value if hasattr(scope, "value") else scope
+            query["scope"] = scope_value
+
+        if team_id:
+            query["team_id"] = team_id
+        elif scope and (scope.value if hasattr(scope, "value") else scope) == "GLOBAL":
+            query["team_id"] = None
+
+        roles = []
+        for doc in collection.find(query):
+            roles.append(UserRoleModel(**doc))
+        return roles
+
+    @classmethod
+    def assign_role(
+        cls, user_id: str, role_name: "RoleName", scope: "RoleScope", team_id: Optional[str] = None
+    ) -> UserRoleModel:
+        """Assign a role to a user - simple and clean."""
+        user_role = UserRoleModel(user_id=user_id, role_name=role_name, scope=scope, team_id=team_id, is_active=True)
+        return cls.create(user_role)
+
+    @classmethod
+    def remove_role_by_id(cls, user_id: str, role_id: str, scope: str, team_id: Optional[str] = None) -> bool:
+        """Remove a role from a user by role_id - simple deactivation."""
+        collection = cls.get_collection()
+
+        try:
+            object_id = ObjectId(role_id)
+        except Exception:
+            return False
+
+        query = {"_id": object_id, "user_id": user_id, "scope": scope, "is_active": True}
+
+        if scope == "TEAM" and team_id:
+            query["team_id"] = team_id
+        elif scope == "GLOBAL":
+            query["team_id"] = None
+
+        result = collection.update_one(query, {"$set": {"is_active": False}})
+
+        return result.modified_count > 0