"""
|
FSM Base Model for Label Studio.
|
|
This module contains only FsmHistoryStateModel - the base class that models
|
inherit from to get FSM integration. State model definitions are in state_models.py
|
to avoid registration issues in LSE.
|
"""
|
|
import logging
|
from typing import Any, Dict
|
|
from django.db import models
|
|
logger = logging.getLogger(__name__)
|
|
|
# =============================================================================
|
# FsmHistoryStateModel - Base Model for FSM Integration
|
# =============================================================================
|
|
|
class FsmHistoryStateModel(models.Model):
|
"""
|
FSM History State Model - Base class for models that participate in FSM state tracking.
|
|
This class provides explicit FSM integration through model lifecycle hooks,
|
replacing the implicit signal-based approach with predictable, testable behavior.
|
|
Key features:
|
- Intercepts save operations to trigger FSM transitions
|
- Tracks field changes for transition logic
|
- Maintains CurrentContext for user/org tracking
|
- Provides explicit transition determination
|
- Fails gracefully - FSM errors don't break saves
|
|
Usage:
|
class Task(FsmHistoryStateModel):
|
# ... model fields ...
|
|
def _determine_fsm_transition(self) -> Optional[str]:
|
if self._state.adding: # Creating new instance
|
return 'task_created'
|
|
changed = self._get_changed_fields()
|
if 'is_labeled' in changed and changed['is_labeled'][1]:
|
return 'task_labeled'
|
|
return None
|
|
def _get_fsm_transition_data(self) -> Dict[str, Any]:
|
return {
|
'project_id': self.project_id,
|
'overlap': self.overlap
|
}
|
"""
|
|
class Meta:
|
abstract = True
|
|
def __init__(self, *args, **kwargs):
|
super().__init__(*args, **kwargs)
|
# Track original field values for change detection
|
# Initialize as empty dict for safe access
|
self._original_values = {}
|
|
@classmethod
|
def from_db(cls, db, field_names, values):
|
"""
|
Override from_db to store raw DB values for lazy capture.
|
|
Django calls this method instead of __init__ when loading models from the database.
|
|
PERFORMANCE: We store the raw field values here without processing them.
|
This avoids accessing any ForeignKey fields (which would trigger queries).
|
We only process these values into _original_values in save() when we actually
|
need them for change detection.
|
"""
|
instance = super().from_db(db, field_names, values)
|
# Initialize as empty dict for safe access
|
instance._original_values = dict(zip(field_names, values))
|
|
return instance
|
|
def __reduce_ex__(self, protocol):
|
"""
|
Override serialization to exclude internal FSM tracking fields.
|
|
Django's serialization uses pickle which calls __reduce_ex__.
|
We exclude FSM tracking fields since they're only needed for runtime
|
change detection, not for serialization/restoration.
|
"""
|
# Get the default reduction
|
reduction = super().__reduce_ex__(protocol)
|
|
# reduction is a tuple: (callable, args, state, ...)
|
# state is the instance __dict__
|
if len(reduction) >= 3 and isinstance(reduction[2], dict):
|
state = reduction[2].copy()
|
# Remove internal FSM fields from serialization
|
state.pop('_original_values', None)
|
# Return new reduction with cleaned state
|
return (reduction[0], reduction[1], state) + reduction[3:]
|
|
return reduction
|
|
def _get_changed_fields(self) -> Dict[str, tuple]:
|
"""
|
Get fields that changed since the last load/save.
|
|
Returns:
|
Dict mapping field names to (old_value, new_value) tuples
|
Note: For ForeignKey fields, old_value will be the PK, new_value will be the object
|
|
Example:
|
changed = self._get_changed_fields()
|
if 'is_labeled' in changed:
|
old_val, new_val = changed['is_labeled']
|
if not old_val and new_val:
|
# Task became labeled
|
pass
|
"""
|
# If no original values captured yet, nothing has changed
|
# Use hasattr check to handle cases where _original_values doesn't exist
|
if not hasattr(self, '_original_values') or not self._original_values:
|
return {}
|
|
changed = {}
|
for field in self._meta.fields:
|
# Only check fields that were captured in _original_values
|
# Fields that were deferred during capture won't be in _original_values
|
# and should be considered unchanged
|
if field.attname not in self._original_values:
|
continue
|
if field.is_relation and field.many_to_many:
|
continue
|
|
old_value = self._original_values[field.attname]
|
new_value = getattr(self, field.attname, None)
|
|
if old_value != new_value:
|
changed[field.attname] = (old_value, new_value)
|
return changed
|
|
def _determine_fsm_transitions(self, is_creating: bool = None, changed_fields: dict = None) -> list:
|
"""
|
Determine which FSM transitions should be triggered based on model state.
|
|
This method automatically discovers registered transitions for this entity
|
and checks which ones should execute based on their trigger metadata.
|
|
Args:
|
is_creating: Whether this is a creation. If None, checks self._state.adding
|
changed_fields: Dict of changed fields. If None, computes them.
|
|
Override this method ONLY if you need custom transition logic beyond
|
what the declarative triggers provide.
|
|
Returns:
|
List of transition names to execute (in order)
|
|
Note:
|
In most cases, you don't need to override this. Just register transitions
|
with appropriate trigger metadata using @register_state_transition decorator.
|
|
Example of custom override (if needed):
|
def _determine_fsm_transitions(self, is_creating=None, changed_fields=None) -> list:
|
# Get default transitions
|
transitions = super()._determine_fsm_transitions(is_creating, changed_fields)
|
|
# Add custom logic
|
if self.some_complex_condition():
|
transitions.append('custom_transition')
|
|
return transitions
|
"""
|
from fsm.registry import transition_registry
|
|
entity_name = self._meta.model_name
|
|
# Use provided is_creating, or fall back to checking _state.adding
|
if is_creating is None:
|
is_creating = self._state.adding
|
|
# Use provided changed_fields, or compute them
|
if changed_fields is None:
|
changed_fields = {} if is_creating else self._get_changed_fields()
|
|
# Debug logging for transition determination
|
if entity_name == 'project' and not is_creating:
|
logger.debug(
|
f'FSM: Determining transitions for {entity_name}',
|
extra={
|
'entity_id': self.pk,
|
'is_creating': is_creating,
|
'changed_fields': list(changed_fields.keys()),
|
'changed_fields_detail': changed_fields,
|
},
|
)
|
|
# Get all registered transitions for this entity
|
registered_transitions = transition_registry.get_transitions_for_entity(entity_name)
|
if not registered_transitions:
|
return []
|
|
transitions_to_execute = []
|
|
for transition_name, transition_class in registered_transitions.items():
|
# Check if this transition should execute based on trigger metadata
|
should_execute = False
|
|
# Check creation trigger
|
if is_creating and getattr(transition_class, '_triggers_on_create', False):
|
should_execute = True
|
|
# Check update triggers
|
elif not is_creating and getattr(transition_class, '_triggers_on_update', True):
|
trigger_fields = getattr(transition_class, '_trigger_fields', [])
|
|
# If no specific fields, check if transition has custom logic
|
if not trigger_fields:
|
# Let the transition's should_execute method decide
|
# We'll add it and let it validate later
|
should_execute = True
|
else:
|
# Check if any trigger fields changed
|
for field in trigger_fields:
|
if field in changed_fields:
|
should_execute = True
|
break
|
|
# If trigger metadata says we should execute, also check the transition's should_execute() method
|
if should_execute:
|
try:
|
# Instantiate the transition to check should_execute() if it exists
|
# We need a minimal context to check should_execute
|
from fsm.transitions import TransitionContext
|
|
# Create a temporary transition instance with full context
|
# Convert changed_fields to the format expected by ModelChangeTransition
|
formatted_changed_fields = {k: {'old': v[0], 'new': v[1]} for k, v in changed_fields.items()}
|
|
# Create transition with all relevant data for should_execute() check
|
temp_transition = transition_class(
|
is_creating=is_creating, changed_fields=formatted_changed_fields
|
)
|
|
# Check if should_execute is overridden (not using the base implementation)
|
# The base implementation always returns True, so we only check if it's been customized
|
from fsm.transitions import BaseTransition
|
|
should_execute_method = getattr(type(temp_transition), 'should_execute', None)
|
base_should_execute = getattr(BaseTransition, 'should_execute', None)
|
|
# Only call should_execute if it's been overridden in the subclass
|
if should_execute_method and should_execute_method != base_should_execute:
|
# Build a minimal context for should_execute check
|
# NOTE: We skip getting current state here to avoid recursion issues
|
# The actual state will be retrieved during transition execution
|
minimal_context = TransitionContext(
|
entity=self,
|
current_user=None, # Will be set properly during execution
|
current_state_object=None, # Skip to avoid recursion
|
current_state=None, # Skip to avoid recursion
|
target_state=None, # Will be computed
|
organization_id=getattr(self, 'organization_id', None),
|
)
|
|
# Get target_state (can use entity from minimal_context)
|
target_state = temp_transition.get_target_state(minimal_context)
|
|
# Update context with computed target_state
|
context = TransitionContext(
|
entity=self,
|
current_user=None,
|
current_state_object=None,
|
current_state=None,
|
target_state=target_state,
|
organization_id=getattr(self, 'organization_id', None),
|
)
|
|
# Call should_execute to do final filtering
|
if not temp_transition.should_execute(context):
|
should_execute = False
|
logger.debug(
|
f'FSM: Transition {transition_name} filtered out by should_execute()',
|
extra={
|
'entity_type': entity_name,
|
'entity_id': self.pk,
|
'transition_name': transition_name,
|
},
|
)
|
except Exception as e:
|
# If should_execute check fails, log but still add the transition
|
# Let it fail during actual execution with proper error handling
|
logger.debug(
|
f'FSM: Error checking should_execute for {transition_name}: {e}',
|
extra={
|
'entity_type': entity_name,
|
'entity_id': self.pk,
|
'transition_name': transition_name,
|
'error': str(e),
|
},
|
)
|
|
if should_execute:
|
transitions_to_execute.append(transition_name)
|
|
return transitions_to_execute
|
|
def _get_fsm_transition_data(self) -> Dict[str, Any]:
|
"""
|
Get data to pass to the FSM transition.
|
|
Override in subclasses to provide transition-specific data that should
|
be stored in the state record's context_data field.
|
|
Returns:
|
Dictionary of data to pass to transition
|
|
Example:
|
def _get_fsm_transition_data(self) -> Dict[str, Any]:
|
return {
|
'project_id': self.project_id,
|
'completed_by_id': self.completed_by_id,
|
'annotation_count': self.annotations.count()
|
}
|
"""
|
return {}
|
|
def _should_execute_fsm(self) -> bool:
|
"""
|
Check if FSM processing should be executed.
|
|
Returns False if:
|
- Feature flag is disabled (cached at request level)
|
- Manually disabled via set_fsm_disabled() (for tests/bulk operations)
|
- Explicitly skipped via instance attribute
|
|
Returns:
|
True if FSM should execute, False otherwise
|
|
PERFORMANCE: Uses cached FSM enabled state from CurrentContext that was set
|
once per request when user was initialized. This is a simple boolean check
|
instead of repeated feature flag lookups and user authentication checks.
|
"""
|
# Check for instance-level skip flag
|
if getattr(self, '_skip_fsm', False):
|
return False
|
|
# Fast path: Check cached FSM enabled state
|
# This was set once per request in CurrentContext.set_user()
|
from core.current_request import CurrentContext
|
|
return CurrentContext.is_fsm_enabled()
|
|
def save(self, *args, **kwargs):
|
"""
|
Override save to trigger FSM transitions based on model changes.
|
|
This method:
|
1. Captures the current state (creating vs updating)
|
2. Performs the actual database save
|
3. Determines if an FSM transition is needed
|
4. Executes the transition if needed
|
5. Gracefully handles FSM errors without breaking the save
|
|
Args:
|
*args: Positional arguments passed to super().save()
|
**kwargs: Keyword arguments passed to super().save()
|
Special kwarg: skip_fsm=True to bypass FSM processing
|
|
Returns:
|
Whatever super().save() returns
|
"""
|
from core.current_request import CurrentContext
|
|
# Check for explicit FSM skip flag
|
skip_fsm = kwargs.pop('skip_fsm', CurrentContext.is_fsm_disabled())
|
|
# Check if this is a creation vs update
|
is_creating = self._state.adding
|
|
# Capture changed fields before save (only for updates)
|
changed_fields = {} if is_creating else self._get_changed_fields()
|
|
# Perform the actual save
|
result = super().save(*args, **kwargs)
|
|
# After successful save, update _original_values to current values
|
# This ensures subsequent saves can detect changes correctly
|
# Store attname values (raw PK for ForeignKey fields) to match from_db() format
|
self._original_values = {}
|
for field in self._meta.fields:
|
if field.is_relation and field.many_to_many:
|
continue
|
self._original_values[field.attname] = getattr(self, field.attname, None)
|
|
# After successful save, trigger FSM transitions if enabled and not skipped
|
should_execute = not skip_fsm and self._should_execute_fsm()
|
|
logger.debug(
|
f'FSM check for {self.__class__.__name__} {self.pk}: skip_fsm={skip_fsm}, should_execute={should_execute}',
|
extra={
|
'entity_type': self.__class__.__name__,
|
'entity_id': self.pk,
|
'skip_fsm': skip_fsm,
|
'should_execute': should_execute,
|
},
|
)
|
if not skip_fsm and should_execute:
|
try:
|
# Pass is_creating and changed_fields that were captured before save()
|
transitions = self._determine_fsm_transitions(is_creating=is_creating, changed_fields=changed_fields)
|
logger.debug(
|
f'FSM transitions determined for {self.__class__.__name__} {self.pk}: {transitions}',
|
extra={
|
'entity_type': self.__class__.__name__,
|
'entity_id': self.pk,
|
'transitions': transitions,
|
'is_creating': is_creating,
|
},
|
)
|
for transition_name in transitions:
|
try:
|
self._execute_fsm_transition(
|
transition_name=transition_name, is_creating=is_creating, changed_fields=changed_fields
|
)
|
except Exception as e:
|
# Log error for this specific transition but continue with others
|
logger.error(
|
f'FSM transition {transition_name} failed for {self.__class__.__name__} {self.pk}',
|
extra={
|
'event': 'fsm.transition_failed_on_save',
|
'entity_type': self.__class__.__name__,
|
'entity_id': self.pk,
|
'transition_name': transition_name,
|
'error': str(e),
|
'is_creating': is_creating,
|
},
|
exc_info=True,
|
)
|
except Exception as e:
|
# Log error in determining transitions
|
logger.error(
|
f'FSM transition discovery failed for {self.__class__.__name__} {self.pk}',
|
extra={
|
'event': 'fsm.transition_discovery_failed',
|
'entity_type': self.__class__.__name__,
|
'entity_id': self.pk,
|
'error': str(e),
|
'is_creating': is_creating,
|
},
|
exc_info=True,
|
)
|
|
return result
|
|
def _execute_fsm_transition(self, transition_name: str, is_creating: bool, changed_fields: Dict[str, tuple]):
|
"""
|
Execute an FSM transition.
|
|
This method handles the actual transition execution, including:
|
- Getting current context (user, org_id)
|
- Preparing transition data
|
- Calling the state manager
|
|
Args:
|
transition_name: Name of the registered transition to execute
|
is_creating: Whether this is a new model creation
|
changed_fields: Dict of changed fields (field_name -> (old, new))
|
|
Note:
|
This is only called after _should_execute_fsm() returns True,
|
so CurrentContext should be available with a valid user.
|
"""
|
from core.current_request import CurrentContext
|
from fsm.state_manager import get_state_manager
|
|
StateManager = get_state_manager()
|
|
# Get context - should be available since _should_execute_fsm passed
|
user = CurrentContext.get_user()
|
org_id = CurrentContext.get_organization_id()
|
|
# Get transition-specific data from the model
|
transition_data = self._get_fsm_transition_data()
|
|
# Add metadata about the change
|
transition_data.update(
|
{
|
'is_creating': is_creating,
|
'changed_fields': {k: {'old': v[0], 'new': v[1]} for k, v in changed_fields.items()},
|
}
|
)
|
|
logger.info(
|
f'Executing FSM transition for {self.__class__.__name__}',
|
extra={
|
'event': 'fsm.transition_executing',
|
'entity_type': self.__class__.__name__,
|
'entity_id': self.pk,
|
'transition_name': transition_name,
|
'is_creating': is_creating,
|
'user_id': user.id if user else None,
|
'organization_id': org_id,
|
},
|
)
|
|
# Execute the registered transition
|
try:
|
StateManager.execute_transition(
|
entity=self,
|
transition_name=transition_name,
|
transition_data=transition_data,
|
user=user,
|
organization_id=org_id,
|
)
|
|
logger.info(
|
f'FSM transition executed successfully for {self.__class__.__name__}',
|
extra={
|
'event': 'fsm.transition_success',
|
'entity_type': self.__class__.__name__,
|
'entity_id': self.pk,
|
'transition_name': transition_name,
|
'user_id': user.id if user else None,
|
'organization_id': org_id,
|
},
|
)
|
except Exception:
|
# Re-raise to be caught by save() method
|
raise
|
