Base guard implementation

Author: Eviee Py

twitchio/ext/commands/bot.py

@@ -138,4 +138,4 @@ async def before_invoke(self, ctx: Context) -> None: ...
 
     async def after_invoke(self, ctx: Context) -> None: ...
 
-    async def check(self, ctx: Context) -> None: ...
+    async def guard(self, ctx: Context) -> None: ...

twitchio/ext/commands/core.py

@@ -36,14 +36,7 @@
 from .types_ import CommandOptions, Component_T
 
 
-__all__ = (
-    "CommandErrorPayload",
-    "Command",
-    "Mixin",
-    "Group",
-    "command",
-    "group",
-)
+__all__ = ("CommandErrorPayload", "Command", "Mixin", "Group", "command", "group", "is_broadcaster")
 
 
 if TYPE_CHECKING:
@@ -55,6 +48,7 @@
 
 
 Coro: TypeAlias = Coroutine[Any, Any, None]
+CoroC: TypeAlias = Coroutine[Any, Any, bool]
 
 
 class CommandErrorPayload:
@@ -76,6 +70,7 @@ def __init__(
         self._name: str = name
         self._callback = callback
         self._aliases: list[str] = kwargs.get("aliases", [])
+        self._guards: list[Callable[..., bool] | Callable[..., CoroC]] = getattr(self._callback, "__command_guards__", [])
 
         self._injected: Component_T | None = None
         self._error: Callable[[Component_T, CommandErrorPayload], Coro] | Callable[[CommandErrorPayload], Coro] | None = None
@@ -116,6 +111,13 @@ def extras(self) -> dict[Any, Any]:
     def has_error(self) -> bool:
         return self._error is not None
 
+    @property
+    def guards(self) -> ...: ...
+
+    @property
+    def callback(self) -> Callable[Concatenate[Component_T, Context, P], Coro] | Callable[Concatenate[Context, P], Coro]:
+        return self._callback
+
     async def _do_conversion(self, context: Context, param: inspect.Parameter, *, annotation: Any, raw: str | None) -> Any:
         name: str = param.name
 
@@ -181,16 +183,15 @@ async def _parse_arguments(self, context: Context) -> ...:
             if param.kind == param.KEYWORD_ONLY:
                 raw = context._view.read_rest()
 
-                if not raw:
-                    if param.default == param.empty:
-                        raise MissingRequiredArgument(param=param)
+                if raw:
+                    result = await self._do_conversion(context, param=param, raw=raw, annotation=param.annotation)
+                    kwargs[param.name] = result
+                    break
 
-                    kwargs[param.name] = param.default
-                    continue
+                if param.default == param.empty:
+                    raise MissingRequiredArgument(param=param)
 
-                result = await self._do_conversion(context, param=param, raw=raw, annotation=param.annotation)
-                kwargs[param.name] = result
-                break
+                kwargs[param.name] = param.default
 
             elif param.kind == param.VAR_POSITIONAL:
                 packed: list[Any] = []
@@ -211,23 +212,25 @@ async def _parse_arguments(self, context: Context) -> ...:
                 raw = context._view.get_quoted_word()
                 context._view.skip_ws()
 
-                if not raw:
-                    if param.default == param.empty:
-                        raise MissingRequiredArgument(param=param)
-
-                    args.append(param.default)
+                if raw:
+                    result = await self._do_conversion(context, param=param, raw=raw, annotation=param.annotation)
+                    args.append(result)
                     continue
 
-                result = await self._do_conversion(context, param=param, raw=raw, annotation=param.annotation)
-                args.append(result)
+                if param.default == param.empty:
+                    raise MissingRequiredArgument(param=param)
+
+                args.append(param.default)
 
         return args, kwargs
 
-    async def _do_checks(self, context: Context) -> ...:
-        # Bot
-        # Component
-        # Command
-        ...
+    async def _run_guards(self, context: Context) -> ...:
+        # TODO ...
+        for guard in self._guards:
+            result = guard(context)
+
+            if not result:
+                raise CheckFailure
 
     async def _invoke(self, context: Context) -> None:
         try:
@@ -243,6 +246,8 @@ async def _invoke(self, context: Context) -> None:
         args: list[Any] = [context, *args]
         args.insert(0, self._injected) if self._injected else None
 
+        await self._run_guards(context)
+
         callback = self._callback(*args, **kwargs)  # type: ignore
 
         try:
@@ -284,6 +289,16 @@ def error(
         self._error = func
         return func
 
+    def add_guard(self) -> None: ...
+
+    def remove_guard(
+        self,
+    ) -> None: ...
+
+    def before_invoke(self) -> None: ...
+
+    def after_invoke(self) -> None: ...
+
 
 class Mixin(Generic[Component_T]):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
@@ -427,3 +442,26 @@ def wrapper(
             return new
 
         return wrapper
+
+
+def guard(predicate: Callable[..., bool] | Callable[..., CoroC]) -> Any:
+    def wrapper(func: Any) -> Any:
+        if isinstance(func, Command):
+            func._guards.append(predicate)
+
+        else:
+            try:
+                func.__command_guards__.append(predicate)
+            except AttributeError:
+                func.__command_guards__ = [predicate]
+
+        return func  # type: ignore
+
+    return wrapper
+
+
+def is_broadcaster() -> Any:
+    def predicate(context: Context) -> bool:
+        return context.chatter.id == context.broadcaster.id
+
+    return guard(predicate)