create_interval_daemon(
lock_file: str,
callback: Callable[[], None],
interval: float,
verbose: bool,
error_callback: Callable[[BaseException], None]
) -> Callable[[], None]Creates a daemon that calls the callback function at fixed intervals.
Arguments: - callback: The function to call at each interval. - interval: Number of seconds between callbacks. - verbose: Whether to print status messages.
Returns:: A (start, stop) function pair for the daemon.
import tempfile
from pathlib import Path
def my_callback():
print("Interval callback!")
start, stop, wait_for_stop = create_interval_daemon(tempfile.mktemp(), my_callback, interval=2.0, verbose=True)
start() # Will print "Interval callback!" every 2 seconds
stop() # Stops the daemon[interval_daemon] Daemon started with 2.0s interval
Interval callback!create_watchdog_daemon(
folder_paths: Union[str, List[str]],
lock_file: str,
callback: Callable[[object], None],
recursive: bool,
verbose: bool,
rate_limit: float
) -> Callable[[], None]Starts a background daemon that watches folder_paths for changes.
Calls callback(event) whenever a file changes.
Arguments: - folder_paths: A path or list of paths to watch. - callback: The function to call when a file changes. Receives the event as argument. - recursive: Whether to watch folders recursively. - lock_file: Optional path to a lock file to ensure only one daemon is running. - rate_limit: Minimum number of seconds between callbacks.
Returns:: A (start, stop) function pair for the daemon.
import tempfile
from pathlib import Path
def my_callback(event):
print("Folder changed!", event)
lock_file_path = tempfile.mktemp()
daemon_start, daemon_stop = create_watchdog_daemon("/bin", lock_file_path, my_callback, verbose=True, recursive=False)
daemon_start()
time.sleep(0.1)
_daemon_start, _daemon_stop = create_watchdog_daemon("/bin", lock_file_path, my_callback, verbose=True, recursive=False)
assert _daemon_start is None
daemon_stop()
_daemon_start, _daemon_stop = create_watchdog_daemon(["/bin", "/"], lock_file_path, my_callback, verbose=True, recursive=False)
assert _daemon_start is not None
_daemon_start()
time.sleep(0.1)
_daemon_stop()[watchdog_daemon] Daemon started.
[watchdog_daemon] Lock file exists at /var/folders/22/k6clk3m10258k07mjcnlswt40000gn/T/tmpchehy2zc. Daemon will not start.
[watchdog_daemon] Daemon started.