# 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(services.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", ]