Coverage for src/taipanstack/resilience/watchdogs/_base.py: 100%

1""" 

2Base watcher abstract class. 

3 

4Provides a shared lifecycle (start/stop) for all background 

5watchdog tasks running on the asyncio event loop. 

6""" 

7 

8import asyncio 

9import logging 

10import math 

11from abc import ABC, abstractmethod 

12 

13from taipanstack.core.result import Err, Ok, Result 

14 

15logger = logging.getLogger("taipanstack.resilience.watchdogs") 

16 

17 

18class BaseWatcher(ABC): 

19 """Abstract base for background watchdog tasks. 

20 

21 Subclasses implement ``_run`` which is called repeatedly at 

22 ``_interval`` seconds until ``stop`` is called. 

23 

24 Args: 

25 interval: Seconds between each poll cycle. 

26 

27 Example: 

28 >>> class MyWatcher(BaseWatcher): 

29 ... async def _run(self) -> None: 

30 ... print("checking...") 

31 >>> watcher = MyWatcher(interval=5.0) 

32 >>> await watcher.start() 

33 

34 """ 

35 

36 def __init__(self, *, interval: float = 5.0) -> None: 

37 """Initialize the base watcher. 

38 

39 Args: 

40 interval: Seconds between each poll cycle. 

41 

42 """ 

43 if ( 

44 not isinstance(interval, (int, float)) 

45 or not math.isfinite(interval) 

46 or interval <= 0 

47 ): 

48 raise ValueError("interval must be a finite positive number") 

49 self._interval = interval 

50 self._stop_event: asyncio.Event = asyncio.Event() 

51 self._task: asyncio.Task[None] | None = None 

52 

53 @property 

54 def is_running(self) -> bool: 

55 """Return ``True`` if the background task is active.""" 

56 return self._task is not None and not self._task.done() 

57 

58 async def start(self) -> Result[None, Exception]: 

59 """Start the background watcher loop. 

60 

61 Returns: 

62 ``Ok(None)`` on success, ``Err`` if already running. 

63 

64 """ 

65 if self.is_running: 

66 return Err(RuntimeError("Watcher is already running")) 

67 

68 self._stop_event.clear() 

69 self._task = asyncio.create_task(self._loop()) 

70 logger.info("%s started (interval=%.1fs)", type(self).__name__, self._interval) 

71 return Ok(None) 

72 

73 async def stop(self) -> None: 

74 """Signal the watcher to stop and wait for it to finish.""" 

75 self._stop_event.set() 

76 if self._task is not None: 

77 try: 

78 await asyncio.wait_for(self._task, timeout=self._interval + 1.0) 

79 except (TimeoutError, asyncio.CancelledError): 

80 self._task.cancel() 

81 self._task = None 

82 logger.info("%s stopped", type(self).__name__) 

83 

84 async def _loop(self) -> None: 

85 """Internal loop that calls ``_run`` at each interval.""" 

86 while not self._stop_event.is_set(): 

87 try: 

88 await self._run() 

89 except Exception as exc: 

90 logger.exception( 

91 "%s encountered an error", 

92 type(self).__name__, 

93 exc_info=exc, 

94 ) 

95 try: 

96 await asyncio.wait_for(self._stop_event.wait(), timeout=self._interval) 

97 except TimeoutError: 

98 continue 

99 

100 @abstractmethod 

101 async def _run(self) -> None: 

102 """Execute a single poll cycle. 

103 

104 Subclasses must override this with the actual monitoring logic. 

105 """ 

106 ... 

107 

108 def _get_extra_repr(self) -> dict[str, object]: 

109 """Return extra fields for ``__repr__``. 

110 

111 Subclasses can override to add custom fields. 

112 """ 

113 return {} 

114 

115 def __repr__(self) -> str: 

116 """Return a developer-friendly representation.""" 

117 extras = self._get_extra_repr() 

118 parts = [f"interval={self._interval}"] 

119 parts.extend(f"{k}={v}" for k, v in extras.items()) 

120 return f"{type(self).__name__}({', '.join(parts)})"