Fixing typing for ListenerMixin.listener (#2376)

Co-authored-by: Adam Hopkins <adam@amhopkins.com>
This commit is contained in:
André Ericson 2022-03-23 14:34:33 +01:00 committed by GitHub
parent 6e0a6871b5
commit 32962d1e1c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 49 additions and 4 deletions

View File

@ -1,8 +1,9 @@
from enum import Enum, auto from enum import Enum, auto
from functools import partial 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.base.meta import SanicMeta
from sanic.exceptions import InvalidUsage
from sanic.models.futures import FutureListener from sanic.models.futures import FutureListener
from sanic.models.handler_types import ListenerType, Sanic from sanic.models.handler_types import ListenerType, Sanic
@ -28,12 +29,33 @@ class ListenerMixin(metaclass=SanicMeta):
def _apply_listener(self, listener: FutureListener): def _apply_listener(self, listener: FutureListener):
raise NotImplementedError # noqa 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( def listener(
self, self,
listener_or_event: Union[ListenerType[Sanic], str], listener_or_event: Union[ListenerType[Sanic], str],
event_or_none: Optional[str] = None, event_or_none: Optional[str] = None,
apply: bool = True, apply: bool = True,
) -> ListenerType[Sanic]: ) -> Union[
ListenerType[Sanic],
Callable[[ListenerType[Sanic]], ListenerType[Sanic]],
]:
""" """
Create a listener from a decorated function. Create a listener from a decorated function.
@ -51,7 +73,9 @@ class ListenerMixin(metaclass=SanicMeta):
:param event: event to listen to :param event: event to listen to
""" """
def register_listener(listener, event): def register_listener(
listener: ListenerType[Sanic], event: str
) -> ListenerType[Sanic]:
nonlocal apply nonlocal apply
future_listener = FutureListener(listener, event) future_listener = FutureListener(listener, event)
@ -61,6 +85,10 @@ class ListenerMixin(metaclass=SanicMeta):
return listener return listener
if callable(listener_or_event): 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) return register_listener(listener_or_event, event_or_none)
else: else:
return partial(register_listener, event=listener_or_event) return partial(register_listener, event=listener_or_event)

View File

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

View File

@ -10,6 +10,7 @@ import pytest
from sanic_testing.testing import HOST, PORT from sanic_testing.testing import HOST, PORT
from sanic.compat import ctrlc_workaround_for_windows from sanic.compat import ctrlc_workaround_for_windows
from sanic.exceptions import InvalidUsage
from sanic.response import HTTPResponse from sanic.response import HTTPResponse
@ -108,3 +109,17 @@ def test_windows_workaround():
assert res == "OK" assert res == "OK"
res = loop.run_until_complete(atest(True)) res = loop.run_until_complete(atest(True))
assert res == "OK" 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)