feat(hotreload): Add more events (#50)
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 35s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 38s

# More HotReload Events
<!-- Create a new issue, if it doesn't exist yet -->
Currently, HotReload only supports file modification events. It should also support file moves, and some other event types.

- [x] By submitting this pull request, I permit cswimr to license my work under
  the [Mozilla Public License Version 2.0](https://www.coastalcommits.com/cswimr/SeaCogs/src/branch/main/LICENSE).

Reviewed-on: #50
This commit is contained in:
cswimr 2025-01-25 19:25:29 -05:00
parent 5384809780
commit 5adc7a2c7b
Signed by: CoastalCommits
GPG key ID: 7E73189F651A553F

View file

@ -1,12 +1,13 @@
from asyncio import run_coroutine_threadsafe from asyncio import run_coroutine_threadsafe
from pathlib import Path from pathlib import Path
from typing import Sequence
from red_commons.logging import RedTraceLogger, getLogger from red_commons.logging import RedTraceLogger, getLogger
from redbot.core import commands from redbot.core import commands
from redbot.core.bot import Red from redbot.core.bot import Red
from redbot.core.core_commands import CoreLogic from redbot.core.core_commands import CoreLogic
from redbot.core.utils.chat_formatting import bold, humanize_list from redbot.core.utils.chat_formatting import bold, humanize_list
from watchdog.events import FileSystemEvent, RegexMatchingEventHandler from watchdog.events import FileSystemEvent, FileSystemMovedEvent, RegexMatchingEventHandler
from watchdog.observers import Observer from watchdog.observers import Observer
@ -15,7 +16,7 @@ class HotReload(commands.Cog):
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"] __author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs" __git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.0.0" __version__ = "1.1.0"
__documentation__ = "https://seacogs.coastalcommits.com/hotreload/" __documentation__ = "https://seacogs.coastalcommits.com/hotreload/"
def __init__(self, bot: Red) -> None: def __init__(self, bot: Red) -> None:
@ -74,18 +75,33 @@ class HotReloadHandler(RegexMatchingEventHandler):
self.path: Path = path self.path: Path = path
self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload.Observer") self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload.Observer")
def on_modified(self, event: FileSystemEvent) -> None: def on_any_event(self, event: FileSystemEvent) -> None:
"""Handle file modification events.""" """Handle filesystem events."""
if event.is_directory: if event.is_directory:
return return
relative_path = Path(event.src_path).relative_to(self.path)
package_name = relative_path.parts[0]
self.logger.info(f"File {'/'.join(relative_path.parts[1:])} in the cog {package_name} has been modified.")
run_coroutine_threadsafe(self.reload_cog(package_name), loop=self.bot.loop)
async def reload_cog(self, cog_name: str) -> None: allowed_events = ("moved", "deleted", "created", "modified")
if event.event_type not in allowed_events:
return
relative_src_path = Path(event.src_path).relative_to(self.path)
src_package_name = relative_src_path.parts[0]
cogs_to_reload = [src_package_name]
if isinstance(event, FileSystemMovedEvent):
dest = f" to {event.dest_path}"
relative_dest_path = Path(event.dest_path).relative_to(self.path)
cogs_to_reload.append(relative_dest_path.parts[0])
else:
dest = ""
self.logger.info(f"File {event.src_path} has been {event.event_type}{dest}.")
run_coroutine_threadsafe(self.reload_cogs(cogs_to_reload), loop=self.bot.loop)
async def reload_cogs(self, cog_names: Sequence[str]) -> None:
"""Reload modified cog.""" """Reload modified cog."""
core_logic = CoreLogic(bot=self.bot) core_logic = CoreLogic(bot=self.bot)
self.logger.info(f"Reloading {cog_name} cog.") self.logger.info(f"Reloading cogs: {humanize_list(cog_names, style='unit')}")
await core_logic._reload(pkg_names=(cog_name,)) await core_logic._reload(pkg_names=cog_names)
self.logger.info(f"Reloaded {cog_name} cog.") self.logger.info(f"Reloaded cogs: {humanize_list(cog_names, style='unit')}")