fix: mypy issues

setup.cfg

@@ -12,6 +12,12 @@ files =
     uvicorn/supervisors/basereload.py,
     uvicorn/importer.py,
     uvicorn/protocols/utils.py,
+    uvicorn/loops,
+    uvicorn/main.py,
+    uvicorn/workers.py,
+    uvicorn/protocols/http/auto.py,
+    uvicorn/protocols/websockets/auto.py,
+    uvicorn/supervisors/__init__.py,
     uvicorn/middleware/debug.py,
     uvicorn/supervisors/watchgodreload.py,
     uvicorn/logging.py,

uvicorn/loops/asyncio.py

@@ -4,7 +4,8 @@
 import sys
 
 
-def asyncio_setup():  # pragma: no cover
+def asyncio_setup() -> None:  # pragma: no cover
+    loop: asyncio.AbstractEventLoop
     if (
         sys.version_info.major >= 3
         and sys.version_info.minor >= 8

uvicorn/loops/auto.py

@@ -1,4 +1,4 @@
-def auto_loop_setup():
+def auto_loop_setup() -> None:
     try:
         import uvloop  # noqa
     except ImportError:  # pragma: no cover

uvicorn/loops/uvloop.py

@@ -3,5 +3,5 @@
 import uvloop
 
 
-def uvloop_setup():
+def uvloop_setup() -> None:
     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

uvicorn/main.py

@@ -5,6 +5,7 @@
 import typing
 
 import click
+from asgiref.typing import ASGIApplication
 
 import uvicorn
 from uvicorn.config import (
@@ -21,17 +22,17 @@
 from uvicorn.server import Server, ServerState  # noqa: F401  # Used to be defined here.
 from uvicorn.supervisors import ChangeReload, Multiprocess
 
-LEVEL_CHOICES = click.Choice(LOG_LEVELS.keys())
-HTTP_CHOICES = click.Choice(HTTP_PROTOCOLS.keys())
-WS_CHOICES = click.Choice(WS_PROTOCOLS.keys())
-LIFESPAN_CHOICES = click.Choice(LIFESPAN.keys())
+LEVEL_CHOICES = click.Choice(list(LOG_LEVELS.keys()))
+HTTP_CHOICES = click.Choice(list(HTTP_PROTOCOLS.keys()))
+WS_CHOICES = click.Choice(list(WS_PROTOCOLS.keys()))
+LIFESPAN_CHOICES = click.Choice(list(LIFESPAN.keys()))
 LOOP_CHOICES = click.Choice([key for key in LOOP_SETUPS.keys() if key != "none"])
 INTERFACE_CHOICES = click.Choice(INTERFACES)
 
 logger = logging.getLogger("uvicorn.error")
 
 
-def print_version(ctx, param, value):
+def print_version(ctx: click.Context, param: click.Parameter, value: bool) -> None:
     if not value or ctx.resilient_parsing:
         return
     click.echo(
@@ -288,7 +289,7 @@ def print_version(ctx, param, value):
     show_default=True,
 )
 def main(
-    app,
+    app: str,
     host: str,
     port: int,
     uds: str,
@@ -326,11 +327,10 @@ def main(
     use_colors: bool,
     app_dir: str,
     factory: bool,
-):
+) -> None:
     sys.path.insert(0, app_dir)
 
     kwargs = {
-        "app": app,
         "host": host,
         "port": port,
         "uds": uds,
@@ -368,10 +368,10 @@ def main(
         "use_colors": use_colors,
         "factory": factory,
     }
-    run(**kwargs)
+    run(app, **kwargs)
 
 
-def run(app, **kwargs):
+def run(app: typing.Union[ASGIApplication, str], **kwargs: typing.Any) -> None:
     config = Config(app, **kwargs)
     server = Server(config=config)
 
@@ -385,12 +385,10 @@ def run(app, **kwargs):
 
     if config.should_reload:
         sock = config.bind_socket()
-        supervisor = ChangeReload(config, target=server.run, sockets=[sock])
-        supervisor.run()
+        ChangeReload(config, target=server.run, sockets=[sock]).run()
     elif config.workers > 1:
         sock = config.bind_socket()
-        supervisor = Multiprocess(config, target=server.run, sockets=[sock])
-        supervisor.run()
+        Multiprocess(config, target=server.run, sockets=[sock]).run()
     else:
         server.run()
 

uvicorn/protocols/http/auto.py

@@ -1,3 +1,7 @@
+import asyncio
+from typing import Type
+
+AutoHTTPProtocol: Type[asyncio.Protocol]
 try:
     import httptools  # noqa
 except ImportError:  # pragma: no cover

uvicorn/protocols/websockets/auto.py

@@ -1,3 +1,7 @@
+import asyncio
+import typing
+
+AutoWebSocketsProtocol: typing.Optional[typing.Type[asyncio.Protocol]]
 try:
     import websockets  # noqa
 except ImportError:  # pragma: no cover

uvicorn/supervisors/__init__.py

@@ -1,8 +1,14 @@
+import typing
+
+from uvicorn.supervisors.basereload import BaseReload
 from uvicorn.supervisors.multiprocess import Multiprocess
 
-try:
-    from uvicorn.supervisors.watchgodreload import WatchGodReload as ChangeReload
-except ImportError:  # pragma: no cover
-    from uvicorn.supervisors.statreload import StatReload as ChangeReload
+if typing.TYPE_CHECKING:
+    ChangeReload: typing.Type[BaseReload]  # pragma: no cover
+else:
+    try:
+        from uvicorn.supervisors.watchgodreload import WatchGodReload as ChangeReload
+    except ImportError:  # pragma: no cover
+        from uvicorn.supervisors.statreload import StatReload as ChangeReload
 
 __all__ = ["Multiprocess", "ChangeReload"]

uvicorn/workers.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import signal
+from typing import Any
 
 from gunicorn.workers.base import Worker
 
@@ -16,7 +17,7 @@ class UvicornWorker(Worker):
 
     CONFIG_KWARGS = {"loop": "auto", "http": "auto"}
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         super(UvicornWorker, self).__init__(*args, **kwargs)
 
         logger = logging.getLogger("uvicorn.error")
@@ -58,24 +59,24 @@ def __init__(self, *args, **kwargs):
 
         self.config = Config(**config_kwargs)
 
-    def init_process(self):
+    def init_process(self) -> None:
         self.config.setup_event_loop()
         super(UvicornWorker, self).init_process()
 
-    def init_signals(self):
+    def init_signals(self) -> None:
         # Reset signals so Gunicorn doesn't swallow subprocess return codes
         # other signals are set up by Server.install_signal_handlers()
         # See: https://github.com/encode/uvicorn/issues/894
         for s in self.SIGNALS:
             signal.signal(s, signal.SIG_DFL)
 
-    def run(self):
+    def run(self) -> None:
         self.config.app = self.wsgi
         server = Server(config=self.config)
         loop = asyncio.get_event_loop()
         loop.run_until_complete(server.serve(sockets=self.sockets))
 
-    async def callback_notify(self):
+    async def callback_notify(self) -> None:
         self.notify()