diff --git a/sanic/app.py b/sanic/app.py index 0686f7ed..3a7e738c 100644 --- a/sanic/app.py +++ b/sanic/app.py @@ -68,6 +68,7 @@ from sanic.models.futures import ( FutureStatic, ) from sanic.models.handler_types import ListenerType, MiddlewareType +from sanic.models.handler_types import Sanic as SanicVar from sanic.request import Request from sanic.response import BaseHTTPResponse, HTTPResponse from sanic.router import Router @@ -184,7 +185,7 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta): ) self.is_running = False self.is_stopping = False - self.listeners: Dict[str, List[ListenerType]] = defaultdict(list) + self.listeners: Dict[str, List[ListenerType[Any]]] = defaultdict(list) self.named_request_middleware: Dict[str, Deque[MiddlewareType]] = {} self.named_response_middleware: Dict[str, Deque[MiddlewareType]] = {} self.reload_dirs: Set[Path] = set() @@ -196,7 +197,7 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta): self.sock = None self.strict_slashes = strict_slashes self.websocket_enabled = False - self.websocket_tasks: Set[Future] = set() + self.websocket_tasks: Set[Future[Any]] = set() # Register alternative method names self.go_fast = self.run @@ -232,7 +233,10 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta): # Registration # -------------------------------------------------------------------- # - def add_task(self, task) -> None: + def add_task( + self, + task: Union[Future[Any], Coroutine[Any, Any, Any], Awaitable[Any]], + ) -> None: """ Schedule a task to run later, after the loop has started. Different from asyncio.ensure_future in that it does not @@ -255,7 +259,9 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta): self.signal(task_name)(partial(self.run_delayed_task, task=task)) self._delayed_tasks.append(task_name) - def register_listener(self, listener: Callable, event: str) -> Any: + def register_listener( + self, listener: ListenerType[SanicVar], event: str + ) -> ListenerType[SanicVar]: """ Register the listener for a given event. @@ -281,7 +287,9 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta): return listener - def register_middleware(self, middleware, attach_to: str = "request"): + def register_middleware( + self, middleware: MiddlewareType, attach_to: str = "request" + ) -> MiddlewareType: """ Register an application level middleware that will be attached to all the API URLs registered under this application. @@ -307,7 +315,7 @@ class Sanic(BaseSanic, metaclass=TouchUpMeta): def register_named_middleware( self, - middleware, + middleware: MiddlewareType, route_names: Iterable[str], attach_to: str = "request", ): diff --git a/sanic/base.py b/sanic/base.py index 5d1358d8..1489f545 100644 --- a/sanic/base.py +++ b/sanic/base.py @@ -23,7 +23,7 @@ class BaseSanic( ): __fake_slots__: Tuple[str, ...] - def __init__(self, name: str = None, *args, **kwargs) -> None: + def __init__(self, name: str = None, *args: Any, **kwargs: Any) -> None: class_name = self.__class__.__name__ if name is None: diff --git a/sanic/blueprints.py b/sanic/blueprints.py index 617ec606..e5e1d333 100644 --- a/sanic/blueprints.py +++ b/sanic/blueprints.py @@ -5,7 +5,16 @@ import asyncio from collections import defaultdict from copy import deepcopy from types import SimpleNamespace -from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Set, Union +from typing import ( + TYPE_CHECKING, + Any, + Dict, + Iterable, + List, + Optional, + Set, + Union, +) from sanic_routing.exceptions import NotFound # type: ignore from sanic_routing.route import Route # type: ignore @@ -142,7 +151,7 @@ class Blueprint(BaseSanic): def reset(self): self._apps: Set[Sanic] = set() self.exceptions: List[RouteHandler] = [] - self.listeners: Dict[str, List[ListenerType]] = {} + self.listeners: Dict[str, List[ListenerType[Any]]] = {} self.middlewares: List[MiddlewareType] = [] self.routes: List[Route] = [] self.statics: List[RouteHandler] = [] @@ -221,7 +230,7 @@ class Blueprint(BaseSanic): version: Optional[Union[int, str, float]] = None, strict_slashes: Optional[bool] = None, version_prefix: str = "/v", - ): + ) -> BlueprintGroup: """ Create a list of blueprints, optionally grouping them under a general URL prefix. diff --git a/sanic/mixins/listeners.py b/sanic/mixins/listeners.py index ebf9b131..39c969b8 100644 --- a/sanic/mixins/listeners.py +++ b/sanic/mixins/listeners.py @@ -3,7 +3,7 @@ from functools import partial from typing import List, Optional, Union from sanic.models.futures import FutureListener -from sanic.models.handler_types import ListenerType +from sanic.models.handler_types import ListenerType, Sanic class ListenerEvent(str, Enum): @@ -27,10 +27,10 @@ class ListenerMixin: def listener( self, - listener_or_event: Union[ListenerType, str], + listener_or_event: Union[ListenerType[Sanic], str], event_or_none: Optional[str] = None, apply: bool = True, - ): + ) -> ListenerType[Sanic]: """ Create a listener from a decorated function. @@ -62,20 +62,32 @@ class ListenerMixin: else: return partial(register_listener, event=listener_or_event) - def main_process_start(self, listener: ListenerType) -> ListenerType: + def main_process_start( + self, listener: ListenerType[Sanic] + ) -> ListenerType[Sanic]: return self.listener(listener, "main_process_start") - def main_process_stop(self, listener: ListenerType) -> ListenerType: + def main_process_stop( + self, listener: ListenerType[Sanic] + ) -> ListenerType[Sanic]: return self.listener(listener, "main_process_stop") - def before_server_start(self, listener: ListenerType) -> ListenerType: + def before_server_start( + self, listener: ListenerType[Sanic] + ) -> ListenerType[Sanic]: return self.listener(listener, "before_server_start") - def after_server_start(self, listener: ListenerType) -> ListenerType: + def after_server_start( + self, listener: ListenerType[Sanic] + ) -> ListenerType[Sanic]: return self.listener(listener, "after_server_start") - def before_server_stop(self, listener: ListenerType) -> ListenerType: + def before_server_stop( + self, listener: ListenerType[Sanic] + ) -> ListenerType[Sanic]: return self.listener(listener, "before_server_stop") - def after_server_stop(self, listener: ListenerType) -> ListenerType: + def after_server_stop( + self, listener: ListenerType[Sanic] + ) -> ListenerType[Sanic]: return self.listener(listener, "after_server_stop")