from beanie import Document, PydanticObjectId from pydantic import Field from typing import Optional, List, Dict, Any from datetime import datetime from enum import Enum import logging # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("account_service") class RoleType(str, Enum): SYSTEM = "system" CUSTOM = "custom" DEPARTMENT = "department" PROJECT = "project" class Role(Document): # Role Identity # id: str name: str = Field(..., min_length=2, max_length=50) # e.g., "admin", "moderator", "user" display_name: str # Human-readable name description: Optional[str] = None # Role Properties role_type: RoleType = RoleType.CUSTOM is_active: bool = True is_default: bool = False # Assigned to new users is_system_role: bool = False # Cannot be deleted # Permissions permissions: List[str] = Field(default_factory=list) # Permission names # Hierarchy parent_role: Optional[str] = None # Reference to parent role child_roles: List[str] = Field(default_factory=list) # Constraints max_users: Optional[int] = None # Maximum users that can have this role expires_at: Optional[datetime] = None # Role expiration # Metadata created_at: datetime = Field(default_factory=datetime.utcnow) updated_at: datetime = Field(default_factory=datetime.utcnow) created_by: Optional[PydanticObjectId] = None # User ID who created this role # Additional Properties color: Optional[str] = None # UI color for the role priority: int = 0 # Role priority for conflict resolution custom_properties: Dict[str, Any] = Field(default_factory=dict) class Config: arbitrary_types_allowed = True class Settings: name = "roles" indexes = [ "name", # Simple index on name field, not unique to avoid conflicts "role_type", "is_active", "is_default", "is_system_role", "created_at", "priority" ] @classmethod async def create_system_roles(cls): """Create default system roles""" try: logger.info("Starting system roles creation...") system_roles = [ { "name": "super_admin", "display_name": "Super Administrator", "description": "Full system access with all permissions", "role_type": RoleType.SYSTEM, "is_system_role": True, "priority": 1000, "color": "#FF0000", "permissions": [ "system.admin", "user.admin", "profile.read.all", "profile.write.all", "analytics.read", "system.settings.read", "system.settings.write" ] }, { "name": "admin", "display_name": "Administrator", "description": "Administrative access to user management", "role_type": RoleType.SYSTEM, "is_system_role": True, "priority": 900, "color": "#FFA500", "permissions": [ "user.read", "user.write", "profile.read.all", "profile.write.all", "analytics.read" ] }, { "name": "moderator", "display_name": "Moderator", "description": "Content moderation and user management", "role_type": RoleType.SYSTEM, "is_system_role": True, "priority": 800, "color": "#00FF00", "permissions": [ "user.read", "profile.read.all" ] }, { "name": "user", "display_name": "Regular User", "description": "Standard user with basic permissions", "role_type": RoleType.SYSTEM, "is_system_role": True, "is_default": True, "priority": 100, "color": "#0000FF", "permissions": [ "profile.read.own", "profile.write.own" ] }, { "name": "guest", "display_name": "Guest", "description": "Limited access for guest users", "role_type": RoleType.SYSTEM, "is_system_role": True, "priority": 50, "color": "#808080", "permissions": [] } ] logger.info(f"System roles to create: {len(system_roles)}") created_count = 0 for role_data in system_roles: try: logger.info(f"Checking if role '{role_data['name']}' exists...") existing = await cls.find_one(cls.name == role_data["name"]) if not existing: logger.info(f"Creating role: {role_data['name']}") role = cls(**role_data) saved_role = await role.save() logger.info(f"Successfully created role: {role_data['name']} with ID: {saved_role.id}") created_count += 1 else: logger.info(f"Role '{role_data['name']}' already exists with ID: {existing.id}") except Exception as role_error: logger.error(f"Error creating role '{role_data['name']}': {role_error}") raise role_error # Verify roles were created total_roles = await cls.find().count() system_roles_count = await cls.find(cls.is_system_role == True).count() logger.info(f"System roles creation completed. Created: {created_count}, Total roles in DB: {total_roles}, System roles: {system_roles_count}") # List all roles for verification all_roles = await cls.find().to_list() logger.info(f"All roles in database: {[role.name for role in all_roles]}") return True except Exception as e: logger.error(f"Failed to create system roles: {e}") raise e async def get_all_permissions(self) -> List[str]: """Get all permissions including inherited from parent roles""" all_permissions = set(self.permissions) # Add permissions from parent roles if self.parent_role: parent = await Role.find_one(Role.name == self.parent_role) if parent: parent_permissions = await parent.get_all_permissions() all_permissions.update(parent_permissions) return list(all_permissions) async def add_permission(self, permission_name: str): """Add a permission to this role""" if permission_name not in self.permissions: self.permissions.append(permission_name) self.updated_at = datetime.utcnow() await self.save() async def remove_permission(self, permission_name: str): """Remove a permission from this role""" if permission_name in self.permissions: self.permissions.remove(permission_name) self.updated_at = datetime.utcnow() await self.save() async def get_user_count(self) -> int: """Get number of users with this role""" from app.models.user_role import UserRole return await UserRole.find( UserRole.role_name == self.name, UserRole.is_active == True ).count() async def can_assign_to_user(self) -> bool: """Check if role can be assigned to more users""" if not self.max_users: return True current_count = await self.get_user_count() return current_count < self.max_users def has_permission(self, permission_name: str) -> bool: """Check if role has a specific permission""" return permission_name in self.permissions async def get_child_roles(self) -> List["Role"]: """Get all child roles""" if not self.child_roles: return [] return await Role.find( {"name": {"$in": self.child_roles}}, Role.is_active == True ).to_list() async def get_parent_role(self) -> Optional["Role"]: """Get parent role if exists""" if not self.parent_role: return None return await Role.find_one(Role.name == self.parent_role) @classmethod async def get_default_role(cls) -> Optional["Role"]: """Get the default role for new users""" return await cls.find_one( cls.is_default == True, cls.is_active == True )