Watcher

Powers grace run --watch: watches your ./bot directory for Python file changes and reloads the corresponding extension modules without restarting the bot.

grace.watcher.Watcher

Watcher(callback)

Wrapper around the watchdog observer that watches a specified directory (./bot) for Python file changes and manages event handling.

PARAMETER	DESCRIPTION
callback	Async, no-argument callback invoked after a reload is handled. TYPE: ReloadCallback

Source code in grace/watcher.py

def __init__(self, callback: ReloadCallback) -> None:
    self.callback: ReloadCallback = callback
    self.observer: BaseObserver = Observer()
    self.watch_path: str = "./bot"

    self.observer.schedule(
        BotEventHandler(self.callback, self.watch_path),
        self.watch_path,
        recursive=True,
    )

callback instance-attribute

callback = callback

observer instance-attribute

observer = Observer()

watch_path instance-attribute

watch_path = './bot'

start

start()

Starts the file system observer.

Source code in grace/watcher.py

def start(self) -> None:
    """Starts the file system observer."""
    info("Starting file watcher...")
    self.observer.start()

stop

stop()

Stops the file system observer and waits for it to shut down.

Source code in grace/watcher.py

def stop(self) -> None:
    """Stops the file system observer and waits for it to shut down."""
    info("Stopping file watcher...")
    self.observer.stop()
    self.observer.join()

grace.watcher.BotEventHandler

BotEventHandler(callback, base_path)

Bases: FileSystemEventHandler

Handles file events in the bot directory and calls the provided async callback.

PARAMETER	DESCRIPTION
callback	Async function to call with the module name. TYPE: ReloadCallback
base_path	Directory path to watch. TYPE: Union[Path, str]

Source code in grace/watcher.py

def __init__(self, callback: ReloadCallback, base_path: Union[Path, str]):
    self.callback = callback
    self.bot_path = Path(base_path).resolve()

callback instance-attribute

callback = callback

bot_path instance-attribute

bot_path = Path(base_path).resolve()

path_to_module_name

path_to_module_name(path)

Converts a file path to a Python module name.

PARAMETER	DESCRIPTION
path	Full path to the Python file. TYPE: Path

RETURNS	DESCRIPTION
str	Dotted module path (e.g., 'bot.module.sub').

Source code in grace/watcher.py

def path_to_module_name(self, path: Path) -> str:
    """
    Converts a file path to a Python module name.

    :param path: Full path to the Python file.
    :type path: Path
    :return: Dotted module path (e.g., 'bot.module.sub').
    :rtype: str
    """
    relative_path = path.resolve().relative_to(self.bot_path)
    parts = relative_path.with_suffix("").parts
    return ".".join(["bot"] + list(parts))

reload_module

reload_module(module_name)

Reloads a module if it's already in sys.modules.

PARAMETER	DESCRIPTION
module_name	Dotted module name to reload. TYPE: str

Source code in grace/watcher.py

def reload_module(self, module_name: str) -> None:
    """
    Reloads a module if it's already in sys.modules.

    :param module_name: Dotted module name to reload.
    :type module_name: str
    """
    try:
        if module_name in sys.modules:
            info(f"Reloading module '{module_name}'")
            importlib.reload(sys.modules[module_name])
    except Exception as e:
        error(f"Failed to reload module {module_name}: {e}")

run_callback

run_callback()

Runs a coroutine callback in the current or a new event loop.

Source code in grace/watcher.py

def run_callback(self) -> None:
    """Runs a coroutine callback in the current or a new event loop."""
    try:
        loop = asyncio.get_running_loop()
        asyncio.ensure_future(self.callback())
    except RuntimeError:
        asyncio.run(self.callback())

on_modified

on_modified(event)

Handles modified Python files by reloading them and calling the callback.

PARAMETER	DESCRIPTION
event	The filesystem event. TYPE: FileSystemEvent

Source code in grace/watcher.py

def on_modified(self, event: FileSystemEvent) -> None:
    """
    Handles modified Python files by reloading them and calling the callback.

    :param event: The filesystem event.
    :type event: FileSystemEvent
    """
    try:
        if event.is_directory:
            return

        src_path: str = str(event.src_path)
        module_path: Path = Path(src_path)

        if module_path.suffix != ".py":
            return

        module_name = self.path_to_module_name(module_path)
        if not module_name:
            return

        self.reload_module(module_name)
        self.run_callback()
    except Exception as e:
        error(f"Failed to reload module: {e}")

on_deleted

on_deleted(event)

Handles deleted Python files by calling the callback with the module name.

PARAMETER	DESCRIPTION
event	The filesystem event. TYPE: FileSystemEvent

Source code in grace/watcher.py

def on_deleted(self, event: FileSystemEvent) -> None:
    """
    Handles deleted Python files by calling the callback with the module name.

    :param event: The filesystem event.
    :type event: FileSystemEvent
    """
    try:
        src_path: str = str(event.src_path)
        module_path: Path = Path(src_path)

        if module_path.suffix != ".py":
            return

        module_name = self.path_to_module_name(module_path)
        if not module_name:
            return

        self.run_callback()
    except Exception as e:
        error(f"Failed to reload module {module_name}: {e}")