from typing import Any from typing import TYPE_CHECKING from typing import Union from discord.ext.commands.errors import ExtensionAlreadyLoaded from discord.ext.commands.errors import ExtensionFailed from discord.ext.commands.errors import ExtensionNotFound from discord.ext.commands.errors import NoEntryPointError if TYPE_CHECKING: from tunebot.abc import BasePluginInstance from tunebot.plugins import ServiceEvent from bot import TuneBot AnyDict = dict[Any, Any] class SimplePluginManager: _plugins: dict[str, "BasePluginInstance"] = {} def __init__(self, bot: "TuneBot") -> None: self.bot = bot def get_plugin(self, name: str) -> Union["BasePluginInstance", None]: """ Retrieves the corresponding `BasePluginInstance` if it exists Returns: Union["BasePluginInstance", None]: [description] """ return self._plugins.get(name) def remove_plugin(self, name: str): """ Unloads/Removes all components related to the `BasePluginInstance` Args: name (str): [description] """ plugin = self.get_plugin(name) if not plugin: return for cog in plugin.cogs: try: self.bot.unload_extension(cog) except Exception: pass del self._plugins[name] def enable_plugin(self, name: str, plugin: "BasePluginInstance"): """ Loads/Activates all cogs/services included within the plugin Args: plugin_name (str): [description] services (list[): [description] cog_names (list[str]): [description] """ self._plugins[name] = plugin for cog_name in plugin.cogs: self.bot.load_extension(cog_name) async def dispatch(self, event: "ServiceEvent", payload: AnyDict = {}): """ Dispatches an event to all registered services Args: event (ServiceEvent): [description] payload (AnyDict, optional): [description]. Defaults to {}. """ for plugin in self._plugins.values(): for service in plugin.services: await service.on_dispatch(event, payload) __all__ = ("SimplePluginManager",)