Coverage for qreward / utils / schedule.py: 100.00%

1"""Compatibility layer — re-exports from qreward.utils.scheduler. 

2 

3This module preserves backward compatibility so that existing code using 

4from qreward.utils.schedule import ... continues to work. The canonical 

5implementation now lives in qreward.utils.scheduler. 

6""" 

7 

8import asyncio (empty)

9import concurrent.futures (empty)

10from collections.abc import Sequence (empty)

11from typing import List, Optional (empty)

12 

13from qreward.utils.scheduler import ( # noqa: F401 — public re-exports (empty)

14 LimiterPool, 

15 RunningTaskPool, 

16 schedule, 

17) 

18from qreward.utils.scheduler.base import _CancelledErrorGroups # noqa: F401 (empty)

19from qreward.utils.scheduler.config import _sentinel_none # noqa: F401 (empty)

20from qreward.utils.scheduler.overload import OverloadChecker (empty)

21 

22 

23# --------------------------------------------------------------------------- 

24# Legacy standalone helpers kept for backward-compatible test imports. 

25# The scheduler/ package uses class methods instead, but existing tests 

26# reference these free functions directly. 

27# --------------------------------------------------------------------------- 

28 

29 

30async def _cancel_async_task( (empty)

31 pending: Sequence[asyncio.Task], 

32 done: List[asyncio.Task], 

33 retry_interval: Optional[float], 

34) -> None: 

35 """Cancel remaining async tasks (legacy standalone helper).""" 

36 while len(done) > 0: (empty)

37 try: (empty)

38 _ = done.pop() (empty)

39 except _CancelledErrorGroups: (empty)

40 pass (empty)

41 if len(pending) > 0: (empty)

42 for task in pending: (empty)

43 if not task.done(): (empty)

44 task.cancel() (empty)

45 try: (empty)

46 await asyncio.wait_for( (empty)

47 fut=asyncio.gather(*pending, return_exceptions=True), 

48 timeout=retry_interval, 

49 ) 

50 except _CancelledErrorGroups: (empty)

51 pass (empty)

52 

53 

54def _cancel_sync_task( (empty)

55 not_done: Sequence[concurrent.futures.Future], 

56 done: List, 

57 retry_interval: Optional[float], 

58) -> None: 

59 """Cancel remaining sync tasks (legacy standalone helper).""" 

60 while len(done) > 0: (empty)

61 try: (empty)

62 _ = done.pop() (empty)

63 except _CancelledErrorGroups: (empty)

64 pass (empty)

65 for task in not_done: (empty)

66 if not task.done(): (empty)

67 task.cancel() (empty)

68 if len(not_done) > 0: (empty)

69 try: (empty)

70 concurrent.futures.wait( (empty)

71 not_done, 

72 timeout=retry_interval, 

73 return_when=concurrent.futures.ALL_COMPLETED, 

74 ) 

75 except _CancelledErrorGroups: (empty)

76 pass (empty)

77 

78 

79def _overload_check(exception: BaseException) -> bool: (empty)

80 """Check if exception indicates server overload (legacy standalone helper).""" 

81 return OverloadChecker.check(exception) (empty)

82 

83 

84def _get_max_wait_time( (empty)

85 basic_wait_time: float, 

86 has_wait_time: float, 

87 max_wait_time: float, 

88) -> float: 

89 """Calculate max wait time considering timeout (legacy standalone helper). 

90 

91 Delegates to ScheduleConfig.get_max_wait_time with a dummy config instance. 

92 """ 

93 if basic_wait_time < 0: (empty)

94 basic_wait_time = 0.01 (empty)

95 if max_wait_time <= 0 or basic_wait_time + has_wait_time < max_wait_time: (empty)

96 return basic_wait_time (empty)

97 if has_wait_time > max_wait_time: (empty)

98 return 0.01 (empty)

99 return max_wait_time - has_wait_time (empty)

100 

101 

102__all__ = [ (empty)

103 "LimiterPool", 

104 "RunningTaskPool", 

105 "_CancelledErrorGroups", 

106 "_cancel_async_task", 

107 "_cancel_sync_task", 

108 "_get_max_wait_time", 

109 "_overload_check", 

110 "_sentinel_none", 

111 "schedule", 

112]