✨ Patch feature

Author: Paillat-dev

main.py

@@ -2,6 +2,7 @@
 import importlib
 import importlib.util
 import asyncio
+import os
 
 from inspect import signature
 from quart import Quart
@@ -10,23 +11,33 @@
 from src.logging import logger, patch
 from os.path import splitext, basename
 from types import ModuleType
-from typing import Any, Coroutine, Callable
+from typing import Any, Callable, TypedDict, TYPE_CHECKING
+from collections.abc import Coroutine
 from src.utils import validate_module, unzip_extensions
 
+if TYPE_CHECKING:
+    FunctionConfig = TypedDict("FunctionConfig", {"enabled": bool})
+    FunctionlistType = list[tuple[Callable[..., Any], FunctionConfig]]
+
 
-# noinspection PyUnusedLocal
 async def start_bot(bot: discord.Bot, token: str):
-    await bot.start(config["bot"]["token"])
+    await bot.start(token)
 
 
 async def start_backend(app: Quart, bot: discord.Bot, token: str):
     from hypercorn.config import Config
     from hypercorn.logging import Logger as HypercornLogger
-    from hypercorn.asyncio import serve
+    from hypercorn.asyncio import serve  # pyright: ignore [reportUnknownVariableType]
 
     class CustomLogger(HypercornLogger):
-        def __init__(self, *args, **kwargs) -> None:
-            super().__init__(*args, **kwargs)
+        def __init__(
+            self,
+            *args,  # pyright: ignore [reportUnknownParameterType,reportMissingParameterType]
+            **kwargs,  # pyright: ignore [reportUnknownParameterType,reportMissingParameterType]
+        ) -> None:
+            super().__init__(
+                *args, **kwargs  # pyright: ignore [reportUnknownArgumentType]
+            )
             if self.error_logger:
                 patch(self.error_logger)
             if self.access_logger:
@@ -42,7 +53,7 @@ def __init__(self, *args, **kwargs) -> None:
     patch("hypercorn.error")
 
 
-def setup_func(func: callable, **kwargs) -> Any:
+def setup_func(func: Callable[..., Any], **kwargs: Any) -> Any:
     parameters = signature(func).parameters
     func_kwargs = {}
     for name, parameter in parameters.items():
@@ -55,23 +66,35 @@ def setup_func(func: callable, **kwargs) -> Any:
     return func(**func_kwargs)
 
 
-async def main():
-    assert (config.get("bot", {}) or {}).get(
-        "token"
-    ), f"No bit token provided in config"
-    unzip_extensions()
+async def load_and_run_patches():
+    for patch_file in iglob("src/extensions/*/patch.py"):
+        extension = os.path.basename(os.path.dirname(patch_file))
+        if config["extensions"].get(extension, {}).get("enabled", False):
+            logger.info(f"Loading patch for extension {extension}")
+            spec = importlib.util.spec_from_file_location(
+                f"src.extensions.{extension}.patch", patch_file
+            )
+            if not spec or not spec.loader:
+                continue
+            patch_module = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(patch_module)
+            if hasattr(patch_module, "patch") and callable(patch_module.patch):
+                await asyncio.to_thread(patch_module.patch)
 
-    bot_functions: list[tuple[Callable, dict[Any]]] = []
-    back_functions: list[tuple[Callable, dict[Any]]] = []
-    startup_functions: list[tuple[Callable, dict[Any]]] = []
+
+def load_extensions() -> (
+    tuple["FunctionlistType", "FunctionlistType", "FunctionlistType"]
+):
+    bot_functions: "FunctionlistType" = []
+    back_functions: "FunctionlistType" = []
+    startup_functions: "FunctionlistType" = []
 
     for extension in iglob("src/extensions/*"):
         name = splitext(basename(extension))[0]
         its_config = config["extensions"].get(name, {})
         logger.info(f"Loading extension {name}")
         module: ModuleType = importlib.import_module(f"src.extensions.{name}")
         if not its_config:
-            # use default config if not present
             its_config = module.default
             config["extensions"][name] = its_config
         if not its_config["enabled"]:
@@ -87,34 +110,64 @@ async def main():
         if hasattr(module, "on_startup") and callable(module.on_startup):
             startup_functions.append((module.on_startup, its_config))
 
