style: add some type hints (#2279)

* style: add some type hints

* fix: *args is a tuple, but overridden as a list

* fix: if touch this, it will be a maybe breaking change

* fix: remove unused import

* style(app): more correct type

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",
     ):

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:

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.

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")