237 lines
8.7 KiB
Python
237 lines
8.7 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Dynamic Options API feature module.
|
|
Provides dynamic options for frontend select/multiselect fields.
|
|
"""
|
|
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
from modules.datamodels.datamodelUam import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Standard role definitions (fallback if database is not available)
|
|
STANDARD_ROLES = [
|
|
{"value": "sysadmin", "label": {"en": "System Administrator", "fr": "Administrateur système"}},
|
|
{"value": "admin", "label": {"en": "Administrator", "fr": "Administrateur"}},
|
|
{"value": "user", "label": {"en": "User", "fr": "Utilisateur"}},
|
|
{"value": "viewer", "label": {"en": "Viewer", "fr": "Visualiseur"}},
|
|
]
|
|
|
|
# Authentication authority options
|
|
AUTH_AUTHORITY_OPTIONS = [
|
|
{"value": "local", "label": {"en": "Local", "fr": "Local"}},
|
|
{"value": "google", "label": {"en": "Google", "fr": "Google"}},
|
|
{"value": "msft", "label": {"en": "Microsoft", "fr": "Microsoft"}},
|
|
]
|
|
|
|
# Connection status options
|
|
# Note: Matches ConnectionStatus enum values (active, expired, revoked, pending)
|
|
# Plus "error" for error states (not in enum but used in UI)
|
|
CONNECTION_STATUS_OPTIONS = [
|
|
{"value": "active", "label": {"en": "Active", "fr": "Actif"}},
|
|
{"value": "expired", "label": {"en": "Expired", "fr": "Expiré"}},
|
|
{"value": "revoked", "label": {"en": "Revoked", "fr": "Révoqué"}},
|
|
{"value": "pending", "label": {"en": "Pending", "fr": "En attente"}},
|
|
{"value": "error", "label": {"en": "Error", "fr": "Erreur"}},
|
|
]
|
|
|
|
|
|
def getOptions(optionsName: str, services, currentUser: Optional[User] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get options for a given options name.
|
|
|
|
Args:
|
|
optionsName: Name of the options set to retrieve (e.g., "user.role", "user.connection")
|
|
services: Services instance for data access
|
|
currentUser: Optional current user for context-aware options
|
|
|
|
Returns:
|
|
List of option dictionaries with "value" and "label" keys
|
|
|
|
Raises:
|
|
ValueError: If optionsName is not recognized
|
|
"""
|
|
logger.debug(f"getOptions called with optionsName='{optionsName}' (repr: {repr(optionsName)})")
|
|
optionsNameLower = optionsName.lower()
|
|
logger.debug(f"optionsNameLower='{optionsNameLower}'")
|
|
|
|
if optionsNameLower == "user.role":
|
|
# Fetch roles from database
|
|
if currentUser:
|
|
try:
|
|
roles = services.interfaceDbApp.getAllRoles()
|
|
|
|
# Convert Role objects to options format
|
|
options = []
|
|
for role in roles:
|
|
# Use English description as label, fallback to roleLabel
|
|
# Handle TextMultilingual object
|
|
if hasattr(role.description, 'get_text'):
|
|
# TextMultilingual object
|
|
label = role.description.get_text('en')
|
|
elif isinstance(role.description, dict):
|
|
# Dict format (backward compatibility)
|
|
label = role.description.get("en", role.roleLabel)
|
|
else:
|
|
# Fallback to roleLabel
|
|
label = role.roleLabel
|
|
|
|
options.append({
|
|
"value": role.roleLabel,
|
|
"label": label
|
|
})
|
|
|
|
# If no roles in database, return standard roles as fallback
|
|
if options:
|
|
return options
|
|
except Exception as e:
|
|
logger.warning(f"Error fetching roles from database, using fallback: {e}")
|
|
|
|
# Fallback to standard roles if database fetch fails or no user context
|
|
return STANDARD_ROLES
|
|
|
|
elif optionsNameLower == "auth.authority":
|
|
return AUTH_AUTHORITY_OPTIONS
|
|
|
|
elif optionsNameLower == "connection.status":
|
|
return CONNECTION_STATUS_OPTIONS
|
|
|
|
elif optionsNameLower == "user.connection":
|
|
# Dynamic options: Get user connections from database
|
|
if not currentUser:
|
|
return []
|
|
|
|
try:
|
|
connections = services.interfaceDbApp.getUserConnections(currentUser.id)
|
|
|
|
return [
|
|
{
|
|
"value": conn.id,
|
|
"label": {
|
|
"en": f"{conn.authority.value} - {conn.externalUsername or conn.externalId}",
|
|
"fr": f"{conn.authority.value} - {conn.externalUsername or conn.externalId}"
|
|
}
|
|
}
|
|
for conn in connections
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"Error fetching user connections for options: {e}")
|
|
return []
|
|
|
|
elif optionsNameLower in ("user", "users"):
|
|
# Dynamic options: Get all users for the current mandate
|
|
if not currentUser:
|
|
return []
|
|
|
|
try:
|
|
users = services.interfaceDbApp.getUsersByMandate(currentUser.mandateId)
|
|
|
|
# Handle both list and PaginatedResult
|
|
if hasattr(users, 'items'):
|
|
userList = users.items
|
|
else:
|
|
userList = users
|
|
|
|
return [
|
|
{
|
|
"value": user.id,
|
|
"label": user.fullName or user.username or user.email or user.id
|
|
}
|
|
for user in userList
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"Error fetching users for options: {e}")
|
|
return []
|
|
|
|
elif optionsNameLower in ("trusteeorganisation", "trustee.organisation"):
|
|
# Dynamic options: Get all trustee organisations
|
|
if not currentUser:
|
|
return []
|
|
|
|
try:
|
|
result = services.interfaceDbTrustee.getAllOrganisations()
|
|
|
|
# Handle PaginatedResult
|
|
items = result.items if hasattr(result, 'items') else result
|
|
|
|
return [
|
|
{
|
|
"value": org.get("id") if isinstance(org, dict) else org.id,
|
|
"label": org.get("label") if isinstance(org, dict) else org.label
|
|
}
|
|
for org in items
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"Error fetching trustee organisations for options: {e}")
|
|
return []
|
|
|
|
elif optionsNameLower in ("trusteerole", "trustee.role"):
|
|
# Dynamic options: Get all trustee roles
|
|
if not currentUser:
|
|
return []
|
|
|
|
try:
|
|
result = services.interfaceDbTrustee.getAllRoles()
|
|
|
|
# Handle PaginatedResult
|
|
items = result.items if hasattr(result, 'items') else result
|
|
|
|
return [
|
|
{
|
|
"value": role.get("id") if isinstance(role, dict) else role.id,
|
|
# TrusteeRole uses 'desc' field, not 'label'
|
|
"label": role.get("desc", role.get("id")) if isinstance(role, dict) else getattr(role, "desc", role.id)
|
|
}
|
|
for role in items
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"Error fetching trustee roles for options: {e}")
|
|
return []
|
|
|
|
elif optionsNameLower in ("trusteecontract", "trustee.contract"):
|
|
# Dynamic options: Get all trustee contracts
|
|
if not currentUser:
|
|
return []
|
|
|
|
try:
|
|
result = services.interfaceDbTrustee.getAllContracts()
|
|
|
|
# Handle PaginatedResult
|
|
items = result.items if hasattr(result, 'items') else result
|
|
|
|
return [
|
|
{
|
|
"value": contract.get("id") if isinstance(contract, dict) else contract.id,
|
|
"label": contract.get("label") if isinstance(contract, dict) else (contract.get("name") if isinstance(contract, dict) else getattr(contract, "label", getattr(contract, "name", contract.id)))
|
|
}
|
|
for contract in items
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"Error fetching trustee contracts for options: {e}")
|
|
return []
|
|
|
|
else:
|
|
logger.error(f"Unknown options name: '{optionsName}' (lower: '{optionsNameLower}')")
|
|
raise ValueError(f"Unknown options name: {optionsName}")
|
|
|
|
|
|
def getAvailableOptionsNames() -> List[str]:
|
|
"""
|
|
Get list of all available options names.
|
|
|
|
Returns:
|
|
List of available options names
|
|
"""
|
|
return [
|
|
"user.role",
|
|
"auth.authority",
|
|
"connection.status",
|
|
"user.connection",
|
|
"User",
|
|
"TrusteeOrganisation",
|
|
"TrusteeRole",
|
|
"TrusteeContract",
|
|
]
|
|
|