""" Utility functions for working with the declarative transition system. This module provides helper functions to make it easier to integrate the new Pydantic-based transition system with existing Label Studio code. """ import logging from typing import Any, Dict, List, Type from django.db.models import Model from fsm.registry import transition_registry from fsm.state_manager import get_state_manager from fsm.transitions import BaseTransition, TransitionValidationError logger = logging.getLogger(__name__) StateManager = get_state_manager() def get_available_transitions(entity: Model, user=None, validate: bool = False) -> Dict[str, Type[BaseTransition]]: """ Get available transitions for an entity. Args: entity: The entity to get transitions for user: User context for validation (only used when validate=True) validate: Whether to validate each transition against current state. When False, returns all registered transitions for the entity type. When True, filters to only transitions valid from current state (may be expensive). Returns: Dictionary mapping transition names to transition classes. When validate=False: All registered transitions for the entity type. When validate=True: Only transitions valid for the current state. """ entity_name = entity._meta.model_name.lower() available = transition_registry.get_transitions_for_entity(entity_name) if not validate: return available valid_transitions = {} for name, transition_class in available.items(): try: # Get current state information using potentially overridden StateManager current_state_object = StateManager.get_current_state_object(entity) current_state = current_state_object.state if current_state_object else None # Build minimal context for validation from .transitions import TransitionContext # Get target state from instance (may need entity context) # For validation purposes, we try to create with minimal/default data try: temp_instance = transition_class() # Create minimal context for dynamic target_state computation minimal_context = TransitionContext( entity=entity, current_user=user, current_state_object=current_state_object, current_state=current_state, target_state=None, # Will be computed organization_id=getattr(entity, 'organization_id', None), ) target_state = temp_instance.get_target_state(minimal_context) except (TypeError, ValueError): # Can't instantiate without required data - include in results # since we can't validate state transitions, we assume they're available valid_transitions[name] = transition_class continue context = TransitionContext( entity=entity, current_user=user, current_state_object=current_state_object, current_state=current_state, target_state=target_state, organization_id=getattr(entity, 'organization_id', None), ) # Use class-level validation that doesn't require an instance if transition_class.can_transition_from_state(context): valid_transitions[name] = transition_class except TransitionValidationError: # Transition is not valid for current state/context - this is expected continue except Exception as e: # Unexpected error during validation - this should be investigated logger.warning( 'Unexpected error validating transition', extra={ 'event': 'fsm.transition_validation_error', 'transition_name': name, 'entity_type': entity._meta.model_name, 'error': str(e), }, exc_info=True, ) continue return valid_transitions def create_transition_from_dict(transition_class: Type[BaseTransition], data: Dict[str, Any]) -> BaseTransition: """ Create a transition instance from a dictionary of data. This handles Pydantic validation and provides clear error messages. Args: transition_class: The transition class to instantiate data: Dictionary of transition data Returns: Validated transition instance Raises: ValueError: If data validation fails """ try: return transition_class(**data) except Exception as e: raise ValueError(f'Failed to create {transition_class.__name__}: {e}') def get_transition_schema(transition_class: Type[BaseTransition]) -> Dict[str, Any]: """ Get the JSON schema for a transition class. Useful for generating API documentation or frontend forms. Args: transition_class: The transition class Returns: JSON schema dictionary """ return transition_class.model_json_schema() def validate_transition_data(transition_class: Type[BaseTransition], data: Dict[str, Any]) -> Dict[str, List[str]]: """ Validate transition data without creating an instance. Args: transition_class: The transition class data: Data to validate Returns: Dictionary of field names to error messages (empty if valid) """ try: transition_class(**data) return {} except Exception as e: # Parse Pydantic validation errors errors = {} if hasattr(e, 'errors'): for error in e.errors(): field = '.'.join(str(loc) for loc in error['loc']) if field not in errors: errors[field] = [] errors[field].append(error['msg']) else: errors['__root__'] = [str(e)] return errors def get_entity_state_flow(entity: Model) -> List[Dict[str, Any]]: """ Get a summary of the state flow for an entity type. This analyzes all registered transitions and builds a flow diagram. Args: entity: Example entity instance Returns: List of state flow information """ entity_name = entity._meta.model_name.lower() transitions = transition_registry.get_transitions_for_entity(entity_name) # Build state flow information states = set() flows = [] for transition_name, transition_class in transitions.items(): # Create instance to get target state try: from .transitions import TransitionContext transition = transition_class() # Create minimal context for dynamic target_state computation minimal_context = ( TransitionContext( entity=entity, current_user=None, current_state_object=None, current_state=None, target_state=None, # Will be computed organization_id=getattr(entity, 'organization_id', None), ) if hasattr(entity, 'pk') else None ) target = transition.get_target_state(minimal_context) states.add(target) flows.append( { 'transition_name': transition_name, 'transition_class': transition_class.__name__, 'target_state': target, 'description': transition_class.__doc__ or '', 'fields': list(transition_class.model_fields.keys()), } ) except Exception: continue return flows