Skip to content

BaseReload

faststream.cli.supervisors.basereload.BaseReload #

BaseReload(
    target: DecoratedCallable,
    args: Tuple[Any, ...],
    reload_delay: Optional[float] = 0.5,
)

A base class for implementing a reloader process.

METHOD DESCRIPTION
run

Runs the reloader process.
startup

Performs startup operations for the reloader process.
restart

Restarts the process.
shutdown

Shuts down the reloader process.
_stop_process

Stops the spawned process.
_start_process

Starts the spawned process.
should_restart

Determines whether the process should be restarted.

Initialize a class instance.

PARAMETER DESCRIPTION
target

The target callable object

TYPE: DecoratedCallable

args

Tuple of arguments to be passed to the target callable

TYPE: Tuple[Any, ...]

reload_delay

Optional delay in seconds before reloading the target callable (default is 0.5 seconds)

TYPE: Optional[float] DEFAULT: 0.5

RETURNS DESCRIPTION
None

None
Source code in faststream/cli/supervisors/basereload.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    target: DecoratedCallable,
    args: Tuple[Any, ...],
    reload_delay: Optional[float] = 0.5,
) -> None:
    """Initialize a class instance.

    Args:
        target: The target callable object
        args: Tuple of arguments to be passed to the target callable
        reload_delay: Optional delay in seconds before reloading the target callable (default is 0.5 seconds)

    Returns:
        None

    """
    self._target = target
    self._args = args

    self.should_exit = threading.Event()
    self.pid = os.getpid()
    self.reload_delay = reload_delay

    set_exit(lambda *_: self.should_exit.set())

pid instance-attribute #

pid: int = getpid()

reload_delay instance-attribute #

reload_delay: Optional[float] = reload_delay

reloader_name class-attribute instance-attribute #

reloader_name: str = ''

should_exit instance-attribute #

should_exit: Event = Event()

restart #

restart() -> None
Source code in faststream/cli/supervisors/basereload.py 
96
97
98
99
def restart(self) -> None:
    self._stop_process()
    logger.info("Process successfully reloaded")
    self._process = self._start_process()

run #

run() -> None
Source code in faststream/cli/supervisors/basereload.py 
85
86
87
88
89
90
def run(self) -> None:
    self.startup()
    while not self.should_exit.wait(self.reload_delay):
        if self.should_restart():  # pragma: no branch
            self.restart()
    self.shutdown()

should_restart #

should_restart() -> bool
Source code in faststream/cli/supervisors/basereload.py 
114
115
def should_restart(self) -> bool:
    raise NotImplementedError("Reload strategies should override should_restart()")

shutdown #

shutdown() -> None
Source code in faststream/cli/supervisors/basereload.py 
101
102
103
def shutdown(self) -> None:
    self._stop_process()
    logger.info(f"Stopping reloader process [{self.pid}]")

startup #

startup() -> None
Source code in faststream/cli/supervisors/basereload.py 
92
93
94
def startup(self) -> None:
    logger.info(f"Started reloader process [{self.pid}] using {self.reloader_name}")
    self._process = self._start_process()