Adding the type guard decorator

 - in api - seemed the most appropriate place for it
 - (if nothing else, because it needed less cross module imports)
 - selected the option which didn't confuse pycharm - as that seems to be harder to fix
 - included a test suite hitting 100% of the module
 - decorator not currently applied to the examples - would seem beyond the scope of a single pull request
 - does including annotations in type guard break everything?
 - tests for the two different syntaxes for Union have been added

src/mewbot/api/v1.py

@@ -19,10 +19,12 @@
 
 from __future__ import annotations
 
+import types
 from collections.abc import AsyncIterable, Iterable
-from typing import Any
+from typing import Any, Callable, TypeVar, Union, get_args, get_origin, get_type_hints
 
 import abc
+import functools
 
 from mewbot.api.registry import ComponentRegistry
 from mewbot.core import (
@@ -492,6 +494,131 @@ def serialise(self) -> BehaviourConfigBlock:
         }
 
 
+TypingComponent = TypeVar("TypingComponent", bound=Union[Trigger, Condition])
+TypingEvent = TypeVar("TypingEvent", bound=InputEvent)
+
+
+def pre_filter_non_matching_events(
+    wrapped: Callable[[TypingComponent, TypingEvent], bool]
+) -> Callable[[TypingComponent, InputEvent], bool]:
+    """
+        Check an input event against the valid event types declared in the signature.
+
+        Introspects the function to determine the types of InputEvent should be passed to it.
+        Uses a decorator to check that the event being passed in is one of those.
+            - If it is, the function is run
+            - If it is not, False is returned
+
+        Type guard exists to provide methods for run time typing validation.
+
+    E.g. a decorator which checks that an InputEvent being passed to a function is of appropriate
+         type.
+
+    The intent of the tools here is to provide a more elegant alternative to constructs like
+
+    .. code-block:: python
+
+        def matches(self, event: InputEvent) -> bool:
+            if not isinstance(event, DiscordMessageCreationEvent):
+                return False
+            [Function body]
+
+    which could be found in a Trigger.
+
+    Instead, using one of the decorators provided in this module, you might write
+
+
+    .. code-block:: python
+
+        @pre_filter_non_matching_events
+        def matches(self, event: InputEvent) -> bool:
+            [Function body]
+
+    Or, as a more full example
+
+    .. code-block:: python
+
+        # Demo Code
+
+        from mewbot.io.discord import DiscordInputEvent, DiscordUserJoinInputEvent
+        from mewbot.io.http import IncomingWebhookEvent
+
+        class A(Trigger):
+
+            @staticmethod
+            def consumes_inputs() -> set[type[InputEvent]]:
+                return {DiscordInputEvent, IncomingWebhookEvent}
+
+            @pre_filter_non_matching_events
+            def matches(self, event: DiscordInputEvent | IncomingWebhookEvent) -> bool:
+                return True
+
+        print(A.matches) # <function A.matches at 0x7f6e703f4a40>
+        print(A.matches.__doc__) # <function A.matches at 0x7f6e703f4a40>
+        print(A().matches(InputEvent())) # False
+        print(A().matches(DiscordInputEvent())) # True
+        print(A().matches(DiscordUserJoinInputEvent("[some text]"))) # True
+
+        :param wrapped:
+        :return:
+    """
+    func_types = get_type_hints(wrapped)
+    if "event" not in func_types:
+        raise TypeError("Received function without 'event' parameter")
+
+    # Flatten the type signature down to the unique InputEvent subclasses.
+    event_types: tuple[type[TypingEvent]] = flatten_types(func_types["event"])
+
+    bad_types = [
+        t for t in event_types if not (isinstance(t, type) and issubclass(t, InputEvent))
+    ]
+    if bad_types:
+        raise TypeError(
+            (
+                f"{wrapped.__qualname__}: "
+                "Can not add filter for non-event type(s): "
+                f"{' '.join([str(x) for x in bad_types])}"
+            )
+        )
+
+    @functools.wraps(wrapped)
+    def match_with_type_check(self: TypingComponent, event: InputEvent) -> bool:
+        # noinspection PyTypeHints
+        if not isinstance(event, event_types):
+            return False
+
+        return wrapped(self, event)
+
+    return match_with_type_check
+
+
+def flatten_types(event_types: type[TypingEvent]) -> tuple[type[TypingEvent]]:
+    """
+    Flattens a possible union of InputEvent types into a tuple of types.
+
+    This is a helper method for pre_filter_non_matching_events.
+
+    The types in the tuple are expected (but not guaranteed) to be InputEvent subtypes.
+    This tuple can be safely used with isinstance() on all supported versions of Python.
+    It can also be iterated to confirm all the types are actually InputEvents.
+    """
+
+    events: tuple[type[TypingEvent]]
+
+    if isinstance(event_types, type):
+        events = (event_types,)
+    elif get_origin(event_types) is Union:
+        events = get_args(event_types)
+    elif isinstance(event_types, getattr(types, "UnionType")):
+        events = get_args(event_types)
+    else:  # pragma: no cover
+        raise ValueError(
+            "Got weird type from type hinting: " + event_types
+        )  # pragma: no cover
+
+    return events
+
+
 __all__ = [
     "IOConfig",
     "Input",
@@ -504,4 +631,5 @@ def serialise(self) -> BehaviourConfigBlock:
     "OutputEvent",
     "InputQueue",
     "OutputQueue",
+    "pre_filter_non_matching_events",
 ]

