Coverage for qreward / utils / scheduler / base.py: 100.00%

1"""Base runner implementing template method pattern for sync/async 

2execution.""" 

3 

4import asyncio (empty)

5import concurrent.futures (empty)

6import logging (empty)

7from abc import ABC, abstractmethod (empty)

8from typing import Any, Callable, Optional, Union (empty)

9 

10from .config import ScheduleConfig, _sentinel_none (empty)

11from .context import ExecutionContext (empty)

12from .overload import OverloadChecker (empty)

13 

14logger = logging.getLogger(__name__) (empty)

15 

16# Cancelled error groups 

17_CancelledErrorGroups = ( (empty)

18 asyncio.CancelledError, 

19 asyncio.TimeoutError, 

20 concurrent.futures.CancelledError, 

21 TimeoutError, 

22) 

23 

24 

25class BaseRunner(ABC): (empty)

26 """Abstract base runner using template method pattern. 

27 

28 Subclasses implement the I/O primitives; shared pure-logic helpers 

29 live here so that sync and async runners share the same logic. 

30 """ 

31 

32 def __init__(self): (empty)

33 self.overload_checker = OverloadChecker() (empty)

34 

35 # ------------------------------------------------------------------ 

36 # Abstract I/O primitives 

37 # ------------------------------------------------------------------ 

38 

39 @abstractmethod (empty)

40 def create_task( (empty)

41 self, func: Callable, *args, **kwargs 

42 ) -> Union[asyncio.Task, concurrent.futures.Future]: 

43 """Create and submit a task for execution.""" 

44 

45 @abstractmethod (empty)

46 def get_task_result( (empty)

47 self, task: Union[asyncio.Task, concurrent.futures.Future] 

48 ) -> Any: 

49 """Get result from a completed task.""" 

50 

51 @abstractmethod (empty)

52 def get_task_exception( (empty)

53 self, task: Union[asyncio.Task, concurrent.futures.Future] 

54 ) -> Optional[BaseException]: 

55 """Get exception from a completed task, or None.""" 

56 

57 @abstractmethod (empty)

58 def is_task_cancelled( (empty)

59 self, task: Union[asyncio.Task, concurrent.futures.Future] 

60 ) -> bool: 

61 """Check if a task was cancelled.""" 

62 

63 # ------------------------------------------------------------------ 

64 # Shared pure-logic helpers (no I/O, no await) 

65 # ------------------------------------------------------------------ 

66 

67 def _handle_exception( (empty)

68 self, 

69 context: ExecutionContext, 

70 config: ScheduleConfig, 

71 finished: Union[asyncio.Task, concurrent.futures.Future], 

72 can_add_speed_up: bool, 

73 ) -> tuple[bool, bool]: 

74 """Handle task exception. 

75 

76 Args: 

77 context: Execution context. 

78 config: Schedule configuration. 

79 finished: Completed task with exception. 

80 can_add_speed_up: Whether speed can be increased. 

81 

82 Returns: 

83 Tuple of (is_retryable, did_increase_speed). 

84 """ 

85 exception = self.get_task_exception(finished) (empty)

86 context.result_exception = exception (empty)

87 context.record_exception(exception) (empty)

88 

89 # Never catch system-level exceptions 

90 if isinstance(exception, (KeyboardInterrupt, SystemExit)): (empty)

91 raise exception (empty)

92 

93 if any(isinstance(exception, t) for t in config.exception_types): (empty)

94 did_increase = False (empty)

95 if can_add_speed_up: (empty)

96 context.increase_speed() (empty)

97 did_increase = True (empty)

98 

99 if self.overload_checker.check(exception): (empty)

100 context.reset_speed() (empty)

101 

102 return True, did_increase (empty)

103 

104 return False, False (empty)

105 

106 @staticmethod (empty)

107 def _compute_cancel_wait( (empty)

108 config: ScheduleConfig, context: ExecutionContext 

109 ) -> float: 

110 """Compute wait time for cancel_tasks calls.""" 

111 return config.adjust_wait_time( (empty)

112 config.retry_interval, context.elapsed, config.timeout 

113 ) 

114 

115 @staticmethod (empty)

116 def _log_finish(context: ExecutionContext) -> None: (empty)

117 """Emit debug log on execution finish.""" 

118 logger.debug( (empty)

119 "[schedule] %s execute finish, " 

120 "executeTimes: %d, speedUpMultiply: %d, " 

121 "consumeTime: %.4f, exceptions: %s", 

122 context.func.__qualname__, 

123 context.cur_times, 

124 context.cur_speed_up_multiply, 

125 context.elapsed, 

126 context.result_exception_list, 

127 ) 

128 

129 @staticmethod (empty)

130 def _return_result( (empty)

131 context: ExecutionContext, 

132 config: ScheduleConfig, 

133 args: tuple, 

134 kwargs: dict, 

135 ) -> Any: 

136 """Return the final result, applying default_result if needed.""" 

137 if context.result_exception is not None: (empty)

138 if config.default_result is not _sentinel_none: (empty)

139 if callable(config.default_result): (empty)

140 return config.default_result(*args, **kwargs) (empty)

141 return config.default_result (empty)

142 raise context.result_exception (empty)

143 return context.result (empty)

144 

145 

146# Re-export AsyncRunner and SyncRunner for backward compatibility. 

147from .async_runner import AsyncRunner # noqa: F401, E402 (empty)

148from .sync_runner import SyncRunner # noqa: F401, E402 (empty)