Fixing typing for ListenerMixin.listener (sanic-org#2376)

Co-authored-by: Adam Hopkins <adam@amhopkins.com>

sanic/mixins/listeners.py

@@ -1,8 +1,9 @@
 from enum import Enum, auto
 from functools import partial
-from typing import List, Optional, Union
+from typing import Callable, List, Optional, Union, overload
 
 from sanic.base.meta import SanicMeta
+from sanic.exceptions import InvalidUsage
 from sanic.models.futures import FutureListener
 from sanic.models.handler_types import ListenerType, Sanic
 
@@ -28,12 +29,33 @@ def __init__(self, *args, **kwargs) -> None:
     def _apply_listener(self, listener: FutureListener):
         raise NotImplementedError  # noqa
 
+    @overload
+    def listener(
+        self,
+        listener_or_event: ListenerType[Sanic],
+        event_or_none: str,
+        apply: bool = ...,
+    ) -> ListenerType[Sanic]:
+        ...
+
+    @overload
+    def listener(
+        self,
+        listener_or_event: str,
+        event_or_none: None = ...,
+        apply: bool = ...,
+    ) -> Callable[[ListenerType[Sanic]], ListenerType[Sanic]]:
+        ...
+
     def listener(
         self,
         listener_or_event: Union[ListenerType[Sanic], str],
         event_or_none: Optional[str] = None,
         apply: bool = True,
-    ) -> ListenerType[Sanic]:
+    ) -> Union[
+        ListenerType[Sanic],
+        Callable[[ListenerType[Sanic]], ListenerType[Sanic]],
+    ]:
         """
         Create a listener from a decorated function.
 
@@ -51,7 +73,9 @@ async def before_server_start(app, loop):
         :param event: event to listen to
         """
 
-        def register_listener(listener, event):
+        def register_listener(
+            listener: ListenerType[Sanic], event: str
+        ) -> ListenerType[Sanic]:
             nonlocal apply
 
             future_listener = FutureListener(listener, event)
@@ -61,6 +85,10 @@ def register_listener(listener, event):
             return listener
 
         if callable(listener_or_event):
+            if event_or_none is None:
+                raise InvalidUsage(
+                    "Invalid event registration: Missing event name."
+                )
             return register_listener(listener_or_event, event_or_none)
         else:
             return partial(register_listener, event=listener_or_event)

sanic/models/handler_types.py

@@ -1,11 +1,13 @@
 from asyncio.events import AbstractEventLoop
 from typing import Any, Callable, Coroutine, Optional, TypeVar, Union
 
+import sanic
+
 from sanic.request import Request
 from sanic.response import BaseHTTPResponse, HTTPResponse
 
 
-Sanic = TypeVar("Sanic")
+Sanic = TypeVar("Sanic", bound="sanic.Sanic")
 
 MiddlewareResponse = Union[
     Optional[HTTPResponse], Coroutine[Any, Any, Optional[HTTPResponse]]

tests/test_signal_handlers.py

@@ -10,6 +10,7 @@
 from sanic_testing.testing import HOST, PORT
 
 from sanic.compat import ctrlc_workaround_for_windows
+from sanic.exceptions import InvalidUsage
 from sanic.response import HTTPResponse
 
 
@@ -108,3 +109,17 @@ async def atest(stop_first):
     assert res == "OK"
     res = loop.run_until_complete(atest(True))
     assert res == "OK"
+
+
+@pytest.mark.skipif(os.name == "nt", reason="May hang CI on py38/windows")
+def test_signals_with_invalid_invocation(app):
+    """Test if sanic register fails with invalid invocation"""
+
+    @app.route("/hello")
+    async def hello_route(request):
+        return HTTPResponse()
+
+    with pytest.raises(
+        InvalidUsage, match="Invalid event registration: Missing event name"
+    ):
+        app.listener(stop)