768 lines
23 KiB
Python
768 lines
23 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Admin RBAC Roles Management routes.
|
|
Provides endpoints for managing roles and role assignments to users.
|
|
|
|
MULTI-TENANT: These are SYSTEM-LEVEL operations requiring isSysAdmin=true.
|
|
Roles are global system resources, not mandate-specific.
|
|
Role assignments are managed via UserMandateRole (not User.roleLabels).
|
|
"""
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Query, Body, Path, Request
|
|
from typing import List, Dict, Any, Optional, Set
|
|
import logging
|
|
|
|
from modules.auth import limiter, requireSysAdmin
|
|
from modules.datamodels.datamodelUam import User, UserInDB
|
|
from modules.datamodels.datamodelRbac import Role
|
|
from modules.datamodels.datamodelMembership import UserMandate, UserMandateRole
|
|
from modules.interfaces.interfaceDbApp import getInterface, getRootInterface
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _getUserRoleLabels(interface, userId: str) -> List[str]:
|
|
"""
|
|
Get role labels for a user from UserMandateRole (across all mandates).
|
|
|
|
Args:
|
|
interface: Database interface
|
|
userId: User ID
|
|
|
|
Returns:
|
|
List of role labels
|
|
"""
|
|
roleLabels: Set[str] = set()
|
|
|
|
# Get all UserMandate records for this user (Pydantic models)
|
|
userMandates = interface.getUserMandates(userId)
|
|
|
|
for um in userMandates:
|
|
# Get all UserMandateRole records for this membership (Pydantic models)
|
|
userMandateRoles = interface.getUserMandateRoles(str(um.id))
|
|
|
|
for umr in userMandateRoles:
|
|
if umr.roleId:
|
|
# Get role by ID to get roleLabel
|
|
role = interface.getRole(str(umr.roleId))
|
|
if role:
|
|
roleLabels.add(role.roleLabel)
|
|
|
|
return list(roleLabels)
|
|
|
|
|
|
def _hasRoleLabel(interface, userId: str, roleLabel: str) -> bool:
|
|
"""
|
|
Check if user has a specific role label (across all mandates).
|
|
"""
|
|
return roleLabel in _getUserRoleLabels(interface, userId)
|
|
|
|
router = APIRouter(
|
|
prefix="/api/admin/rbac/roles",
|
|
tags=["Admin RBAC Roles"],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("60/minute")
|
|
def list_roles(
|
|
request: Request,
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get list of all available roles with metadata.
|
|
MULTI-TENANT: SysAdmin-only (roles are system resources).
|
|
|
|
Returns:
|
|
- List of role dictionaries with role label, description, and user count
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get all roles from database
|
|
dbRoles = interface.getAllRoles()
|
|
|
|
# Count role assignments from UserMandateRole table
|
|
roleCounts = interface.countRoleAssignments()
|
|
|
|
# Convert Role objects to dictionaries and add user counts
|
|
result = []
|
|
for role in dbRoles:
|
|
result.append({
|
|
"id": role.id,
|
|
"roleLabel": role.roleLabel,
|
|
"description": role.description,
|
|
"userCount": roleCounts.get(str(role.id), 0),
|
|
"isSystemRole": role.isSystemRole
|
|
})
|
|
|
|
return result
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error listing roles: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to list roles: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/options", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("60/minute")
|
|
def get_role_options(
|
|
request: Request,
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get role options for select dropdowns.
|
|
MULTI-TENANT: SysAdmin-only.
|
|
|
|
Returns:
|
|
- List of role option dictionaries with value and label
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get all roles from database
|
|
dbRoles = interface.getAllRoles()
|
|
|
|
# Convert to options format
|
|
options = []
|
|
for role in dbRoles:
|
|
# Use English description as label, fallback to roleLabel
|
|
label = role.description.get("en", role.roleLabel) if isinstance(role.description, dict) else role.roleLabel
|
|
options.append({
|
|
"value": role.roleLabel,
|
|
"label": label
|
|
})
|
|
|
|
return options
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting role options: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to get role options: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.post("/", response_model=Dict[str, Any])
|
|
@limiter.limit("30/minute")
|
|
def create_role(
|
|
request: Request,
|
|
role: Role = Body(...),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a new role.
|
|
MULTI-TENANT: SysAdmin-only (roles are system resources).
|
|
|
|
Request Body:
|
|
- role: Role object to create
|
|
|
|
Returns:
|
|
- Created role dictionary
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
createdRole = interface.createRole(role)
|
|
|
|
return {
|
|
"id": createdRole.id,
|
|
"roleLabel": createdRole.roleLabel,
|
|
"description": createdRole.description,
|
|
"isSystemRole": createdRole.isSystemRole
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=str(e)
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error creating role: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to create role: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/{roleId}", response_model=Dict[str, Any])
|
|
@limiter.limit("60/minute")
|
|
def get_role(
|
|
request: Request,
|
|
roleId: str = Path(..., description="Role ID"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get a role by ID.
|
|
MULTI-TENANT: SysAdmin-only.
|
|
|
|
Path Parameters:
|
|
- roleId: Role ID
|
|
|
|
Returns:
|
|
- Role dictionary
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
role = interface.getRole(roleId)
|
|
if not role:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Role {roleId} not found"
|
|
)
|
|
|
|
return {
|
|
"id": role.id,
|
|
"roleLabel": role.roleLabel,
|
|
"description": role.description,
|
|
"isSystemRole": role.isSystemRole
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting role: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to get role: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.put("/{roleId}", response_model=Dict[str, Any])
|
|
@limiter.limit("30/minute")
|
|
def update_role(
|
|
request: Request,
|
|
roleId: str = Path(..., description="Role ID"),
|
|
role: Role = Body(...),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Update an existing role.
|
|
MULTI-TENANT: SysAdmin-only.
|
|
|
|
Path Parameters:
|
|
- roleId: Role ID
|
|
|
|
Request Body:
|
|
- role: Updated Role object
|
|
|
|
Returns:
|
|
- Updated role dictionary
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
updatedRole = interface.updateRole(roleId, role)
|
|
|
|
return {
|
|
"id": updatedRole.id,
|
|
"roleLabel": updatedRole.roleLabel,
|
|
"description": updatedRole.description,
|
|
"isSystemRole": updatedRole.isSystemRole
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=str(e)
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error updating role: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to update role: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.delete("/{roleId}", response_model=Dict[str, str])
|
|
@limiter.limit("30/minute")
|
|
def delete_role(
|
|
request: Request,
|
|
roleId: str = Path(..., description="Role ID"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, str]:
|
|
"""
|
|
Delete a role.
|
|
MULTI-TENANT: SysAdmin-only.
|
|
|
|
Path Parameters:
|
|
- roleId: Role ID
|
|
|
|
Returns:
|
|
- Success message
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
success = interface.deleteRole(roleId)
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Role {roleId} not found"
|
|
)
|
|
|
|
return {"message": f"Role {roleId} deleted successfully"}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=str(e)
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error deleting role: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to delete role: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/users", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("60/minute")
|
|
def list_users_with_roles(
|
|
request: Request,
|
|
roleLabel: Optional[str] = Query(None, description="Filter by role label"),
|
|
mandateId: Optional[str] = Query(None, description="Filter by mandate ID (via UserMandate)"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get list of users with their role assignments.
|
|
MULTI-TENANT: SysAdmin-only, can see all users across mandates.
|
|
|
|
Query Parameters:
|
|
- roleLabel: Optional filter by role label
|
|
- mandateId: Optional filter by mandate ID (via UserMandate table)
|
|
|
|
Returns:
|
|
- List of user dictionaries with role assignments
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get all users via interface method (Pydantic models)
|
|
users = interface.getAllUsers()
|
|
|
|
# Filter by mandate if specified (via UserMandate table)
|
|
if mandateId:
|
|
userMandates = interface.getUserMandatesByMandate(mandateId)
|
|
mandateUserIds = {str(um.userId) for um in userMandates}
|
|
users = [u for u in users if str(u.id) in mandateUserIds]
|
|
|
|
# Filter by role if specified (via UserMandateRole)
|
|
if roleLabel:
|
|
users = [u for u in users if _hasRoleLabel(interface, str(u.id), roleLabel)]
|
|
|
|
# Format response
|
|
result = []
|
|
for user in users:
|
|
userRoleLabels = _getUserRoleLabels(interface, str(user.id))
|
|
result.append({
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"fullName": user.fullName,
|
|
"isSysAdmin": user.isSysAdmin,
|
|
"enabled": user.enabled,
|
|
"roleLabels": userRoleLabels,
|
|
"roleCount": len(userRoleLabels)
|
|
})
|
|
|
|
return result
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error listing users with roles: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to list users with roles: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/users/{userId}", response_model=Dict[str, Any])
|
|
@limiter.limit("60/minute")
|
|
def get_user_roles(
|
|
request: Request,
|
|
userId: str = Path(..., description="User ID"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get role assignments for a specific user.
|
|
MULTI-TENANT: SysAdmin-only.
|
|
|
|
Path Parameters:
|
|
- userId: User ID
|
|
|
|
Returns:
|
|
- User dictionary with role assignments
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get user
|
|
user = interface.getUser(userId)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"User {userId} not found"
|
|
)
|
|
|
|
userRoleLabels = _getUserRoleLabels(interface, str(user.id))
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"fullName": user.fullName,
|
|
"isSysAdmin": user.isSysAdmin,
|
|
"enabled": user.enabled,
|
|
"roleLabels": userRoleLabels,
|
|
"roleCount": len(userRoleLabels)
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting user roles: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to get user roles: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.put("/users/{userId}/roles", response_model=Dict[str, Any])
|
|
@limiter.limit("30/minute")
|
|
def update_user_roles(
|
|
request: Request,
|
|
userId: str = Path(..., description="User ID"),
|
|
newRoleLabels: List[str] = Body(..., description="List of role labels to assign"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Update role assignments for a specific user.
|
|
MULTI-TENANT: SysAdmin-only. Updates roles in user's first mandate.
|
|
|
|
Path Parameters:
|
|
- userId: User ID
|
|
|
|
Request Body:
|
|
- newRoleLabels: List of role labels to assign (e.g., ["admin", "user"])
|
|
|
|
Returns:
|
|
- Updated user dictionary with role assignments
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get user
|
|
user = interface.getUser(userId)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"User {userId} not found"
|
|
)
|
|
|
|
# Validate role labels (basic validation - check against standard roles)
|
|
standardRoles = ["sysadmin", "admin", "user", "viewer"]
|
|
for roleLabel in newRoleLabels:
|
|
if roleLabel not in standardRoles:
|
|
logger.warning(f"Non-standard role label assigned: {roleLabel}")
|
|
|
|
# Get user's first mandate (for role assignment)
|
|
userMandates = interface.getUserMandates(userId)
|
|
if not userMandates:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"User {userId} has no mandate memberships. Add to mandate first."
|
|
)
|
|
|
|
userMandateId = str(userMandates[0].id)
|
|
|
|
# Get current roles for this mandate (Pydantic models)
|
|
existingRoles = interface.getUserMandateRoles(userMandateId)
|
|
existingRoleIds = {str(r.roleId) for r in existingRoles}
|
|
|
|
# Convert roleLabels to roleIds
|
|
newRoleIds = set()
|
|
for roleLabel in newRoleLabels:
|
|
role = interface.getRoleByLabel(roleLabel)
|
|
if role:
|
|
newRoleIds.add(str(role.id))
|
|
|
|
# Remove roles that are no longer needed
|
|
for existingRole in existingRoles:
|
|
if str(existingRole.roleId) not in newRoleIds:
|
|
interface.removeRoleFromUserMandate(userMandateId, str(existingRole.roleId))
|
|
|
|
# Add new roles
|
|
for roleId in newRoleIds:
|
|
if roleId not in existingRoleIds:
|
|
newRole = UserMandateRole(userMandateId=userMandateId, roleId=roleId)
|
|
interface.db.recordCreate(UserMandateRole, newRole.model_dump())
|
|
|
|
logger.info(f"Updated roles for user {userId}: {newRoleLabels} by SysAdmin {currentUser.id}")
|
|
|
|
userRoleLabels = _getUserRoleLabels(interface, userId)
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"fullName": user.fullName,
|
|
"isSysAdmin": user.isSysAdmin,
|
|
"enabled": user.enabled,
|
|
"roleLabels": userRoleLabels,
|
|
"roleCount": len(userRoleLabels)
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error updating user roles: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to update user roles: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.post("/users/{userId}/roles/{roleLabel}", response_model=Dict[str, Any])
|
|
@limiter.limit("30/minute")
|
|
def add_user_role(
|
|
request: Request,
|
|
userId: str = Path(..., description="User ID"),
|
|
roleLabel: str = Path(..., description="Role label to add"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Add a role to a user (if not already assigned).
|
|
MULTI-TENANT: SysAdmin-only. Adds role to user's first mandate.
|
|
|
|
Path Parameters:
|
|
- userId: User ID
|
|
- roleLabel: Role label to add
|
|
|
|
Returns:
|
|
- Updated user dictionary with role assignments
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get user
|
|
user = interface.getUser(userId)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"User {userId} not found"
|
|
)
|
|
|
|
# Get role by label
|
|
role = interface.getRoleByLabel(roleLabel)
|
|
if not role:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Role '{roleLabel}' not found"
|
|
)
|
|
|
|
# Get user's first mandate
|
|
userMandates = interface.getUserMandates(userId)
|
|
if not userMandates:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"User {userId} has no mandate memberships. Add to mandate first."
|
|
)
|
|
|
|
userMandateId = str(userMandates[0].id)
|
|
|
|
# Check if role is already assigned - use interface method
|
|
existingRoles = interface.getUserMandateRoles(userMandateId)
|
|
roleAlreadyAssigned = any(str(r.roleId) == str(role.id) for r in existingRoles)
|
|
|
|
if not roleAlreadyAssigned:
|
|
# Add the role via interface method
|
|
interface.addRoleToUserMandate(userMandateId, str(role.id))
|
|
logger.info(f"Added role {roleLabel} to user {userId} by SysAdmin {currentUser.id}")
|
|
|
|
userRoleLabels = _getUserRoleLabels(interface, userId)
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"fullName": user.fullName,
|
|
"isSysAdmin": user.isSysAdmin,
|
|
"enabled": user.enabled,
|
|
"roleLabels": userRoleLabels,
|
|
"roleCount": len(userRoleLabels)
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error adding role to user: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to add role to user: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.delete("/users/{userId}/roles/{roleLabel}", response_model=Dict[str, Any])
|
|
@limiter.limit("30/minute")
|
|
def remove_user_role(
|
|
request: Request,
|
|
userId: str = Path(..., description="User ID"),
|
|
roleLabel: str = Path(..., description="Role label to remove"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Remove a role from a user.
|
|
MULTI-TENANT: SysAdmin-only. Removes role from all user's mandates.
|
|
|
|
Path Parameters:
|
|
- userId: User ID
|
|
- roleLabel: Role label to remove
|
|
|
|
Returns:
|
|
- Updated user dictionary with role assignments
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get user
|
|
user = interface.getUser(userId)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"User {userId} not found"
|
|
)
|
|
|
|
# Get role by label
|
|
role = interface.getRoleByLabel(roleLabel)
|
|
if not role:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Role '{roleLabel}' not found"
|
|
)
|
|
|
|
# Remove role from all user's mandates
|
|
userMandates = interface.getUserMandates(userId)
|
|
roleRemoved = False
|
|
|
|
for um in userMandates:
|
|
userMandateId = str(um.id)
|
|
|
|
# Remove role via interface method
|
|
if interface.removeRoleFromUserMandate(userMandateId, str(role.id)):
|
|
roleRemoved = True
|
|
|
|
if roleRemoved:
|
|
logger.info(f"Removed role {roleLabel} from user {userId} by SysAdmin {currentUser.id}")
|
|
|
|
userRoleLabels = _getUserRoleLabels(interface, userId)
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"fullName": user.fullName,
|
|
"isSysAdmin": user.isSysAdmin,
|
|
"enabled": user.enabled,
|
|
"roleLabels": userRoleLabels,
|
|
"roleCount": len(userRoleLabels)
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error removing role from user: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to remove role from user: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/roles/{roleLabel}/users", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("60/minute")
|
|
def get_users_with_role(
|
|
request: Request,
|
|
roleLabel: str = Path(..., description="Role label"),
|
|
mandateId: Optional[str] = Query(None, description="Filter by mandate ID (via UserMandate)"),
|
|
currentUser: User = Depends(requireSysAdmin)
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all users with a specific role.
|
|
MULTI-TENANT: SysAdmin-only.
|
|
|
|
Path Parameters:
|
|
- roleLabel: Role label
|
|
|
|
Query Parameters:
|
|
- mandateId: Optional filter by mandate ID (via UserMandate table)
|
|
|
|
Returns:
|
|
- List of users with the specified role
|
|
"""
|
|
try:
|
|
interface = getRootInterface()
|
|
|
|
# Get role by label
|
|
role = interface.getRoleByLabel(roleLabel)
|
|
if not role:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Role '{roleLabel}' not found"
|
|
)
|
|
|
|
# Get all UserMandateRole assignments for this role (Pydantic models)
|
|
roleAssignments = interface.getUserMandateRolesByRole(str(role.id))
|
|
|
|
# Get unique userMandateIds
|
|
userMandateIds = {str(ra.userMandateId) for ra in roleAssignments}
|
|
|
|
# Get userIds from UserMandate records
|
|
userIds: Set[str] = set()
|
|
for userMandateId in userMandateIds:
|
|
um = interface.getUserMandateById(userMandateId)
|
|
if um:
|
|
# Filter by mandate if specified
|
|
if mandateId and str(um.mandateId) != mandateId:
|
|
continue
|
|
userIds.add(str(um.userId))
|
|
|
|
# Get users and format response
|
|
result = []
|
|
for userId in userIds:
|
|
user = interface.getUser(userId)
|
|
if user:
|
|
userRoleLabels = _getUserRoleLabels(interface, userId)
|
|
result.append({
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"fullName": user.fullName,
|
|
"isSysAdmin": user.isSysAdmin,
|
|
"enabled": user.enabled,
|
|
"roleLabels": userRoleLabels,
|
|
"roleCount": len(userRoleLabels)
|
|
})
|
|
|
|
return result
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting users with role: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to get users with role: {str(e)}"
|
|
)
|