40 lines
1.0 KiB
Python
40 lines
1.0 KiB
Python
"""
|
|
Task state management for Sanguo Orchestrator
|
|
Defines Task state machine and transitions
|
|
"""
|
|
import enum
|
|
from dataclasses import dataclass
|
|
|
|
|
|
class TaskState(enum.Enum):
|
|
"""Task execution states"""
|
|
PENDING = "pending"
|
|
RUNNING = "running"
|
|
DONE = "done"
|
|
FAILED = "failed"
|
|
|
|
|
|
@dataclass
|
|
class Task:
|
|
"""Represents a task in the orchestrator"""
|
|
task_id: str
|
|
task_type: str
|
|
status: TaskState = TaskState.PENDING
|
|
result_id: int | None = None
|
|
error_msg: str | None = None
|
|
|
|
def start(self):
|
|
"""Transition from PENDING to RUNNING"""
|
|
if self.status != TaskState.PENDING:
|
|
raise ValueError(f"不能从 {self.status.name} 启动")
|
|
self.status = TaskState.RUNNING
|
|
|
|
def complete(self, result_id: int):
|
|
"""Transition to DONE with result"""
|
|
self.status = TaskState.DONE
|
|
self.result_id = result_id
|
|
|
|
def fail(self, error_msg: str):
|
|
"""Transition to FAILED with error message"""
|
|
self.status = TaskState.FAILED
|
|
self.error_msg = error_msg |