Coverage for faststream / _internal / cli / supervisors / basereload.py: 98%

1import os (empty)

2import threading (empty)

3from multiprocessing.context import SpawnProcess (empty)

4from typing import TYPE_CHECKING (empty)

5 

6from faststream._internal.cli.supervisors.utils import get_subprocess, set_exit (empty)

7from faststream._internal.logger import logger (empty)

8 

9if TYPE_CHECKING: (empty)

10 from faststream._internal.cli.dto import RunArgs, RunFunction 

11 

12 

13class BaseReload: (empty)

14 """A base class for implementing a reloader process.""" 

15 

16 _process: SpawnProcess (empty)

17 

18 reload_delay: float | None (empty)

19 should_exit: threading.Event (empty)

20 pid: int (empty)

21 reloader_name: str = "" (empty)

22 

23 def __init__( (empty)

24 self, 

25 target: "RunFunction", 

26 args: "RunArgs", 

27 reload_delay: float | None = 0.5, 

28 ) -> None: 

29 self._target = target (empty)

30 self._args = args (empty)

31 

32 self.should_exit = threading.Event() (empty)

33 self.pid = os.getpid() (empty)

34 self.reload_delay = reload_delay (empty)

35 

36 set_exit(lambda *_: self.should_exit.set(), sync=True) (empty)

37 

38 def run(self) -> None: (empty)

39 self.startup() (empty)

40 while not self.should_exit.wait(self.reload_delay): (empty)

41 if self.should_restart(): # pragma: no branch (empty)

42 self.restart() (empty)

43 self.shutdown() (empty)

44 

45 def startup(self) -> None: (empty)

46 logger.info( (empty)

47 "Started reloader process [%s] using %s", 

48 self.pid, 

49 self.reloader_name, 

50 ) 

51 self._process = self.start_process() (empty)

52 

53 def restart(self) -> None: (empty)

54 self._stop_process() (empty)

55 logger.info("Process successfully reloaded") (empty)

56 self._process = self.start_process() (empty)

57 

58 def shutdown(self) -> None: (empty)

59 self._stop_process() (empty)

60 logger.info("Stopping reloader process [%s]", self.pid) (empty)

61 

62 def _stop_process(self) -> None: (empty)

63 self._process.terminate() (empty)

64 self._process.join() (empty)

65 

66 def start_process(self, worker_id: int | None = None) -> SpawnProcess: (empty)

67 self._args.extra_options["worker_id"] = worker_id (empty)

68 process = get_subprocess(target=self._target, args=(self._args,)) (empty)

69 process.start() (empty)

70 return process (empty)

71 

72 def should_restart(self) -> bool: (empty)

73 msg = "Reload strategies should override should_restart()" 

74 raise NotImplementedError(msg)