Check if transition to target phase is allowed.
(self, target_phase: WorkflowPhase)
| 168 | return phase_weights.get(self.phase, 0.0) |
| 169 | |
| 170 | def can_transition_to(self, target_phase: WorkflowPhase) -> bool: |
| 171 | """Check if transition to target phase is allowed.""" |
| 172 | allowed_transitions = { |
| 173 | WorkflowPhase.INITIAL: [WorkflowPhase.URL_COLLECTION], |
| 174 | WorkflowPhase.URL_COLLECTION: [WorkflowPhase.URL_VALIDATION, WorkflowPhase.ERROR], |
| 175 | WorkflowPhase.URL_VALIDATION: [WorkflowPhase.SCHEMA_DEFINITION, WorkflowPhase.URL_COLLECTION, WorkflowPhase.ERROR], |
| 176 | WorkflowPhase.SCHEMA_DEFINITION: [WorkflowPhase.SCHEMA_VALIDATION, WorkflowPhase.ERROR], |
| 177 | WorkflowPhase.SCHEMA_VALIDATION: [WorkflowPhase.CODE_GENERATION, WorkflowPhase.SCHEMA_DEFINITION, WorkflowPhase.ERROR], |
| 178 | WorkflowPhase.CODE_GENERATION: [WorkflowPhase.READY_TO_EXECUTE, WorkflowPhase.ERROR], |
| 179 | WorkflowPhase.READY_TO_EXECUTE: [WorkflowPhase.EXECUTING, WorkflowPhase.CODE_GENERATION, WorkflowPhase.ERROR], |
| 180 | WorkflowPhase.EXECUTING: [WorkflowPhase.COMPLETED, WorkflowPhase.ERROR], |
| 181 | WorkflowPhase.COMPLETED: [WorkflowPhase.INITIAL], # Can restart |
| 182 | WorkflowPhase.ERROR: [WorkflowPhase.INITIAL] # Can restart |
| 183 | } |
| 184 | |
| 185 | return target_phase in allowed_transitions.get(self.phase, []) |
| 186 | |
| 187 | class Config: |
| 188 | json_encoders = { |
no outgoing calls
no test coverage detected