-    startup_coros: list[Coroutine] = []
-    coros: list[Coroutine] = []
-
-    bot = None
-    back_bot = None
-    app = None
-
-    if bot_functions:
-        bot = discord.Bot(intents=discord.Intents.default())
-        for function, its_config in bot_functions:
-            setup_func(function, bot=bot, config=its_config)
-        coros.append(start_bot(bot, config["bot"]["token"]))
-
-    if back_functions:
-        back_bot = discord.Bot(intents=discord.Intents.default())
-        app = Quart("backend")
-        for function, its_config in back_functions:
-            setup_func(function, app=app, bot=back_bot, config=its_config)
-        coros.append(start_backend(app, back_bot, config["bot"]["token"]))
+    return bot_functions, back_functions, startup_functions
+
+
+async def setup_and_start_bot(
+    bot_functions: "FunctionlistType",
+):
+    bot = discord.Bot(intents=discord.Intents.default())
+    for function, its_config in bot_functions:
+        setup_func(function, bot=bot, config=its_config)
+    await start_bot(bot, config["bot"]["token"])
+
+
+async def setup_and_start_backend(
+    back_functions: "FunctionlistType",
+):
+    back_bot = discord.Bot(intents=discord.Intents.default())
+    app = Quart("backend")
+    for function, its_config in back_functions:
+        setup_func(function, app=app, bot=back_bot, config=its_config)
+    await start_backend(app, back_bot, config["bot"]["token"])
+
+
+async def run_startup_functions(
+    startup_functions: "FunctionlistType",
+    app: Quart | None,
+    back_bot: discord.Bot | None,
+):
+    startup_coros = [
+        setup_func(function, app=app, bot=back_bot, config=its_config)
+        for function, its_config in startup_functions
+    ]
+    await asyncio.gather(*startup_coros)
+
+
+async def main(run_bot: bool = True, run_backend: bool = True):
+    assert config.get("bot", {}).get("token"), "No bot token provided in config"
+    unzip_extensions()
+
+    await load_and_run_patches()
+
+    bot_functions, back_functions, startup_functions = load_extensions()
+
+    coros: list[Coroutine[Any, Any, Any]] = []
+    if bot_functions and run_bot:
+        coros.append(setup_and_start_bot(bot_functions))
+    if back_functions and run_backend:
+        coros.append(setup_and_start_backend(back_functions))
     assert coros, "No extensions to run"
 
     if startup_functions:
-        for function, its_config in startup_functions:
-            startup_coros.append(
-                setup_func(function, app=app, bot=back_bot, config=its_config)
-            )
+        app = Quart("backend") if (back_functions and run_backend) else None
+        back_bot = (
+            discord.Bot(intents=discord.Intents.default())
+            if (back_functions and run_backend)
+            else None
+        )
+        await run_startup_functions(startup_functions, app, back_bot)
 
-    await asyncio.gather(*startup_coros)
     await asyncio.gather(*coros)
 
     store_config()

readme.md

@@ -118,6 +118,64 @@ schema = {
 ```
 We really encourage you to follow these instructions, even if you’re coding privately, as it will make your code more readable and maintainable in the long run.
 
+## Using Patch Files
+
+Botkit supports the use of patch files to modify or extend the functionality of the bot or its dependencies before the main extension code runs. This is particularly useful for applying global changes or monkey-patching existing classes.
+
+### How It Works
+
+1. Create a file named `patch.py` in your extension's directory.
+2. Define a `patch()` function in this file. This function will be called before the extension is loaded.
+3. The `patch()` function can modify global state, patch classes, or perform any other setup needed.
+
+### Example: Error Handling Patch
+
+Here's an example from the `nice-errors` extension that demonstrates how to use a patch file to enhance error handling:
+
+```python
+# nice-errors/patch.py
+
+import discord
+from discord import Interaction
+from discord.ui import Item
+from typing_extensions import override
+
+def patch():
+    class PatchedView(discord.ui.View):
+        @override
+        async def on_error(
+            self,
+            error: Exception,
+            item: Item,
+            interaction: Interaction,
+        ) -> None:
+            if not isinstance(error, discord.Forbidden):
+                await interaction.respond(
+                    "Whoops! An error occurred while executing this command",
+                    ephemeral=True,
+                )
+                raise error
+            await interaction.respond(
+                f"Whoops! I don't have permission to do that\n`{error.args[0].split(':')[-1].strip()}`",
+                ephemeral=True,
+            )
+
+    discord.ui.View = PatchedView
+
+```
+
+This patch modifies the `discord.ui.View` class to provide more user-friendly error messages. It catches exceptions and responds to the user with an appropriate message, enhancing the overall user experience.
+
+### When to Use Patch Files
+
+Patch files are powerful but should be used judiciously. They are best suited for:
+
+1. Applying global changes that affect multiple parts of your bot.
+2. Modifying third-party libraries when you can't or don't want to fork them.
+3. Implementing cross-cutting concerns like logging or error handling.
+
+Remember that patches are applied early in the bot's lifecycle, so they can affect all subsequent code. Use them carefully and document their effects clearly.
+
 ## Using scripts
 
 ### `check-listings`

src/extensions/nice-errors/patch.py

@@ -0,0 +1,29 @@
+import discord
+from discord import Interaction
+from discord.ui import Item
+from typing_extensions import override
+
+
+def patch():
+
+    class PatchedView(discord.ui.View):
+
+        @override
+        async def on_error(
+            self,
+            error: Exception,
+            item: Item,  # pyright: ignore[reportMissingTypeArgument,reportUnknownParameterType]
+            interaction: Interaction,
+        ) -> None:
+            if not isinstance(error, discord.Forbidden):
+                await interaction.respond(  # pyright: ignore[reportUnknownMemberType]
+                    "Whoops! An error occurred while executing this command",
+                    ephemeral=True,
+                )
+                raise error
+            await interaction.respond(  # pyright: ignore[reportUnknownMemberType]
+                f"Whoops! I don't have permission to do that\n`{error.args[0].split(':')[-1].strip()}`",
+                ephemeral=True,
+            )
+
+    discord.ui.View = PatchedView