Coverage for merco/scheduler/jobs.py: 0%

1"""任务管理""" 

2 

3from dataclasses import dataclass, field 

4from datetime import datetime 

5from typing import Optional 

6from enum import Enum 

7 

8 

9class TaskStatus(str, Enum): 

10 PENDING = "pending" 

11 RUNNING = "running" 

12 COMPLETED = "completed" 

13 FAILED = "failed" 

14 CANCELLED = "cancelled" 

15 

16 

17@dataclass 

18class Task: 

19 """任务对象""" 

20 id: str 

21 name: str 

22 description: str 

23 status: TaskStatus = TaskStatus.PENDING 

24 created_at: datetime = field(default_factory=datetime.now) 

25 started_at: Optional[datetime] = None 

26 completed_at: Optional[datetime] = None 

27 result: Optional[str] = None 

28 error: Optional[str] = None 

29 

30 

31class TaskManager: 

32 """任务管理器""" 

33 

34 def __init__(self): 

35 self._tasks: dict[str, Task] = {} 

36 

37 def create(self, name: str, description: str) -> Task: 

38 """创建任务""" 

39 import uuid 

40 task = Task(id=str(uuid.uuid4())[:8], name=name, description=description) 

41 self._tasks[task.id] = task 

42 return task 

43 

44 def get(self, task_id: str) -> Optional[Task]: 

45 """获取任务""" 

46 return self._tasks.get(task_id) 

47 

48 def update_status(self, task_id: str, status: TaskStatus, result: str = None, error: str = None): 

49 """更新任务状态""" 

50 task = self._tasks.get(task_id) 

51 if task: 

52 task.status = status 

53 task.result = result 

54 task.error = error 

55 if status == TaskStatus.RUNNING: 

56 task.started_at = datetime.now() 

57 elif status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED): 

58 task.completed_at = datetime.now() 

59 

60 def list_tasks(self, status: TaskStatus = None) -> list[Task]: 

61 """列出任务""" 

62 tasks = list(self._tasks.values()) 

63 if status: 

64 tasks = [t for t in tasks if t.status == status] 

65 return tasks