""" FSM Transitions for Project model. This module defines declarative transitions for the Project entity, replacing the previous signal-based approach with explicit, testable transitions. """ from typing import Any, Dict, Optional from core.utils.common import load_func from django.conf import settings from fsm.registry import register_state_transition from fsm.state_choices import ProjectStateChoices from fsm.state_manager import StateManager from fsm.transitions import ModelChangeTransition, TransitionContext from fsm.utils import get_or_initialize_state, infer_entity_state_from_data @register_state_transition('project', 'project_created', triggers_on_create=True, triggers_on_update=False) class ProjectCreatedTransition(ModelChangeTransition): """ Transition when a new project is created. This is the initial state transition that occurs when a project is first saved to the database. Trigger: Automatically on creation (triggers_on_create=True, triggers_on_update=False) """ def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return ProjectStateChoices.CREATED def should_execute(self, context: TransitionContext) -> bool: """Only execute on creation, never on updates.""" return self.is_creating def get_reason(self, context: TransitionContext) -> str: """Return detailed reason for project creation.""" return 'Project created' def transition(self, context: TransitionContext) -> Dict[str, Any]: """ Execute project creation transition. Args: context: Transition context containing project and user information Returns: Context data to store with the state record """ project = context.entity return { 'organization_id': project.organization_id, 'title': project.title, 'created_by_id': project.created_by_id if project.created_by_id else None, 'label_config_present': bool(project.label_config), } # Note: Project state transitions (IN_PROGRESS, COMPLETED) are triggered by task state changes # via update_project_state_after_task_change() helper function, not by direct project model changes. @register_state_transition('project', 'project_in_progress', triggers_on_create=False, triggers_on_update=False) class ProjectInProgressTransition(ModelChangeTransition): """ Transition when project moves to IN_PROGRESS state. Triggered when: First annotation is submitted on any task From: CREATED -> IN_PROGRESS """ def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return ProjectStateChoices.IN_PROGRESS def get_reason(self, context: TransitionContext) -> str: return 'Project moved to in progress - first annotation submitted' def transition(self, context: TransitionContext) -> Dict[str, Any]: project = context.entity return { 'organization_id': project.organization_id, 'total_tasks': project.tasks.count(), } @register_state_transition('project', 'project_completed', triggers_on_create=False, triggers_on_update=False) class ProjectCompletedTransition(ModelChangeTransition): """ Transition when project moves to COMPLETED state. Triggered when: All tasks in project are COMPLETED From: IN_PROGRESS -> COMPLETED """ def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return ProjectStateChoices.COMPLETED def get_reason(self, context: TransitionContext) -> str: return 'Project completed - all tasks completed' def transition(self, context: TransitionContext) -> Dict[str, Any]: project = context.entity return { 'organization_id': project.organization_id, 'total_tasks': project.tasks.count(), } @register_state_transition( 'project', 'project_in_progress_from_completed', triggers_on_create=False, triggers_on_update=False ) class ProjectInProgressFromCompletedTransition(ModelChangeTransition): """ Transition when project moves back to IN_PROGRESS from COMPLETED. Triggered when: Any task becomes not COMPLETED (e.g., annotations deleted) From: COMPLETED -> IN_PROGRESS """ def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return ProjectStateChoices.IN_PROGRESS def get_reason(self, context: TransitionContext) -> str: return 'Project moved back to in progress - task became incomplete' def transition(self, context: TransitionContext) -> Dict[str, Any]: project = context.entity return { 'organization_id': project.organization_id, 'total_tasks': project.tasks.count(), } def sync_project_state(project, user=None, reason=None, context_data=None): current_state = StateManager.get_current_state_value(project) inferred_state = infer_entity_state_from_data(project) if current_state is None: get_or_initialize_state(project, user=user, inferred_state=inferred_state) return if current_state != inferred_state: if inferred_state == ProjectStateChoices.IN_PROGRESS: # Select in progress transition based on current state if current_state == ProjectStateChoices.CREATED: StateManager.execute_transition(entity=project, transition_name='project_in_progress', user=user) elif current_state == ProjectStateChoices.COMPLETED: StateManager.execute_transition( entity=project, transition_name='project_in_progress_from_completed', user=user ) elif inferred_state == ProjectStateChoices.COMPLETED: StateManager.execute_transition(entity=project, transition_name='project_completed', user=user) def update_project_state_after_task_change(project, user=None): update_func = load_func(settings.FSM_SYNC_PROJECT_STATE) return update_func(project, user)