Coverage for interfaces / schedulable_task.py: 100%

1# interfaces/schedulable_task.py 

2from abc import ABC, abstractmethod 

3from enum import IntEnum, Enum 

4from typing import Dict 

5 

6 

7class TaskPriority(IntEnum): 

8 """태스크 우선순위. 낮은 숫자 = 높은 우선순위.""" 

9 CRITICAL = 0 # 매수/매도 주문 

10 HIGH = 10 # User-initiated queries 

11 NORMAL = 50 # Strategy scheduler 

12 LOW = 100 # Background batch jobs 

13 

14 

15class TaskState(str, Enum): 

16 """태스크 실행 상태.""" 

17 IDLE = "idle" 

18 RUNNING = "running" 

19 SUSPENDED = "suspended" 

20 STOPPED = "stopped" 

21 

22 

23class SchedulableTask(ABC): 

24 """스케줄러가 관리하는 태스크의 추상 인터페이스. 

25 

26 모든 백그라운드/포그라운드 태스크는 이 인터페이스를 구현하여 

27 BackgroundScheduler/ForegroundScheduler에 등록할 수 있다. 

28 """ 

29 

30 @property 

31 @abstractmethod 

32 def task_name(self) -> str: 

33 """태스크 고유 식별자.""" 

34 ... 

35 

36 @property 

37 @abstractmethod 

38 def priority(self) -> TaskPriority: 

39 """태스크 우선순위.""" 

40 ... 

41 

42 @abstractmethod 

43 async def start(self) -> None: 

44 """태스크의 비동기 루프 또는 1회 실행을 시작한다.""" 

45 ... 

46 

47 @abstractmethod 

48 async def stop(self) -> None: 

49 """태스크를 정상 종료하고, 실행 중인 코루틴을 취소한다.""" 

50 ... 

51 

52 @abstractmethod 

53 async def suspend(self) -> None: 

54 """태스크를 일시 중지한다 (예: 포그라운드 액션이 API 대역폭 필요 시).""" 

55 ... 

56 

57 @abstractmethod 

58 async def resume(self) -> None: 

59 """일시 중지된 태스크를 재개한다.""" 

60 ... 

61 

62 @property 

63 @abstractmethod 

64 def state(self) -> TaskState: 

65 """현재 태스크 상태.""" 

66 ... 

67 

68 @abstractmethod 

69 def get_progress(self) -> Dict: 

70 """태스크 진행률 및 상태 정보를 반환한다. 

71 

72 모든 구현체는 최소한 {"running": bool} 키를 포함해야 한다. 

73 진행률이 있는 태스크는 processed, total, elapsed 등을 추가로 포함한다. 

74 """ 

75 ...