tests/api/type_guard/test_type_guard_checks_no_future_annotations.py

@@ -0,0 +1,157 @@
+# SPDX-FileCopyrightText: 2023 Mewbot Developers <[email protected]>
+#
+# SPDX-License-Identifier: BSD-2-Clause
+
+"""
+Tests pre_filter_non_matching_events using a dummy Trigger.
+"""
+
+# pylint: disable=duplicate-code
+# this is testing for identical behavior in two very similar cases
+# duplication is inevitable
+
+from typing import Union
+
+import pytest
+
+from mewbot.api.v1 import InputEvent, Trigger, pre_filter_non_matching_events
+from mewbot.io.http import IncomingWebhookEvent
+from mewbot.io.socket import SocketInputEvent
+
+TYPE_ERROR_MESSAGE = r".*.matches: Can not add filter for non-event type\(s\): .*"
+
+
+class SimpleTestTrigger(Trigger):
+    """Testing Trigger - Single Event sub-type."""
+
+    @staticmethod
+    def consumes_inputs() -> set[type[InputEvent]]:
+        """
+        Consume a specific type of Events.
+
+        :return:
+        """
+        return {SocketInputEvent}
+
+    @pre_filter_non_matching_events
+    def matches(self, event: SocketInputEvent) -> bool:
+        """
+        If the event is of the right type, matches.
+
+        :param event:
+        :return:
+        """
+        return True
+
+
+class UnionTestTrigger(Trigger):
+    """
+    Dummy trigger for testing.
+    """
+
+    @staticmethod
+    def consumes_inputs() -> set[type[InputEvent]]:
+        """
+        Consumes two radically different events.
+
+        :return:
+        """
+        return {SocketInputEvent, IncomingWebhookEvent}
+
+    @pre_filter_non_matching_events
+    def matches(self, event: Union[SocketInputEvent, IncomingWebhookEvent]) -> bool:
+        """
+        If the event is of the right type, matches.
+
+        :param event:
+        :return:
+        """
+        return True
+
+
+class PEP604TestTrigger(Trigger):
+    """
+    Dummy trigger for testing.
+    """
+
+    @staticmethod
+    def consumes_inputs() -> set[type[InputEvent]]:
+        """
+        Consumes two radically different events.
+
+        :return:
+        """
+        return {SocketInputEvent, IncomingWebhookEvent}
+
+    @pre_filter_non_matching_events
+    def matches(self, event: SocketInputEvent | IncomingWebhookEvent) -> bool:
+        """
+        If the event is of the right type, matches.
+
+        :param event:
+        :return:
+        """
+        return True
+
+
+class TestCheckInputEventAgainstSigTypesAnnotations:
+    """
+    Check the type guard annotation pre_filter_non_matching_events preforms as expected.
+    """
+
+    @pytest.mark.parametrize(
+        "clazz", [SimpleTestTrigger, UnionTestTrigger, PEP604TestTrigger]
+    )
+    def test_match_on_base_input_event_fails_annotations(self, clazz: type[Trigger]) -> None:
+        """
+        Calls the decorated function with the base class for InputEvent.
+
+        Should always fail.
+        :return:
+        """
+        assert clazz().matches(InputEvent()) is False
+
+    @pytest.mark.parametrize(
+        "clazz", [SimpleTestTrigger, UnionTestTrigger, PEP604TestTrigger]
+    )
+    def test_match_two_of_two_valid_event_types_annotations(
+        self, clazz: type[Trigger]
+    ) -> None:
+        """
+        Tests that a valid event type does match - for IncomingWebhookEvent.
+
+        :return:
+        """
+        test_socket_event = SocketInputEvent(data=b"some bytes")
+
+        assert clazz().matches(test_socket_event) is True
+
+    @pytest.mark.parametrize("clazz", [UnionTestTrigger, PEP604TestTrigger])
+    def test_match_one_of_two_valid_event_types_annotations(
+        self, clazz: type[Trigger]
+    ) -> None:
+        """
+        Tests that a valid event type does match - for IncomingWebhookEvent.
+
+        :return:
+        """
+        test_webhook_event = IncomingWebhookEvent(text="test_text")
+
+        assert clazz().matches(test_webhook_event) is True
+
+    @pytest.mark.parametrize(
+        "clazz", [SimpleTestTrigger, UnionTestTrigger, PEP604TestTrigger]
+    )
+    def test_complete_the_wrong_class_in_as_an_event_annotations(
+        self, clazz: type[Trigger]
+    ) -> None:
+        """
+        Tests that the decorator raises a type error when we pass something complete wrong in.
+
+        In place of an event
+        :return:
+        """
+        try:
+            clazz().matches(None)  # type: ignore
+        except TypeError:
+            pass